From 878c70993cb7307f9ccab587b286553604b9db8d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 7 Sep 2026 17:06:34 -0400 Subject: [PATCH 1/7] Export CPS parent pointers as person-level parent ids Resolve PEPAR1/PEPAR2 against (PH_SEQ, A_LINENO) to the co-resident member's person_id and export them as parent_1_id / parent_2_id, the relation behind own_children_in_household rather than only its cardinality (microcosm#884, consumed by policyengine-us#9404). - eligibility_inputs: one shared pointer normalization behind both the count and the ids, int64 storage (a clone-remapped person_id runs past 2**53), and gate invariants asserting co-residence, no self-parenting, and that own_children_in_household equals the number of people whose parent ids name that person. - acs_inputs / operator_boundary: the ACS spine declares the pointers unknown (0) natively. Its synthesized PEPAR values are a reference-person link, not the ASEC measured pointer, and an ACS-minted person_id is renumbered by the assembly offset after that mapping runs. - multispine_pool: the parent ids are native-complete on both arms, so they never enter the QRF transfer plan. They are not PolicyEngine-US variables, so _target_encoding would treat an id as continuous and hand an ACS child a fractional interpolation between two ASEC person ids. The transfer surface is unchanged at 118 targets. - puf_support: the support clone now shifts person-referencing columns with person_id, preserving the 0 sentinel; without it a clone's child pointed at the other arm's parent. - source_stages.json + regenerated sources.yaml, with the frozen generation-0 digest re-pinned. Co-Authored-By: Claude Fable 5.1 --- .../src/microcosm/build/us/source_stages.json | 6 +- .../src/microcosm/build/us/spec/sources.yaml | 16 +- .../microcosm/build/us_runtime/acs_inputs.py | 40 ++- .../build/us_runtime/eligibility_inputs.py | 296 +++++++++++++++++- .../build/us_runtime/multispine_pool.py | 10 + .../build/us_runtime/operator_boundary.py | 12 + .../microcosm/build/us_runtime/puf_support.py | 69 ++++ tools/generate_us_bundle_from_constants.py | 2 +- 8 files changed, 429 insertions(+), 22 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/us/source_stages.json b/packages/microcosm-build/src/microcosm/build/us/source_stages.json index 8b5d06c0c8..72b56c12ab 100644 --- a/packages/microcosm-build/src/microcosm/build/us/source_stages.json +++ b/packages/microcosm-build/src/microcosm/build/us/source_stages.json @@ -2227,13 +2227,17 @@ "is_blind", "is_full_time_college_student", "own_children_in_household", + "parent_1_id", + "parent_2_id", "veterans_benefits" ], "nonnegative_outputs": [ "own_children_in_household", + "parent_1_id", + "parent_2_id", "veterans_benefits" ], - "notes": "Direct mappings of measured ASEC demographic variables reproducing the retired enhanced-CPS pipeline's constructions (each cited to the retired pipeline's source files in the stage module): is_disabled <- any PEDIS* difficulty item == 1, aligned with reported SSI (under-65 SSI_VAL > 0 implies disabled \u2014 the retired align_reported_ssi_disability rule from datasets/cps/takeup.py, applied to is_disabled in cps.py); is_blind <- PEDISEYE == 1; is_full_time_college_student <- A_HSCOL == 2 & A_FTPT == 1 (the retired primary branch; census_cps.py loads A_FTPT, so the published surface ran the strict form \u2014 the A_HSCOL-only fallback for loads without A_FTPT is not reproduced); own_children_in_household <- count of household members whose PEPAR1/PEPAR2 parent pointers resolve to the person within PH_SEQ; veterans_benefits <- VET_VAL. Nothing is imputed. Without this stage every input takes its engine default (False/0), erasing SNAP disability, student, parent/child, and veteran exemption channels (microcosm #244/#248). is_veteran is formula-owned in PolicyEngine-US (derived from veterans_benefits) and is deliberately not exported (microcosm #24)." + "notes": "Direct mappings of measured ASEC demographic variables reproducing the retired enhanced-CPS pipeline's constructions (each cited to the retired pipeline's source files in the stage module): is_disabled <- any PEDIS* difficulty item == 1, aligned with reported SSI (under-65 SSI_VAL > 0 implies disabled \u2014 the retired align_reported_ssi_disability rule from datasets/cps/takeup.py, applied to is_disabled in cps.py); is_blind <- PEDISEYE == 1; is_full_time_college_student <- A_HSCOL == 2 & A_FTPT == 1 (the retired primary branch; census_cps.py loads A_FTPT, so the published surface ran the strict form \u2014 the A_HSCOL-only fallback for loads without A_FTPT is not reproduced); own_children_in_household <- count of household members whose PEPAR1/PEPAR2 parent pointers resolve to the person within PH_SEQ; parent_1_id/parent_2_id <- the person_id of the co-resident member the person's PEPAR1/PEPAR2 pointer names within PH_SEQ, 0 when the pointer is absent or resolves to no co-resident line (microcosm#884; the relation behind the count, consumed by policyengine-us#9404); veterans_benefits <- VET_VAL. Nothing is imputed. Without this stage every input takes its engine default (False/0), erasing SNAP disability, student, parent/child, and veteran exemption channels (microcosm #244/#248). is_veteran is formula-owned in PolicyEngine-US (derived from veterans_benefits) and is deliberately not exported (microcosm #24)." }, { "stage": "pregnancy", diff --git a/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml b/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml index b952fd60e4..156a37cfe6 100644 --- a/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml +++ b/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml @@ -99,7 +99,7 @@ sources: stage_asset: id: source_stages path: microcosm.build.us/source_stages.json - sha256: 9f4f983091fb18dea8f4524cc100be20f9fca2b006435461592f8edf53c9a5d8 + sha256: e75217c0db3075bf2664492167d31ce7ee82c2c5f6a70e3b67e8bb6f4ddd7ecf stage_manifest: version: 1 country: us @@ -2222,9 +2222,13 @@ stages: - is_blind - is_full_time_college_student - own_children_in_household + - parent_1_id + - parent_2_id - veterans_benefits nonnegative_outputs: - own_children_in_household + - parent_1_id + - parent_2_id - veterans_benefits notes: 'Direct mappings of measured ASEC demographic variables reproducing the retired enhanced-CPS pipeline''s constructions (each cited to the retired pipeline''s source files in the stage module): @@ -2234,10 +2238,12 @@ stages: A_FTPT == 1 (the retired primary branch; census_cps.py loads A_FTPT, so the published surface ran the strict form — the A_HSCOL-only fallback for loads without A_FTPT is not reproduced); own_children_in_household <- count of household members whose PEPAR1/PEPAR2 parent pointers resolve to the person within PH_SEQ; - veterans_benefits <- VET_VAL. Nothing is imputed. Without this stage every input takes its engine - default (False/0), erasing SNAP disability, student, parent/child, and veteran exemption channels - (microcosm #244/#248). is_veteran is formula-owned in PolicyEngine-US (derived from veterans_benefits) - and is deliberately not exported (microcosm #24).' + parent_1_id/parent_2_id <- the person_id of the co-resident member the person''s PEPAR1/PEPAR2 pointer + names within PH_SEQ, 0 when the pointer is absent or resolves to no co-resident line (microcosm#884; + the relation behind the count, consumed by policyengine-us#9404); veterans_benefits <- VET_VAL. Nothing + is imputed. Without this stage every input takes its engine default (False/0), erasing SNAP disability, + student, parent/child, and veteran exemption channels (microcosm #244/#248). is_veteran is formula-owned + in PolicyEngine-US (derived from veterans_benefits) and is deliberately not exported (microcosm #24).' - stage: pregnancy survey: Census CPS ASEC + CDC natality-derived national pregnancy rate source: https://www.cdc.gov/nchs/nvss/births.htm diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/acs_inputs.py b/packages/microcosm-build/src/microcosm/build/us_runtime/acs_inputs.py index d6a03bb092..1806ca1928 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/acs_inputs.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/acs_inputs.py @@ -20,7 +20,11 @@ from microcosm.frame import US_SCHEMA, Frame -__all__ = ["AcsNativeInputResult", "map_acs_native_inputs"] +__all__ = [ + "ACS_UNRESOLVED_PARENT_ID_MAPPINGS", + "AcsNativeInputResult", + "map_acs_native_inputs", +] _INFLATION_FACTOR_DENOMINATOR = 1_000_000.0 @@ -37,6 +41,28 @@ 4: "RENTER", # occupied without payment: non-owner housing tenure } +#: The ACS spine's value for the ASEC parent pointers, and why. +#: +#: ``acs_pums._with_structural_columns`` synthesizes PEPAR1/PEPAR2 from +#: RELSHIPP: for a biological/adopted/step child of the reference person +#: (25/26/27) PEPAR1 is always the reference person's line and PEPAR2 always +#: the spouse's, and every other record — grandchild, foster child, child of a +#: non-reference adult — gets nothing. That is a reference-person link, not the +#: ASEC's measured parent pointer, so exporting it under the same name would +#: conflate two constructions; and an ACS-minted ``person_id`` is renumbered by +#: the assembly offset after this mapping runs, so a resolved id would be stale +#: before it reached the pool. The ACS spine therefore reports the pointer as +#: unknown (0), which is exactly the value PolicyEngine-US reads as "fall back +#: to the count-based proxy" (microcosm#884, policyengine-us#9404). +_ACS_UNRESOLVED_PARENT_POINTER_TRANSFORMATION = ( + "0 (unknown): the ACS pointer is a synthesized reference-person link, " + "not the ASEC measured parent pointer" +) +ACS_UNRESOLVED_PARENT_ID_MAPPINGS: tuple[tuple[str, str], ...] = ( + ("parent_1_id", "PEPAR1"), + ("parent_2_id", "PEPAR2"), +) + _FORMULA_OWNED_AGGREGATES = frozenset( { "employment_income", @@ -151,6 +177,18 @@ def map_acs_native_inputs(frame: Frame) -> AcsNativeInputResult: register=native, ) + for output, pointer_column in ACS_UNRESOLVED_PARENT_ID_MAPPINGS: + if pointer_column in person: + _add_native( + person, + output, + np.zeros(len(person), dtype=np.int64), + entity="person", + source_columns=(pointer_column,), + transformation=_ACS_UNRESOLVED_PARENT_POINTER_TRANSFORMATION, + register=native, + ) + _map_tenure(person, household, spm_unit, register=native) _map_housing_amounts(person, household, register=native) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py index dc7e698264..a3ccc5a52f 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py @@ -37,6 +37,19 @@ - ``own_children_in_household`` ← count of household members whose parent pointers (PEPAR1/PEPAR2, parent line numbers within PH_SEQ) resolve to the person. +- ``parent_1_id`` / ``parent_2_id`` ← the ``person_id`` of the household + member the person's PEPAR1/PEPAR2 pointer resolves to within the same + PH_SEQ, and ``0`` when the pointer is absent (<= 0), does not resolve + to a co-resident line, or points at the person themselves. These are + the same resolution ``own_children_in_household`` counts, kept as the + relation rather than only its cardinality, so downstream rules can ask + *whose* child a person is (microcosm#884, + policyengine-us#9404). ``person_id`` already exists on the person table + when this stage runs — the stage aligns its output on ``person_id`` — + so the pointers resolve straight to the exported identifier and no + later id-mapping pass is needed. Stored as int64, never float: the PUF + support clone remaps person ids up to ``10**16``, past the 2**53 bound + where float64 stops representing integers exactly. - ``veterans_benefits`` ← VET_VAL (veterans' payments received). Ownership note: ``is_veteran`` is formula-owned in PolicyEngine-US @@ -44,10 +57,21 @@ dollars and never the derived flag — storing formula-owned variables is the over-export failure of microcosm #24. -Healing behavior: a frame that already carries all five columns *with -signal* passes through untouched (idempotent). A frame carrying a +Healing behavior: a frame that already carries every output column +*with signal* passes through untouched (idempotent). A frame carrying a constant ``is_disabled`` — indistinguishable from the engine's broadcast default — is recomputed from the raw columns rather than trusted. + +Clone/pool note: the operator is declared ``pre_clone`` and +``cps_source``-scoped (``POOL_OPERATOR_CONTRACTS`` in +``multispine_pool``), so the pointers are resolved once per source +person, against PH_SEQ values ``asec_pool`` has already made unique +across pooled ASEC years. The PUF support clone that follows renumbers +``person_id``; :func:`microcosm.build.us_runtime.puf_support.clone_us_frame_for_puf_support` +therefore shifts these two columns with the rest of the person id +surface, preserving ``0``. The gate below re-checks co-residence and the +``own_children_in_household`` identity on whatever frame it is handed, +so a missed remap fails closed instead of exporting a stale pointer. """ from __future__ import annotations @@ -75,6 +99,7 @@ __all__ = [ "US_ELIGIBILITY_INPUTS_NONCONSTANT_PERSON_COLUMNS", "US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS", + "US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS", "US_ELIGIBILITY_INPUTS_REQUIRED_SOURCE_COLUMNS", "US_ELIGIBILITY_INPUTS_STAGE_NAME", "derive_us_eligibility_inputs_from_manifest", @@ -86,12 +111,22 @@ US_ELIGIBILITY_INPUTS_STAGE_NAME = "eligibility_inputs" +#: Person-referencing id columns this stage owns, in PEPAR pointer order. +#: They hold a co-resident parent's ``person_id`` (``0`` = unknown or +#: absent), so any operator that renumbers ``person_id`` must shift these +#: with it — see ``puf_support.clone_us_frame_for_puf_support``. +US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: tuple[str, ...] = ( + "parent_1_id", + "parent_2_id", +) + #: The PolicyEngine-facing person input columns this stage owns. US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS: tuple[str, ...] = ( "is_disabled", "is_blind", "is_full_time_college_student", "own_children_in_household", + *US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, "veterans_benefits", ) @@ -110,6 +145,15 @@ "PEDISREM", ) +#: The two ASEC parent line pointers, in ``parent_1_id``/``parent_2_id`` +#: order. Positional correspondence with +#: :data:`US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS` is load-bearing. +_PARENT_POINTER_COLUMNS: tuple[str, ...] = ("PEPAR1", "PEPAR2") + +#: The frame's person identifier. Not an ASEC source column: it is the +#: structural id the stage aligns its output on, so it is already present. +_PERSON_ID_COLUMN = "person_id" + #: Raw CPS ASEC person columns the derivation reads. PH_SEQ/A_LINENO are #: the household sequence and person line number the parent pointers #: (PEPAR1/PEPAR2) resolve against; SSI_VAL and A_AGE feed the reported-SSI @@ -118,8 +162,7 @@ *_DISABILITY_DIFFICULTY_COLUMNS, "A_HSCOL", "A_FTPT", - "PEPAR1", - "PEPAR2", + *_PARENT_POINTER_COLUMNS, "PH_SEQ", "A_LINENO", "VET_VAL", @@ -169,6 +212,22 @@ def us_eligibility_inputs_stage_spec() -> SourceStageSpec: return stage_map[US_ELIGIBILITY_INPUTS_STAGE_NAME] +def _parent_pointers(frame: pd.DataFrame) -> tuple[np.ndarray, list[np.ndarray]]: + """The household key and the two numeric parent line pointers. + + Shared by the count and the id resolution so the two can never read + the pointers differently — the release gate asserts the identity + between them, and one normalization is what makes it hold. + """ + + household = pd.to_numeric(frame["PH_SEQ"], errors="coerce").to_numpy() + pointers = [ + pd.to_numeric(frame[column], errors="coerce").to_numpy() + for column in _PARENT_POINTER_COLUMNS + ] + return household, pointers + + def _own_children_in_household(frame: pd.DataFrame) -> np.ndarray: """Count each person's own children in the household via parent pointers. @@ -178,10 +237,9 @@ def _own_children_in_household(frame: pd.DataFrame) -> np.ndarray: """ counts: dict[tuple[float, float], int] = {} - ph_seq = pd.to_numeric(frame["PH_SEQ"], errors="coerce").to_numpy() - for pointer_column in ("PEPAR1", "PEPAR2"): - pointer = pd.to_numeric(frame[pointer_column], errors="coerce").to_numpy() - for household, parent_line in zip(ph_seq, pointer, strict=True): + household_key, pointers = _parent_pointers(frame) + for pointer in pointers: + for household, parent_line in zip(household_key, pointer, strict=True): if parent_line > 0: key = (household, parent_line) counts[key] = counts.get(key, 0) + 1 @@ -189,12 +247,72 @@ def _own_children_in_household(frame: pd.DataFrame) -> np.ndarray: return np.array( [ float(counts.get((household, line), 0)) - for household, line in zip(ph_seq, line_number, strict=True) + for household, line in zip(household_key, line_number, strict=True) ], dtype=np.float64, ) +def _parent_person_ids(frame: pd.DataFrame) -> dict[str, np.ndarray]: + """Resolve PEPAR1/PEPAR2 to the pointed-at member's ``person_id``. + + The pointer is a within-household A_LINENO, so the resolution keys on + ``(PH_SEQ, A_LINENO)`` — exactly the pair + :func:`_own_children_in_household` counts against. A pointer that is + absent (``<= 0``), missing, or names a line no co-resident row + carries resolves to ``0``: "unknown or absent", the value + PolicyEngine-US reads as "fall back to the count-based proxy". + + Raises: + SourceRuntimeError: If ``person_id`` is absent, or if a + ``(PH_SEQ, A_LINENO)`` pair repeats. A repeated line makes the + pointer ambiguous and would silently break the count identity, + so it fails closed rather than picking a row. + """ + + if _PERSON_ID_COLUMN not in frame.columns: + raise SourceRuntimeError( + f"US eligibility-inputs parent-id resolution requires the person " + f"table's {_PERSON_ID_COLUMN!r} column; the stage aligns its " + "output on it, so it is present before the derivation runs." + ) + household_key, pointers = _parent_pointers(frame) + line_number = pd.to_numeric(frame["A_LINENO"], errors="coerce").to_numpy() + person_id = pd.to_numeric(frame[_PERSON_ID_COLUMN], errors="raise").to_numpy( + dtype=np.int64 + ) + person_id_by_line: dict[tuple[float, float], int] = {} + for household, line, identifier in zip( + household_key, line_number, person_id, strict=True + ): + if not np.isfinite(household) or not np.isfinite(line): + # An unusable key can never be pointed at: the count treats it + # the same way, so both surfaces read 0 for that row. + continue + key = (float(household), float(line)) + if key in person_id_by_line: + raise SourceRuntimeError( + "US eligibility-inputs parent-id resolution requires " + "(PH_SEQ, A_LINENO) to identify at most one person; " + f"household {key[0]:.0f} repeats line {key[1]:.0f}." + ) + person_id_by_line[key] = int(identifier) + resolved: dict[str, np.ndarray] = {} + for column, pointer in zip( + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, pointers, strict=True + ): + resolved[column] = np.array( + [ + person_id_by_line.get((float(household), float(parent_line)), 0) + if parent_line > 0 and np.isfinite(household) + else 0 + for household, parent_line in zip(household_key, pointer, strict=True) + ], + dtype=np.int64, + ) + return resolved + + def derive_us_eligibility_inputs_from_manifest( frame: pd.DataFrame | None, operation: SourceOperationSpec, @@ -256,6 +374,8 @@ def _numeric(column: str) -> pd.Series: (_numeric("A_HSCOL") == 2) & (_numeric("A_FTPT") == 1) ).to_numpy() result["own_children_in_household"] = _own_children_in_household(frame) + for column, values in _parent_person_ids(frame).items(): + result[column] = values result["veterans_benefits"] = np.maximum( _numeric("VET_VAL").fillna(0.0).to_numpy(dtype=np.float64), 0.0 ) @@ -276,7 +396,7 @@ def _disabled_carries_signal(person: pd.DataFrame) -> bool: def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> Frame: """Run the ``eligibility_inputs`` manifest stage over a US frame. - A frame already carrying all five output columns with a non-constant + A frame already carrying every output column with a non-constant disability distribution passes through untouched (idempotent). Any other surface — columns missing, or disability constant at the engine default — is recomputed from the raw ASEC columns. @@ -290,13 +410,15 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> time_period: The dataset's time period. Returns: - A new frame whose person table carries all five eligibility - columns. + A new frame whose person table carries every eligibility column. + The parent-id columns land as int64 so a clone-remapped + ``person_id`` past 2**53 stays exact. Raises: ValueError: If the frame is not US-schema or the stage output does not cover every person. - SourceRuntimeError: If required raw ASEC columns are missing. + SourceRuntimeError: If required raw ASEC columns are missing, or if + the parent pointers cannot be resolved unambiguously. """ if frame.schema != US_SCHEMA: @@ -331,6 +453,10 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> for column in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS: if column in bool_columns: tables["person"][column] = aligned[column].to_numpy(dtype=bool) + elif column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + tables["person"][column] = pd.to_numeric( + aligned[column], errors="raise" + ).to_numpy(dtype=np.int64) else: tables["person"][column] = aligned[column].to_numpy(dtype=np.float64) return Frame( @@ -364,7 +490,7 @@ def _share(mask: np.ndarray) -> float: for column in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS if column in person.columns } - return { + summary: dict[str, object] = { "disabled_share": _share(disabled), "full_time_college_student_share": _share(student), "parent_share": _share(children.to_numpy() > 0), @@ -375,6 +501,139 @@ def _share(mask: np.ndarray) -> float: "veterans_benefits_share_band": list(_VETERANS_BENEFITS_SHARE_BAND), "unique_counts": unique_counts, } + if all( + column in person.columns for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + ): + pointed_at_any = np.zeros(len(person), dtype=bool) + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + values = pd.to_numeric(person[column], errors="coerce").fillna(0).to_numpy() + pointed_at_any |= values > 0 + summary[f"{column}_resolved_share"] = _share(values > 0) + # Diagnostic only: reported, never banded. The invariants below are + # exact identities, so this stage does not guess an empirical band + # for a surface no published build has measured yet. + summary["any_parent_id_resolved_share"] = _share(pointed_at_any) + return summary + + +def _parent_id_invariant_failures(person: pd.DataFrame) -> list[str]: + """Exact identities every resolved parent-id surface must satisfy. + + These are not plausibility bands. Each is a structural consequence of + "``parent_k_id`` is the ``person_id`` of a co-resident parent, or 0", + so any violation is a build defect — a stale pointer carried through + an id remap, a dropped household member, a donated id — rather than an + unusual population. + """ + + failures: list[str] = [] + if _PERSON_ID_COLUMN not in person.columns: + return [ + f"{_PERSON_ID_COLUMN!r} is absent, so the parent-id identities " + "cannot be checked." + ] + person_id = pd.to_numeric(person[_PERSON_ID_COLUMN], errors="coerce") + if person_id.isna().any(): + return [f"{_PERSON_ID_COLUMN!r} carries non-numeric values."] + person_id = person_id.to_numpy(dtype=np.int64) + known_ids = set(person_id.tolist()) + + resolved: dict[str, np.ndarray] = {} + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + values = pd.to_numeric(person[column], errors="coerce") + if values.isna().any(): + failures.append( + f"{column}: {int(values.isna().sum())} row(s) carry no value; " + "an unknown parent must be stored as 0, never as null." + ) + values = values.fillna(0) + integral = values.to_numpy(dtype=np.float64) + if not np.all(integral == np.floor(integral)): + failures.append(f"{column}: carries non-integral person ids.") + column_ids = values.fillna(0).to_numpy(dtype=np.int64) + if (column_ids < 0).any(): + failures.append(f"{column}: carries negative person ids.") + resolved[column] = column_ids + + for column, column_ids in resolved.items(): + pointed = column_ids > 0 + unknown = { + int(value) + for value in np.unique(column_ids[pointed]) + if int(value) not in known_ids + } + if unknown: + failures.append( + f"{column}: {len(unknown)} value(s) name no person in the " + f"table (examples {sorted(unknown)[:5]}) — a stale pointer " + "survives an id remap that did not shift this column." + ) + self_pointed = pointed & (column_ids == person_id) + if self_pointed.any(): + failures.append( + f"{column}: {int(self_pointed.sum())} person(s) are recorded " + "as their own parent." + ) + + household_column = "person_household_id" + if household_column in person.columns: + household = pd.to_numeric(person[household_column], errors="coerce") + if household.isna().any(): + failures.append(f"{household_column!r} carries non-numeric values.") + else: + household_by_id = dict( + zip(person_id.tolist(), household.to_numpy().tolist(), strict=True) + ) + for column, column_ids in resolved.items(): + pointed = column_ids > 0 + if not pointed.any(): + continue + parent_household = np.array( + [ + household_by_id.get(int(value), np.nan) + for value in column_ids[pointed] + ], + dtype=np.float64, + ) + offsite = parent_household != household.to_numpy()[pointed] + if offsite.any(): + failures.append( + f"{column}: {int(offsite.sum())} pointer(s) name a " + "person outside the pointing person's household; a " + "parent id is only ever a co-resident." + ) + + expected = np.zeros(len(person), dtype=np.float64) + position_by_id = { + identifier: index for index, identifier in enumerate(person_id.tolist()) + } + for column_ids in resolved.values(): + for value in column_ids[column_ids > 0].tolist(): + position = position_by_id.get(int(value)) + if position is not None: + expected[position] += 1.0 + observed = ( + pd.to_numeric(person["own_children_in_household"], errors="coerce") + .fillna(-1.0) + .to_numpy(dtype=np.float64) + ) + mismatched = observed != expected + if mismatched.any(): + examples = [ + { + "person_id": int(person_id[index]), + "own_children_in_household": float(observed[index]), + "pointed_at_by": float(expected[index]), + } + for index in np.flatnonzero(mismatched)[:5] + ] + failures.append( + f"own_children_in_household disagrees with the parent pointers " + f"for {int(mismatched.sum())} person(s): the count must equal the " + f"number of people whose parent id names them. Examples: " + f"{examples}." + ) + return failures def us_eligibility_inputs_signal_gate(frame: Frame) -> GateResult: @@ -384,6 +643,14 @@ def us_eligibility_inputs_signal_gate(frame: Frame) -> GateResult: disabled, full-time-student, parent, or veterans'-payment shares leave their plausibility bands — each of which reproduces (or inverts) the everyone-defaults failure of microcosm #244. + + It also asserts the parent-id identities from + :func:`_parent_id_invariant_failures`: every resolved pointer names a + co-resident person, nobody parents themselves, and + ``own_children_in_household`` equals the number of people whose parent + ids name that person (microcosm#884). Those are exact identities, so + they fail closed on a stale or donated id rather than on an unusual + population. """ person = frame.table("person") @@ -428,6 +695,7 @@ def us_eligibility_inputs_signal_gate(frame: Frame) -> GateResult: failures.append( f"{label} {share:.3f} outside plausibility band [{low}, {high}]." ) + failures.extend(_parent_id_invariant_failures(person)) return GateResult( name="eligibility_inputs_signal", passed=not failures, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py b/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py index 5397a2d9f1..9bfdd254ea 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py @@ -44,6 +44,7 @@ ) from microcosm.build.us_runtime.education_inputs import with_us_education_inputs from microcosm.build.us_runtime.eligibility_inputs import ( + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, with_us_eligibility_inputs, ) from microcosm.build.us_runtime.energy_subsidy import ( @@ -675,6 +676,15 @@ class SourceOperatorContract: "age", "is_female", "is_household_head", + # The parent ids are complete on both arms without a transfer: + # the eligibility operator resolves them on the CPS projection + # and ``map_acs_native_inputs`` writes the ACS spine's declared + # 0. They must never enter the QRF plan — parent_1_id is not a + # PolicyEngine-US variable, so ``_target_encoding`` would treat + # it as continuous and hand an ACS child a fractional + # interpolation between two unrelated ASEC person ids + # (microcosm#884). + *US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, } ), "household": frozenset({"tenure_type"}), diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py b/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py index 9452473527..36405d1b4f 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py @@ -14,6 +14,10 @@ import pandas as pd +from microcosm.build.us_runtime.acs_inputs import ( + _ACS_UNRESOLVED_PARENT_POINTER_TRANSFORMATION, + ACS_UNRESOLVED_PARENT_ID_MAPPINGS, +) from microcosm.build.us_runtime.acs_transfer import ACS_DERIVED_TRANSFER_INPUTS from microcosm.build.us_runtime.adult_care import US_ADULT_CARE_OUTPUT_COLUMNS from microcosm.build.us_runtime.child_support import ( @@ -190,6 +194,14 @@ ("TAXAMT", "ADJHSG", "RELSHIPP"), "TAXAMT * ADJHSG / 1_000_000; reference-person carry", ), + **{ + output: ( + "person", + (pointer_column,), + _ACS_UNRESOLVED_PARENT_POINTER_TRANSFORMATION, + ) + for output, pointer_column in ACS_UNRESOLVED_PARENT_ID_MAPPINGS + }, } _CAPITAL_GAINS_TAIL_PROVENANCE_COLUMNS = frozenset( { diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py b/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py index e11719325d..f0839ea840 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py @@ -24,6 +24,9 @@ ACS_PUMS_EARNINGS_SOURCE_COLUMNS, resolve_acs_pums_earnings_universe, ) +from microcosm.build.us_runtime.eligibility_inputs import ( + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, +) from microcosm.build.us_runtime.puf_e01000_reconciliation import ( PUF_SCHEDULE_D_JOINT_COLUMNS, puf_capital_gains_joint_metrics, @@ -56,6 +59,15 @@ QRF: Any | None = None +#: Person columns whose *values* are ``person_id``s rather than payload. +#: The support clone renumbers ``person_id``, so these move with it; 0 is the +#: "no such person" sentinel and is left alone. Source identities +#: (``source_person_id`` and friends) are deliberately not here — several +#: operators key clone-stable draws on them precisely because they do not move. +US_PERSON_REFERENCE_ID_COLUMNS: tuple[str, ...] = ( + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS +) + __all__ = [ "BASE_ASEC_SUPPORT_CHANNEL", "PUF_TAX_DETAIL_DEMOGRAPHIC_SOURCES", @@ -73,6 +85,7 @@ "PUF_DONOR_SOURCE_ADJUSTED_GROSS_INCOME_COLUMN", "US_PUF_DONOR_MORTGAGE_QUARANTINE_FIELDS", "US_PUF_DONOR_MORTGAGE_OUTLIER_CEILING", + "US_PERSON_REFERENCE_ID_COLUMNS", "US_PUF_SUPPORT_FIT_NAME", "US_PUF_SUPPORT_STAGE_NAME", "assert_formula_owned_blocklist_current", @@ -2153,6 +2166,7 @@ def _clone_entity_table( id_multiplier: int, ) -> pd.DataFrame: id_columns = _entity_id_columns(schema, entity) + reference_columns = _person_reference_columns(table, schema=schema, entity=entity) source_id = support_source_id_column(entity) channel_column = support_channel_column(entity) clone_index_column = support_clone_index_column(entity) @@ -2170,6 +2184,12 @@ def _clone_entity_table( clone_index=clone_index, id_multiplier=id_multiplier, ) + for column in reference_columns: + clone[column] = _remap_person_reference_ids( + clone[column].to_numpy(), + clone_index=clone_index, + id_multiplier=id_multiplier, + ) clones.append(clone) result = pd.concat(clones, ignore_index=True) if result[primary_id].duplicated().any(): @@ -2192,6 +2212,7 @@ def _clone_preassembled_entity_table( """Clone one preassembled entity table while preserving source provenance.""" id_columns = _entity_id_columns(schema, entity) + reference_columns = _person_reference_columns(table, schema=schema, entity=entity) clone_index_column = support_clone_index_column(entity) primary_id = schema.entity_id_column(entity) @@ -2204,6 +2225,12 @@ def _clone_preassembled_entity_table( clone_index=PUF_TAX_DETAIL_CLONE_INDEX, id_multiplier=id_multiplier, ) + for column in reference_columns: + detail[column] = _remap_person_reference_ids( + detail[column].to_numpy(), + clone_index=PUF_TAX_DETAIL_CLONE_INDEX, + id_multiplier=id_multiplier, + ) result = pd.concat([native, detail], ignore_index=True) if result[primary_id].duplicated().any(): duplicates = result.loc[result[primary_id].duplicated(), primary_id].unique() @@ -2257,6 +2284,48 @@ def _entity_id_columns(schema: EntitySchema, entity: str) -> tuple[str, ...]: return (schema.entity_id_column(entity),) +def _person_reference_columns( + table: pd.DataFrame, *, schema: EntitySchema, entity: str +) -> tuple[str, ...]: + """Non-structural person columns whose values are ``person_id``s. + + The clone renumbers ``person_id``, so a column that *names* a person has + to move with it or it silently points into the other support arm. These + are the only such columns on the US person table + (:data:`US_PERSON_REFERENCE_ID_COLUMNS`); ``source_person_id`` and its + siblings are deliberately excluded — a source identity is the + clone-stable key several operators draw on. + """ + + if entity != schema.person_entity: + return () + return tuple( + column for column in US_PERSON_REFERENCE_ID_COLUMNS if column in table.columns + ) + + +def _remap_person_reference_ids( + values: Sequence[Any], + *, + clone_index: int, + id_multiplier: int, +) -> np.ndarray: + """Shift a person-referencing column, leaving the 0 sentinel alone. + + ``0`` means "no such person" on these columns, so it must not be shifted + into ``clone_index * id_multiplier`` — a valid-looking id naming a + stranger. + """ + + shifted = _remap_ids( + values, + clone_index=clone_index, + id_multiplier=id_multiplier, + ) + original = _validated_integral_ids(values, label="PUF support person references") + return np.where(original == 0, 0, shifted) + + def _validate_channels(channels: Sequence[str]) -> tuple[str, ...]: if isinstance(channels, str): raise ValueError("support channels must be a sequence of names, not a string.") diff --git a/tools/generate_us_bundle_from_constants.py b/tools/generate_us_bundle_from_constants.py index 131991f695..5dc45835db 100644 --- a/tools/generate_us_bundle_from_constants.py +++ b/tools/generate_us_bundle_from_constants.py @@ -122,7 +122,7 @@ def _domain_builders() -> dict[str, Callable[[], object]]: # frozen files, is the forward YAML -> legacy-payload path. FROZEN_LEGACY_RESOURCE_SHA256 = { "source_stages.json": ( - "9f4f983091fb18dea8f4524cc100be20f9fca2b006435461592f8edf53c9a5d8" + "e75217c0db3075bf2664492167d31ce7ee82c2c5f6a70e3b67e8bb6f4ddd7ecf" ), "support_spine.json": ( "68f37dc6ae6e0cde7ebccb53f88dd4a800e63456f838fa214ff98d1db8d815be" From 1c34a2b860621802d098f857e240889c070ea7cd Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 7 Sep 2026 17:14:57 -0400 Subject: [PATCH 2/7] Cover the parent-id surface with tests and keep ids out of input mass - Stage tests: two-parent, single-parent, three-generation (the grandchild names only the parent), unresolved and absent pointers, int64 dtype, person_id-not-line-number, missing person_id, and a repeated (PH_SEQ, A_LINENO). - Clone tests: each support arm's child names that arm's parent, and the 0 sentinel is not shifted into a valid-looking id. - Gate tests: a pointer naming no person, one outside the household, self-parenting, a count disagreeing with the pointers, a null id, and the person_id-0 case, which cannot be named by a 0-means-unknown pointer and is reported rather than failed. - ACS tests: the native mapping writes the declared unknown sentinel, the receipt satisfies the operator boundary, and the columns stay absent when the raw pointers are. - us_input_mass_totals now drops the person-referencing id columns: the weighted sum of an arbitrary integer label is not a mass, and it moves whenever person ids are renumbered, so comparing it against a reference release on a different id space would fail for no reason. Co-Authored-By: Claude Fable 5.1 --- .../microcosm/build/us_runtime/__init__.py | 2 + .../build/us_runtime/eligibility_inputs.py | 43 ++- .../microcosm/build/us_runtime/input_mass.py | 42 ++- .../tests/test_us_acs_inputs.py | 51 ++++ .../tests/test_us_eligibility_inputs.py | 273 +++++++++++++++++- 5 files changed, 398 insertions(+), 13 deletions(-) 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 a88d1de098..74e4757cba 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py @@ -243,6 +243,7 @@ from microcosm.build.us_runtime.eligibility_inputs import ( US_ELIGIBILITY_INPUTS_NONCONSTANT_PERSON_COLUMNS, US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS, + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, US_ELIGIBILITY_INPUTS_REQUIRED_SOURCE_COLUMNS, US_ELIGIBILITY_INPUTS_STAGE_NAME, derive_us_eligibility_inputs_from_manifest, @@ -1249,6 +1250,7 @@ "us_pregnancy_summary", "with_us_pregnancy_inputs", "US_ELIGIBILITY_INPUTS_NONCONSTANT_PERSON_COLUMNS", + "US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS", "US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS", "US_ELIGIBILITY_INPUTS_REQUIRED_SOURCE_COLUMNS", "US_ELIGIBILITY_INPUTS_STAGE_NAME", diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py index a3ccc5a52f..9b65f293ef 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py @@ -40,7 +40,9 @@ - ``parent_1_id`` / ``parent_2_id`` ← the ``person_id`` of the household member the person's PEPAR1/PEPAR2 pointer resolves to within the same PH_SEQ, and ``0`` when the pointer is absent (<= 0), does not resolve - to a co-resident line, or points at the person themselves. These are + to a co-resident line. A person recorded as their own parent is left + as the data says and reported by the gate, never quietly rewritten. + These are the same resolution ``own_children_in_household`` counts, kept as the relation rather than only its cardinality, so downstream rules can ask *whose* child a person is (microcosm#884, @@ -50,6 +52,17 @@ later id-mapping pass is needed. Stored as int64, never float: the PUF support clone remaps person ids up to ``10**16``, past the 2**53 bound where float64 stops representing integers exactly. + + Sentinel note: ``assign_us_unit_structure`` mints ``person_id`` as a + 0-based ``arange``, so exactly one person in a frame carries id 0 and + cannot be named by a 0-means-unknown pointer. Their co-resident + children therefore read 0 and fall back to PolicyEngine-US's + count-based proxy; :func:`us_eligibility_inputs_summary` reports how + many pointers that costs (``pointers_unnameable_at_person_id_zero``) + so it stays visible rather than silent, and the identity check + excludes that one person for the same reason. Widening the sentinel is + a contract change on the PolicyEngine-US side (policyengine-us#9404 + specifies 0), not something this stage decides. - ``veterans_benefits`` ← VET_VAL (veterans' payments received). Ownership note: ``is_veteran`` is formula-owned in PolicyEngine-US @@ -513,9 +526,32 @@ def _share(mask: np.ndarray) -> float: # exact identities, so this stage does not guess an empirical band # for a surface no published build has measured yet. summary["any_parent_id_resolved_share"] = _share(pointed_at_any) + summary["pointers_unnameable_at_person_id_zero"] = ( + _pointers_unnameable_at_person_id_zero(person) + ) return summary +def _pointers_unnameable_at_person_id_zero(person: pd.DataFrame) -> int: + """Pointers lost because the parent they name is the person holding id 0. + + ``person_id`` is 0-based and ``0`` is the "unknown parent" sentinel, so + one person per frame cannot be named. Their children read 0 and fall back + to the count-based proxy; this reports how many that is. + """ + + if _PERSON_ID_COLUMN not in person.columns: + return 0 + identifiers = pd.to_numeric(person[_PERSON_ID_COLUMN], errors="coerce") + at_zero = identifiers == 0 + if not bool(at_zero.any()): + return 0 + children = pd.to_numeric( + person.loc[at_zero, "own_children_in_household"], errors="coerce" + ) + return int(children.fillna(0.0).sum()) + + def _parent_id_invariant_failures(person: pd.DataFrame) -> list[str]: """Exact identities every resolved parent-id surface must satisfy. @@ -617,7 +653,10 @@ def _parent_id_invariant_failures(person: pd.DataFrame) -> list[str]: .fillna(-1.0) .to_numpy(dtype=np.float64) ) - mismatched = observed != expected + # The person holding id 0 cannot be named by a 0-means-unknown pointer, + # so their count is unverifiable here by construction. It is reported as + # ``pointers_unnameable_at_person_id_zero`` rather than silently passing. + mismatched = (observed != expected) & (person_id != 0) if mismatched.any(): examples = [ { diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py b/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py index 2aa0ba66ff..79b780ebdf 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py @@ -1,15 +1,43 @@ -"""US alias for the shared weighted per-column input-mass helper. +"""US weighted per-column input-mass helper. -The implementation was promoted to :mod:`microcosm.build.input_mass` when the -UK terminal battery adopted the same #278 input-mass parity gate (#609); the -computation is schema-driven and never carried US-specific logic. This module -remains so existing US call sites keep their import path. +The computation was promoted to :mod:`microcosm.build.input_mass` when the UK +terminal battery adopted the same #278 input-mass parity gate (#609); it is +schema-driven and carries no country logic. This module remains so existing US +call sites keep their import path, and now applies the one US-specific +subtraction the shared helper cannot make: the person-referencing id columns. + +``input_mass_totals`` treats every numeric non-structural column as a measured +layer and reports its weighted sum, but its structural set is only the entity +ids and memberships. ``parent_1_id`` is neither: it is an arbitrary integer +label whose weighted sum means nothing, and it moves whenever person ids are +renumbered — so comparing it against a reference release built on a different +id space would fail the parity gate for no reason (microcosm#884). """ from __future__ import annotations +from collections.abc import Iterable, Mapping + from microcosm.build.input_mass import input_mass_totals +from microcosm.build.us_runtime.puf_support import US_PERSON_REFERENCE_ID_COLUMNS +from microcosm.frame import Frame + +__all__ = ["US_INPUT_MASS_EXCLUDED_COLUMNS", "us_input_mass_totals"] + +#: Columns whose weighted sum is not a mass. See the module docstring. +US_INPUT_MASS_EXCLUDED_COLUMNS: frozenset[str] = frozenset( + US_PERSON_REFERENCE_ID_COLUMNS +) + -__all__ = ["us_input_mass_totals"] +def us_input_mass_totals( + frame: Frame, + *, + columns: Iterable[str] | None = None, +) -> Mapping[str, float]: + """Weighted per-column totals, less the person-referencing id columns.""" -us_input_mass_totals = input_mass_totals + totals = dict(input_mass_totals(frame, columns=columns)) + for column in US_INPUT_MASS_EXCLUDED_COLUMNS: + totals.pop(column, None) + return totals diff --git a/packages/microcosm-build/tests/test_us_acs_inputs.py b/packages/microcosm-build/tests/test_us_acs_inputs.py index d476a453a8..8ae28e141c 100644 --- a/packages/microcosm-build/tests/test_us_acs_inputs.py +++ b/packages/microcosm-build/tests/test_us_acs_inputs.py @@ -27,6 +27,10 @@ def _acs_frame() -> Frame: "SSIP": [200.0, np.nan], "RETP": [300.0, np.nan], "INTP": [-100.0, np.nan], + # acs_pums synthesizes these: the child (RELSHIPP 25) is pointed + # at the reference person's line. + "PEPAR1": [0, 1], + "PEPAR2": [0, 0], } ) tables = { @@ -314,3 +318,50 @@ def test_acs_unmapped_missing_source_columns_stay_absent() -> None: assert "acs_interest_dividend_rental_income" not in result.frame.table("person") assert "acs_retirement_income" not in result.native_inputs assert "acs_interest_dividend_rental_income" not in result.native_inputs + + +def test_acs_parent_pointers_map_to_the_declared_unknown_sentinel() -> None: + # microcosm#884: the ACS PEPAR values are a synthesized reference-person + # link, and an ACS-minted person_id is renumbered by the assembly offset + # after this mapping runs. The spine therefore declares the pointer + # unknown rather than exporting either construct as a parent id. + result = map_acs_native_inputs(_acs_frame()) + person = result.frame.table("person") + assert person["parent_1_id"].tolist() == [0, 0] + assert person["parent_2_id"].tolist() == [0, 0] + assert person["parent_1_id"].dtype == np.int64 + for output, source in (("parent_1_id", "PEPAR1"), ("parent_2_id", "PEPAR2")): + receipt = result.native_inputs[output] + assert receipt["entity"] == "person" + assert receipt["source_columns"] == [source] + assert receipt["missing_rows"] == 0 + assert receipt["observed_rows"] == 2 + + +def test_acs_parent_pointer_receipt_satisfies_the_operator_boundary() -> None: + from microcosm.build.us_runtime.operator_boundary import ( + assert_operator_free_source_frame, + ) + + result = map_acs_native_inputs(_acs_frame()) + assert_operator_free_source_frame( + result.frame, + label="ACS native-mapped pool input", + native_inputs=result.native_inputs, + ) + + +def test_acs_parent_pointers_stay_absent_without_the_raw_pointers() -> None: + frame = _acs_frame() + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables["person"] = tables["person"].drop(columns=["PEPAR1", "PEPAR2"]) + stripped = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + ) + result = map_acs_native_inputs(stripped) + person = result.frame.table("person") + assert "parent_1_id" not in person.columns + assert "parent_2_id" not in person.columns diff --git a/packages/microcosm-build/tests/test_us_eligibility_inputs.py b/packages/microcosm-build/tests/test_us_eligibility_inputs.py index c4a98df4d0..6f29146b4f 100644 --- a/packages/microcosm-build/tests/test_us_eligibility_inputs.py +++ b/packages/microcosm-build/tests/test_us_eligibility_inputs.py @@ -10,6 +10,7 @@ from microcosm.build.source_runtime import SourceRuntimeError from microcosm.build.us_runtime import ( US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS, + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, US_ELIGIBILITY_INPUTS_REQUIRED_SOURCE_COLUMNS, US_ELIGIBILITY_INPUTS_STAGE_NAME, derive_us_eligibility_inputs_from_manifest, @@ -93,6 +94,21 @@ def _us_frame( ) +def _replace_person(frame: Frame, person: pd.DataFrame) -> Frame: + """The frame with a hand-edited person table, for gate-failure cases.""" + + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables["person"] = person + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + metadata=frame.metadata, + ) + + def _plausible_rows() -> list[dict]: """Fifty persons whose weighted shares land inside every band. @@ -106,8 +122,14 @@ def _plausible_rows() -> list[dict]: rows: list[dict] = [{} for _ in range(10)] # lines 1-10: parents rows += [ - {"PEPAR1": parent_line, "A_AGE": 8} for parent_line in range(1, 11) - ] # lines 11-20: one child per parent + # lines 11-20: one child per parent. The first child names a second + # parent as well, so parent_2_id carries signal without moving the + # parent share: line 2 already parents line 12. + {"PEPAR1": parent_line, "PEPAR2": 2, "A_AGE": 8} + if parent_line == 1 + else {"PEPAR1": parent_line, "A_AGE": 8} + for parent_line in range(1, 11) + ] rows += [ {"PEDISPHY": 1}, {"PEDISREM": 1}, @@ -130,15 +152,25 @@ def _plausible_rows() -> list[dict]: class TestManifestDeclaration: - def test_stage_is_declared_with_the_five_output_columns(self) -> None: + def test_stage_is_declared_with_its_output_columns(self) -> None: spec = us_eligibility_inputs_stage_spec() assert spec.stage == US_ELIGIBILITY_INPUTS_STAGE_NAME assert tuple(spec.outputs) == US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS assert set(spec.nonnegative_outputs) == { "own_children_in_household", + *US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, "veterans_benefits", } + def test_parent_ids_are_declared_outputs(self) -> None: + assert US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS == ( + "parent_1_id", + "parent_2_id", + ) + assert set(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS) <= set( + US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS + ) + def test_handler_is_registered(self) -> None: handlers = us_source_operation_handlers() assert ( @@ -218,6 +250,91 @@ def test_parent_pointers_do_not_cross_households(self) -> None: result = self._derive(table) assert result["own_children_in_household"].tolist() == [0.0, 1.0, 0.0] + def test_two_parent_household_resolves_both_pointers(self) -> None: + table = _person_table( + [ + {}, # line 1 / person_id 1: parent + {}, # line 2 / person_id 2: parent + {"PEPAR1": 1, "PEPAR2": 2, "A_AGE": 9}, + ] + ) + result = self._derive(table) + assert result["parent_1_id"].tolist() == [0, 0, 1] + assert result["parent_2_id"].tolist() == [0, 0, 2] + assert result["own_children_in_household"].tolist() == [1.0, 1.0, 0.0] + + def test_single_parent_leaves_the_second_pointer_unset(self) -> None: + table = _person_table([{}, {"PEPAR1": 1, "A_AGE": 4}]) + result = self._derive(table) + assert result["parent_1_id"].tolist() == [0, 1] + assert result["parent_2_id"].tolist() == [0, 0] + + def test_three_generation_household_points_children_at_their_own_parent( + self, + ) -> None: + # Line 1 grandparent, line 2 their adult child, line 3 the grandchild. + # The grandchild's pointer names line 2 only: the count-based proxy + # would treat the grandparent as a second parent (microcosm#884). + table = _person_table( + [ + {"A_AGE": 62}, + {"PEPAR1": 1, "A_AGE": 30}, + {"PEPAR1": 2, "A_AGE": 6}, + ] + ) + result = self._derive(table) + assert result["parent_1_id"].tolist() == [0, 1, 2] + assert result["parent_2_id"].tolist() == [0, 0, 0] + assert result["own_children_in_household"].tolist() == [1.0, 1.0, 0.0] + + def test_unresolved_pointer_reads_zero(self) -> None: + # Line 9 does not exist in the household, and line 1 of the other + # household is not reachable from household 2. + table = _person_table( + [ + {"PH_SEQ": 1, "A_LINENO": 1}, + {"PH_SEQ": 2, "A_LINENO": 1}, + {"PH_SEQ": 2, "A_LINENO": 2, "PEPAR1": 9, "PEPAR2": 1, "A_AGE": 8}, + ] + ) + result = self._derive(table) + assert result["parent_1_id"].tolist() == [0, 0, 0] + assert result["parent_2_id"].tolist() == [0, 0, 2] + + def test_absent_pointer_sentinels_read_zero(self) -> None: + table = _person_table([{}, {"PEPAR1": 0, "PEPAR2": -1, "A_AGE": 8}]) + result = self._derive(table) + assert result["parent_1_id"].tolist() == [0, 0] + assert result["parent_2_id"].tolist() == [0, 0] + + def test_parent_ids_are_int64_not_float(self) -> None: + table = _person_table([{}, {"PEPAR1": 1, "A_AGE": 8}]) + result = self._derive(table) + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + assert result[column].dtype == np.int64 + + def test_parent_ids_use_person_id_not_the_line_number(self) -> None: + table = _person_table([{"person_id": 5_000}, {"PEPAR1": 1, "A_AGE": 8}]) + result = self._derive(table) + assert result["parent_1_id"].tolist() == [0, 5_000] + + def test_missing_person_id_is_named(self) -> None: + table = _person_table([{}, {"PEPAR1": 1, "A_AGE": 8}]).drop( + columns=["person_id"] + ) + with pytest.raises(SourceRuntimeError, match="person_id"): + self._derive(table) + + def test_repeated_household_line_is_refused(self) -> None: + table = _person_table( + [ + {"PH_SEQ": 1, "A_LINENO": 1}, + {"PH_SEQ": 1, "A_LINENO": 1}, + ] + ) + with pytest.raises(SourceRuntimeError, match="repeats line"): + self._derive(table) + def test_veterans_benefits_map_vet_val_and_clip_negatives(self) -> None: table = _person_table([{"VET_VAL": 8_400}, {"VET_VAL": -1}, {}]) result = self._derive(table) @@ -258,7 +375,7 @@ def test_unexpected_parameters_are_refused(self) -> None: class TestFrameIntegration: - def test_with_us_eligibility_inputs_writes_all_five_columns(self) -> None: + def test_with_us_eligibility_inputs_writes_every_output_column(self) -> None: frame = with_us_eligibility_inputs( _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD ) @@ -285,6 +402,52 @@ def test_parent_pointer_count_is_derived_once_then_cloned(self) -> None: assert set(person.loc[parent, support_clone_index_column("person")]) == {0, 1} assert person.loc[parent, "own_children_in_household"].tolist() == [1.0, 1.0] + def test_parent_ids_name_the_parent_row_person_id(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person") + assert person["parent_1_id"].dtype == np.int64 + assert ( + person["parent_1_id"].iloc[_CHILD_INDEX] + == person["person_id"].iloc[_PARENT_INDEX] + ) + # The first child names a second parent; every other row is unset. + assert person["parent_2_id"].iloc[_CHILD_INDEX] == person["person_id"].iloc[1] + assert person["parent_1_id"].iloc[_PARENT_INDEX] == 0 + assert person["parent_2_id"].iloc[_PARENT_INDEX] == 0 + + def test_parent_ids_are_remapped_onto_each_support_clone(self) -> None: + direct = with_us_eligibility_inputs( + _us_frame([{}, {"PEPAR1": 1, "A_AGE": 8}]), + seed=0, + time_period=TIME_PERIOD, + ) + expanded = clone_us_frame_for_puf_support(direct) + person = expanded.table("person") + clone_column = support_clone_index_column("person") + for clone_index in sorted(person[clone_column].unique()): + arm = person[person[clone_column] == clone_index] + child = arm[arm["A_LINENO"] == 2].iloc[0] + parent = arm[arm["A_LINENO"] == 1].iloc[0] + # The pointer must name this arm's parent, not the other arm's. + assert child["parent_1_id"] == parent["person_id"] + assert child["parent_1_id"] in set(arm["person_id"]) + assert parent["parent_1_id"] == 0 + + def test_absent_parent_sentinel_is_not_shifted_by_the_clone(self) -> None: + direct = with_us_eligibility_inputs( + _us_frame([{}, {"PEPAR1": 1, "A_AGE": 8}]), + seed=0, + time_period=TIME_PERIOD, + ) + expanded = clone_us_frame_for_puf_support(direct) + person = expanded.table("person") + parents = person["A_LINENO"].eq(1) + # 0 means "no parent"; shifting it would mint a valid-looking id. + assert person.loc[parents, "parent_1_id"].tolist() == [0, 0] + assert person.loc[parents, "parent_2_id"].tolist() == [0, 0] + def test_frame_with_signal_passes_through_untouched(self) -> None: derived = with_us_eligibility_inputs( _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD @@ -300,6 +463,8 @@ def test_constant_default_landmine_is_recomputed_from_raw(self) -> None: "is_blind": False, "is_full_time_college_student": False, "own_children_in_household": 0.0, + "parent_1_id": 0, + "parent_2_id": 0, "veterans_benefits": 0.0, } rows = [ @@ -351,6 +516,8 @@ def test_constant_default_surface_fails(self) -> None: "is_blind": False, "is_full_time_college_student": False, "own_children_in_household": 0.0, + "parent_1_id": 0, + "parent_2_id": 0, "veterans_benefits": 0.0, } frame = _us_frame([{**constants} for _ in range(4)]) @@ -358,6 +525,104 @@ def test_constant_default_surface_fails(self) -> None: assert not gate.passed assert any("constant" in failure for failure in gate.failures) + def test_parent_id_naming_no_person_fails(self) -> None: + # The failure a missed id remap produces: a pointer that survives + # into a frame whose person ids have moved. + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").copy() + person.loc[person.index[_CHILD_INDEX], "parent_1_id"] = 10_000_000 + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + assert not gate.passed + assert any("name no person" in failure for failure in gate.failures) + + def test_parent_id_outside_the_household_fails(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame( + [ + {"PH_SEQ": 1, "A_LINENO": 1}, + {"PH_SEQ": 1, "A_LINENO": 2, "PEPAR1": 1, "A_AGE": 8}, + {"PH_SEQ": 2, "A_LINENO": 1}, + {"PH_SEQ": 2, "A_LINENO": 2, "PEPAR1": 1, "A_AGE": 8}, + ] + ), + seed=0, + time_period=TIME_PERIOD, + ) + person = frame.table("person").copy() + # Point household 1's child at household 2's parent. + person.loc[person.index[1], "parent_1_id"] = int(person["person_id"].iloc[2]) + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + assert not gate.passed + assert any( + "outside the pointing person's household" in failure + for failure in gate.failures + ) + + def test_self_parenting_fails(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").copy() + index = person.index[_CHILD_INDEX] + person.loc[index, "parent_1_id"] = int(person.loc[index, "person_id"]) + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + assert not gate.passed + assert any("their own parent" in failure for failure in gate.failures) + + def test_count_disagreeing_with_the_pointers_fails(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").copy() + person.loc[person.index[_PARENT_INDEX], "own_children_in_household"] = 7.0 + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + assert not gate.passed + assert any( + "disagrees with the parent pointers" in failure for failure in gate.failures + ) + + def test_null_parent_id_fails(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").copy() + person["parent_1_id"] = person["parent_1_id"].astype("float64") + person.loc[person.index[_CHILD_INDEX], "parent_1_id"] = np.nan + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + assert not gate.passed + assert any("never as null" in failure for failure in gate.failures) + + def test_person_id_zero_cannot_be_named_and_is_reported(self) -> None: + # person_id is 0-based in production, so one person can never be + # named by a 0-means-unknown pointer. The gate must not fail on it, + # and the cost must be visible. + frame = with_us_eligibility_inputs( + _us_frame( + [ + {"person_id": 0}, + {"PEPAR1": 1, "A_AGE": 8, "person_id": 1}, + {"PEDISPHY": 1, "person_id": 2}, + {"VET_VAL": 9_000, "person_id": 3}, + {"A_HSCOL": 2, "A_FTPT": 1, "A_AGE": 20, "person_id": 4}, + ] + ), + seed=0, + time_period=TIME_PERIOD, + ) + person = frame.table("person") + assert person["own_children_in_household"].iloc[0] == 1.0 + assert person["parent_1_id"].iloc[1] == 0 + summary = us_eligibility_inputs_summary(frame) + assert summary["pointers_unnameable_at_person_id_zero"] == 1 + failures = [ + failure + for failure in us_eligibility_inputs_signal_gate(frame).failures + if "parent pointers" in failure + ] + assert failures == [] + def test_everyone_disabled_fails_the_share_band(self) -> None: rows = [{"PEDISPHY": 1} for _ in range(10)] frame = with_us_eligibility_inputs( From b651a9a781394207dba6e2baf98910356f847e1d Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 25 Sep 2026 20:12:30 -0400 Subject: [PATCH 3/7] Re-pin the US spec digests, add the changelog fragment, record the design The new eligibility-stage outputs move the source_stages asset digest, the bundle spec_sha256 (now d3469f47...), and, because puf_support is an attested kernel module, the seed-protocol, seed-map, resource-semantics, source-manifest and full-checkpoint identity digests. Re-pinned each at its observed value, bumped the field-usage counts by the four new authored fields, re-pinned the source_stages pointer claim and pointer inventory, and regenerated docs/evidence/spec-engine/us-f0-coverage.json with tools/spec_engine_coverage.py. The operator-ordering doc's eligibility row now says why the stage is pre-clone and CPS-scoped, and that its outputs name a person_id. Co-Authored-By: Claude Opus 5.5 --- changelog.d/884-parent-ids-export.added.md | 1 + docs/us-multispine-operator-ordering.md | 2 +- .../src/microcosm/build/spec_engine/field_usage.py | 8 ++++---- .../tests/test_us_bundle_core_contracts.py | 2 +- packages/microcosm-build/tests/test_us_spec_bundle.py | 2 +- tools/spec_engine_coverage.py | 2 +- 6 files changed, 9 insertions(+), 8 deletions(-) create mode 100644 changelog.d/884-parent-ids-export.added.md diff --git a/changelog.d/884-parent-ids-export.added.md b/changelog.d/884-parent-ids-export.added.md new file mode 100644 index 0000000000..de17eb27a1 --- /dev/null +++ b/changelog.d/884-parent-ids-export.added.md @@ -0,0 +1 @@ +US person input columns `parent_1_id` and `parent_2_id`: the CPS ASEC `PEPAR1`/`PEPAR2` pointers resolved to a co-resident parent's `person_id` (0 when unknown or absent), so downstream rules can ask whose child a person is rather than only how many children an adult has. diff --git a/docs/us-multispine-operator-ordering.md b/docs/us-multispine-operator-ordering.md index 8db0ba8b35..f3f62e66aa 100644 --- a/docs/us-multispine-operator-ordering.md +++ b/docs/us-multispine-operator-ordering.md @@ -1066,7 +1066,7 @@ runs: | Relationships | pre | Household-head status is a rent-recipient predictor. | | Medicare take-up | post | Rowwise carry/completion is clone-safe. | | Housing inputs | pre | Rent is drawn once per source household and then cloned unchanged. | -| Eligibility inputs | pre | Raw `PH_SEQ`/`A_LINENO` parent pointers would count every cloned child twice. | +| Eligibility inputs | pre | Raw `PH_SEQ`/`A_LINENO` parent pointers would count every cloned child twice, and `PH_SEQ` collides between the ASEC and ACS spines, so the stage runs on the CPS-evidence projection. Its `parent_1_id`/`parent_2_id` outputs name a `person_id`, so the support clone shifts them with the rest of the person id surface ([microcosm#884](https://github.com/PolicyEngine/microcosm/issues/884)). | | Pregnancy | post | Stable source-identity hashes share draws across clones. | | WIC | post | Remapped family grouping and source-identity draws are clone-safe. | | Housing assistance | post | The QRF intentionally replaces only the PUF support role. | diff --git a/packages/microcosm-build/src/microcosm/build/spec_engine/field_usage.py b/packages/microcosm-build/src/microcosm/build/spec_engine/field_usage.py index c70675d908..56f39195cf 100644 --- a/packages/microcosm-build/src/microcosm/build/spec_engine/field_usage.py +++ b/packages/microcosm-build/src/microcosm/build/spec_engine/field_usage.py @@ -26,9 +26,9 @@ ) from .schemas import load_schema_registry -EXPECTED_AUTHORED_FIELD_COUNT = 32_400 +EXPECTED_AUTHORED_FIELD_COUNT = 32_404 EXPECTED_RESOLVED_BINDING_FIELD_COUNT = 9_780 -EXPECTED_CONFIGURATION_FIELD_COUNT = 42_180 +EXPECTED_CONFIGURATION_FIELD_COUNT = 42_184 class FieldUsageError(AssertionError): @@ -485,8 +485,8 @@ def _path_inventory(rows: Sequence[tuple[str, object]]) -> tuple[int, str]: "d6782c5de5bbed1bdc6bf653c4a6d4aadcad4ccc72d35e1092e130fcb04680a3", ), "source_stages": ( - 1_712, - "a46d5e657d919db23ee7752d844bd8bbe59c15321dec07102837452d7994d8dd", + 1_716, + "3e808ed9f5a791c0dfb85be0c59adf803366bf6e8c7a9f5a58f92974e044b607", ), "spine_assembly_household_mass_shares": ( 2, diff --git a/packages/microcosm-build/tests/test_us_bundle_core_contracts.py b/packages/microcosm-build/tests/test_us_bundle_core_contracts.py index 66569c76ad..a57ec64f89 100644 --- a/packages/microcosm-build/tests/test_us_bundle_core_contracts.py +++ b/packages/microcosm-build/tests/test_us_bundle_core_contracts.py @@ -129,7 +129,7 @@ def test_source_surface_classification_is_complete() -> None: } assert normative["stage_asset"] == { "id": "source_stages", - "sha256": "9f4f983091fb18dea8f4524cc100be20f9fca2b006435461592f8edf53c9a5d8", + "sha256": "e75217c0db3075bf2664492167d31ce7ee82c2c5f6a70e3b67e8bb6f4ddd7ecf", } assert operational["stage_asset"] == { "path": "microcosm.build.us/source_stages.json" diff --git a/packages/microcosm-build/tests/test_us_spec_bundle.py b/packages/microcosm-build/tests/test_us_spec_bundle.py index f47ae69bd4..4e9249314d 100644 --- a/packages/microcosm-build/tests/test_us_spec_bundle.py +++ b/packages/microcosm-build/tests/test_us_spec_bundle.py @@ -191,7 +191,7 @@ def _load_generator_module(): LEGACY_COMPATIBILITY_SHA256 = { "source_stages.json": ( - "9f4f983091fb18dea8f4524cc100be20f9fca2b006435461592f8edf53c9a5d8" + "e75217c0db3075bf2664492167d31ce7ee82c2c5f6a70e3b67e8bb6f4ddd7ecf" ), "support_spine.json": ( "68f37dc6ae6e0cde7ebccb53f88dd4a800e63456f838fa214ff98d1db8d815be" diff --git a/tools/spec_engine_coverage.py b/tools/spec_engine_coverage.py index 0ea30f3c74..c066d573ce 100644 --- a/tools/spec_engine_coverage.py +++ b/tools/spec_engine_coverage.py @@ -41,7 +41,7 @@ REPORT_SCHEMA_VERSION = 3 EXPECTED_POINTER_INVENTORY_SHA256 = ( - "21b6fe2f6b96307f8ec7d20d4162ee2e0702f363bb28ebd3fa47a4cf7fc12ed4" + "8f655670b9794adae905ede64bdceba6395eeb8d642b6c5896d5c516f1f6a6a7" ) DEFAULT_REPORT_PATH = ( Path(__file__).resolve().parents[1] From 808ee8feb3c35febcd498615875476239bfc3d11 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 27 Sep 2026 02:19:31 -0400 Subject: [PATCH 4/7] Address review of #1032: clones, ACS rows, existing frames, int32 consumers Independent review (2026-09-25) returned request-changes with seven findings and the rebased branch failed CI in four lanes. Changes: - Capital-gains tail clone shifts parent ids with person_id, keeping the 0 sentinel (_shift_person_references); the preassembled PUF tax-detail rows, which never had a pointer, carry 0 rather than null through the support clone's reference remap. - The release gate no longer demands own_children_in_household equal the pointers naming a person. ACS rows impute the count with pointers declared unknown, and a clone moves the unnameable id-0 parent, so the gate asserts pointed_at_by <= count; the exact equality is asserted where the relationships are measured, when the stage derives them. - A frame that already carries the pre-#884 outputs with signal (a --base-h5 input) gets only parent_1_id/parent_2_id appended, resolved from the raw pointers when present and 0 otherwise; no other column is recomputed. Parent ids leave the non-constant release set: all-zero means unknown. - Co-residence compares exact int64 household ids (float64 conflated ids past 2**53), and a new check requires every person_id in a household carrying a parent link to fit PolicyEngine-US's int32 storage. - Pool resume ignores a transferred/simulated checkpoint whose recorded person schema lacks an eligibility output, since the byte-stable legacy identity cannot see new outputs. - The input-mass exclusion moves into the shared helper as PERSON_REFERENCE_COLUMNS, restoring main's exact US alias. - Re-pinned at observed values after the rebase and these edits: spec_sha256 1cf1f44d..., the five inventory digests, field-usage mode/effect counts, and the loader golden vector (attested kernel sources moved). Co-Authored-By: Claude Opus 5.5 --- docs/evidence/spec-engine/us-f0-coverage.json | 48 ++-- .../src/microcosm/build/input_mass.py | 16 +- .../build/spec_engine/inventory_coverage.py | 10 +- .../build/us_runtime/eligibility_inputs.py | 229 ++++++++++++++---- .../microcosm/build/us_runtime/input_mass.py | 42 +--- .../us_runtime/puf_capital_gains_tail.py | 19 ++ .../microcosm/build/us_runtime/puf_support.py | 13 +- .../tests/test_spec_engine_coverage_tool.py | 10 +- .../tests/test_spec_engine_field_usage.py | 10 +- .../tests/test_spec_engine_loader.py | 2 +- .../tests/test_us_eligibility_inputs.py | 114 ++++++++- .../tests/test_us_input_mass.py | 20 ++ .../tests/test_us_multispine_pool.py | 17 +- .../tests/test_us_multispine_pool_tool.py | 45 +++- .../tests/test_us_puf_capital_gains_tail.py | 32 +++ tools/build_us_multispine_pool.py | 45 ++++ 16 files changed, 528 insertions(+), 144 deletions(-) diff --git a/docs/evidence/spec-engine/us-f0-coverage.json b/docs/evidence/spec-engine/us-f0-coverage.json index 75804e57cc..c153d52291 100644 --- a/docs/evidence/spec-engine/us-f0-coverage.json +++ b/docs/evidence/spec-engine/us-f0-coverage.json @@ -4,9 +4,9 @@ "version": 1 }, "country": "us", - "documentation_sha256": "911b21d13cc214601cfb537c538fed2851e5166e8f72de5ebfce5f2918af43d3", + "documentation_sha256": "0931cbd3aae2abcd8c072b78eba40c1c006b1bf7fa23c6a6681f9a895e47070e", "field_usage": { - "authored_normative_field_count": 32400, + "authored_normative_field_count": 32404, "claim_count": 49, "claims": [ { @@ -514,8 +514,8 @@ ], "mode": "legacy_behavior", "pointer_class": "all", - "pointer_count": 1712, - "pointer_sha256": "a46d5e657d919db23ee7752d844bd8bbe59c15321dec07102837452d7994d8dd", + "pointer_count": 1716, + "pointer_sha256": "3e808ed9f5a791c0dfb85be0c59adf803366bf6e8c7a9f5a58f92974e044b607", "rationale": null, "relative_sink_prefix": "/source_manifest/stages", "source_prefix": "/authored/spec~1sources.yaml/stages", @@ -772,20 +772,20 @@ "verifier": "vintages" } ], - "configuration_field_count": 42180, - "consumed_field_count": 42180, + "configuration_field_count": 42184, + "consumed_field_count": 42184, "generation0_effect_counts": { - "legacy_behavior": 38494, + "legacy_behavior": 38498, "no_generation0_effect": 3686 }, "mode_counts": { "compiler_semantic": 27717, "front_end_validation": 354, "identity_only": 103, - "legacy_behavior": 14006 + "legacy_behavior": 14010 }, "multiple_primary_use_field_count": 0, - "pointer_inventory_sha256": "21b6fe2f6b96307f8ec7d20d4162ee2e0702f363bb28ebd3fa47a4cf7fc12ed4", + "pointer_inventory_sha256": "8f655670b9794adae905ede64bdceba6395eeb8d642b6c5896d5c516f1f6a6a7", "resolved_binding_field_count": 9780, "unused_field_count": 0 }, @@ -1361,12 +1361,12 @@ ], "expected": { "producer_count": 38, - "sha256": "6abbfad73014ce3bac9fbd549f16e4a5e645a3b0fbd450b56e186876f3f0dd95" + "sha256": "cc34ba11288d01145fc2dc1a669df3f7c4ef2019d73afb02dd8eeee852b16d20" }, "failures": [], "observed": { "producer_count": 38, - "sha256": "6abbfad73014ce3bac9fbd549f16e4a5e645a3b0fbd450b56e186876f3f0dd95" + "sha256": "cc34ba11288d01145fc2dc1a669df3f7c4ef2019d73afb02dd8eeee852b16d20" }, "status": "covered" }, @@ -1656,13 +1656,13 @@ "compiler_ir.node_slices" ], "expected": { - "map_sha256": "dad44808366198b8826823b837c7b592c9ecee2299adbd9f8a20251dd6d0b00a", - "protocol_sha256": "5f6d3b527060bc6f15f1871c64a42994d4882ee5962f3350ea8a71a83ffcd6bf" + "map_sha256": "3cf10a524310ea0ada18ddc65cfa585d1d0652d23e7d7a180c050ac686224e28", + "protocol_sha256": "91989ed5e366e19598f86335b2e66723dcb2b6481a78d0dbeb83183a1aa09bd4" }, "failures": [], "observed": { - "map_sha256": "dad44808366198b8826823b837c7b592c9ecee2299adbd9f8a20251dd6d0b00a", - "protocol_sha256": "5f6d3b527060bc6f15f1871c64a42994d4882ee5962f3350ea8a71a83ffcd6bf" + "map_sha256": "3cf10a524310ea0ada18ddc65cfa585d1d0652d23e7d7a180c050ac686224e28", + "protocol_sha256": "91989ed5e366e19598f86335b2e66723dcb2b6481a78d0dbeb83183a1aa09bd4" }, "status": "covered" }, @@ -1677,7 +1677,7 @@ "compiler_ir.seed_stream_map" ], "expected": { - "implementation_sha256": "5f6d3b527060bc6f15f1871c64a42994d4882ee5962f3350ea8a71a83ffcd6bf", + "implementation_sha256": "91989ed5e366e19598f86335b2e66723dcb2b6481a78d0dbeb83183a1aa09bd4", "protocol": "legacy-v1", "streams": [ "build_model", @@ -1698,7 +1698,7 @@ }, "failures": [], "observed": { - "implementation_sha256": "5f6d3b527060bc6f15f1871c64a42994d4882ee5962f3350ea8a71a83ffcd6bf", + "implementation_sha256": "91989ed5e366e19598f86335b2e66723dcb2b6481a78d0dbeb83183a1aa09bd4", "protocol": "legacy-v1", "streams": [ "build_model", @@ -1775,12 +1775,12 @@ "legacy_adapter.imputation.late_producer_resource_semantics" ], "expected": { - "sha256": "a87278445d928ea20e15935a8cc07312d6e819e34409b04e8c814f17910d828f", + "sha256": "f4f8bfeb79268a4043a843b47cec1d1c0e58e9f5dd6bc23417ada95446dd7b8e", "stages": 37 }, "failures": [], "observed": { - "sha256": "a87278445d928ea20e15935a8cc07312d6e819e34409b04e8c814f17910d828f", + "sha256": "f4f8bfeb79268a4043a843b47cec1d1c0e58e9f5dd6bc23417ada95446dd7b8e", "stages": 38 }, "status": "covered" @@ -1887,7 +1887,7 @@ "alpha", "zeta" ], - "sha256": "b1d5c65254b51897b27f0f5e44e344781ce9e01e305aaf1bfc042cc32bc417dc" + "sha256": "b7a336af7cb300905d09ad9a33aa1a624f878c2471352d4e252fa69db55aee17" }, "failures": [], "observed": { @@ -1911,7 +1911,7 @@ "alpha", "zeta" ], - "sha256": "b1d5c65254b51897b27f0f5e44e344781ce9e01e305aaf1bfc042cc32bc417dc" + "sha256": "b7a336af7cb300905d09ad9a33aa1a624f878c2471352d4e252fa69db55aee17" }, "status": "covered" }, @@ -1975,7 +1975,7 @@ "take_up_contract", "us_qbi_reconciliation_contract" ], - "sha256": "5b65545ed6f351b30e94ab1405e13d2c7f692ae0c8d4959badcfe9bb9d52d455" + "sha256": "61a5484eab9fae1d6651b3ae15f1793d4934ea2f6ed2c08d05716b745c221014" }, "status": "covered" }, @@ -2599,7 +2599,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "f830589331dad583398485de5724dace0714d8526fb63b37f81cd783b1cb8f30" + "spec_sha256": "1cf1f44d9a11458f59ea7d75dce68303816cab3e66668a6c6f6ecfe62d56f8dc" } }, "report_schema_version": 3, @@ -2609,7 +2609,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "f830589331dad583398485de5724dace0714d8526fb63b37f81cd783b1cb8f30" + "spec_sha256": "1cf1f44d9a11458f59ea7d75dce68303816cab3e66668a6c6f6ecfe62d56f8dc" }, "status": "pass" } diff --git a/packages/microcosm-build/src/microcosm/build/input_mass.py b/packages/microcosm-build/src/microcosm/build/input_mass.py index 3bde6f9149..7d8ec68e8f 100644 --- a/packages/microcosm-build/src/microcosm/build/input_mass.py +++ b/packages/microcosm-build/src/microcosm/build/input_mass.py @@ -22,7 +22,15 @@ from microcosm.frame import Frame -__all__ = ["input_mass_totals"] +__all__ = ["PERSON_REFERENCE_COLUMNS", "input_mass_totals"] + +#: Person columns whose values name another person (a co-resident parent's +#: ``person_id``, ``0`` = unknown; microcosm#884). Like entity ids they are +#: labels, not measured quantities: their weighted sum means nothing and moves +#: whenever person ids are renumbered, so a parity comparison against a release +#: on a different id space would fail for no reason. They are skipped as +#: structural in every schema that carries them. +PERSON_REFERENCE_COLUMNS: frozenset[str] = frozenset({"parent_1_id", "parent_2_id"}) def input_mass_totals( @@ -35,8 +43,8 @@ def input_mass_totals( 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. + ``True`` mass. String/enum columns and structural columns (entity ids, + person membership ids, and :data:`PERSON_REFERENCE_COLUMNS`) are skipped. Args: frame: A microcosm frame in any country schema. @@ -56,7 +64,7 @@ def input_mass_totals( """ schema = frame.schema - structural = {schema.person_id_column} + structural = {schema.person_id_column, *PERSON_REFERENCE_COLUMNS} for group in schema.group_entities: structural.add(schema.id_column(group)) structural.add(schema.membership_column(group)) diff --git a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py index e3e5769c92..6c30877dce 100644 --- a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py +++ b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py @@ -350,18 +350,18 @@ "acs_person_predictors": "878c788a6f037d7aca12b3586ea034eff04f3034ffa11935a736493042551f25", "authority": "9d4a9672a0f03039b1fe874b9fe21ed575be0d29f14afc396d03cdf5c809bdd2", "early_families": "4aa9f736fd76e83955477ad1667e58f48f264783f05bdc7f0102cd32d61323bd", - "full_checkpoint": "b1d5c65254b51897b27f0f5e44e344781ce9e01e305aaf1bfc042cc32bc417dc", + "full_checkpoint": "b7a336af7cb300905d09ad9a33aa1a624f878c2471352d4e252fa69db55aee17", "gap_fill_schedule": "1c31f9868f7884347cc19cf1ff65da43f950b9114941a715bab168246db414a7", "graph_nodes": "40cd51ffdfe2e9d9d08d48c08e8ded9de1e4b134783bab05c4abc6ad5c72ca1e", "geography_assignment": "f49425ca8734ac559c73cf44f6458d86d3162a48956b98a27e6e758959361585", "late_families": "d91f9ff0eb52f43e7b6eed3d5c58c37abe1620c3a11021da15dae9c10e16d382", - "late_resource_semantics": "6abbfad73014ce3bac9fbd549f16e4a5e645a3b0fbd450b56e186876f3f0dd95", + "late_resource_semantics": "cc34ba11288d01145fc2dc1a669df3f7c4ef2019d73afb02dd8eeee852b16d20", "late_schedule": "e59c019d3d454eac99ac0ac209b6c5b6faaf9bdfcaeee18c36a25be19bf7da2f", "ownership": "5f64f0aac49e2313177564f71876bffc8c81b3ded4df701e70930e60e9c98356", "primary_tuples": "fdf23da429f8c501e198a2f5b719763d523a41565c2153fdf17b0344f21bcd7c", - "seed_map": "dad44808366198b8826823b837c7b592c9ecee2299adbd9f8a20251dd6d0b00a", - "seed_protocol": "5f6d3b527060bc6f15f1871c64a42994d4882ee5962f3350ea8a71a83ffcd6bf", - "source_manifest": "a87278445d928ea20e15935a8cc07312d6e819e34409b04e8c814f17910d828f", + "seed_map": "3cf10a524310ea0ada18ddc65cfa585d1d0652d23e7d7a180c050ac686224e28", + "seed_protocol": "91989ed5e366e19598f86335b2e66723dcb2b6481a78d0dbeb83183a1aa09bd4", + "source_manifest": "f4f8bfeb79268a4043a843b47cec1d1c0e58e9f5dd6bc23417ada95446dd7b8e", "take_up": "9522ce40f7dea569312dd7a7beb474e5a5afadd2f1e10cb5876534c3ef623d35", "tail": "ac92829c88a1a4fb6460d61190918d5d99c6c377fc8dd8f62f02b332d09bf59c", } diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py index 9b65f293ef..36a50e11a2 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py @@ -82,9 +82,14 @@ across pooled ASEC years. The PUF support clone that follows renumbers ``person_id``; :func:`microcosm.build.us_runtime.puf_support.clone_us_frame_for_puf_support` therefore shifts these two columns with the rest of the person id -surface, preserving ``0``. The gate below re-checks co-residence and the -``own_children_in_household`` identity on whatever frame it is handed, -so a missed remap fails closed instead of exporting a stale pointer. +surface, preserving ``0``. The gate below re-checks co-residence, the +count bound, and int32 safety on whatever frame it is handed, so a missed +remap fails closed instead of exporting a stale pointer. + +Existing-frame note: a frame that already carries the pre-#884 outputs +with signal (a ``--base-h5`` release input) gets only the two id columns +appended — resolved from the raw pointers if it still has them, ``0`` +otherwise — and no other column is recomputed. """ from __future__ import annotations @@ -143,11 +148,28 @@ "veterans_benefits", ) +#: The columns this stage owned before the parent ids (microcosm#884). +_LEGACY_OUTPUT_COLUMNS: tuple[str, ...] = tuple( + column + for column in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS + if column not in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS +) + #: Release gates require these person columns to carry signal (≥2 values). +#: The parent ids are not among them: all-zero parent ids mean "unknown", +#: which PolicyEngine-US reads as "use the count-based proxy", so a frame +#: that has no parent pointers to resolve (an ACS-only or legacy base) is +#: valid with zeros. Their integrity is asserted by +#: :func:`_parent_id_invariant_failures` instead. US_ELIGIBILITY_INPUTS_NONCONSTANT_PERSON_COLUMNS: tuple[str, ...] = ( - US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS + _LEGACY_OUTPUT_COLUMNS ) +#: PolicyEngine-US stores ``int`` variables as int32, and matches a parent id +#: against the household members' ``person_id``. Every id in a household that +#: carries a parent link must therefore survive an int32 cast unchanged. +_INT32_MAX = 2**31 - 1 + #: The six ASEC disability-difficulty items; 1 means "yes". _DISABILITY_DIFFICULTY_COLUMNS: tuple[str, ...] = ( "PEDISDRS", @@ -163,6 +185,13 @@ #: :data:`US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS` is load-bearing. _PARENT_POINTER_COLUMNS: tuple[str, ...] = ("PEPAR1", "PEPAR2") +#: Raw columns the parent-id resolution needs on top of ``person_id``. +_PARENT_ID_SOURCE_COLUMNS: tuple[str, ...] = ( + *_PARENT_POINTER_COLUMNS, + "PH_SEQ", + "A_LINENO", +) + #: The frame's person identifier. Not an ASEC source column: it is the #: structural id the stage aligns its output on, so it is already present. _PERSON_ID_COLUMN = "person_id" @@ -437,11 +466,20 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> if frame.schema != US_SCHEMA: raise ValueError("US eligibility inputs require the US schema.") person = frame.table("person") - have_all = all( - column in person.columns for column in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS + have_legacy = all( + column in person.columns for column in _LEGACY_OUTPUT_COLUMNS + ) and _disabled_carries_signal(person) + have_parent_ids = all( + column in person.columns for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS ) - if have_all and _disabled_carries_signal(person): + if have_legacy and have_parent_ids: return frame + if have_legacy: + # A frame built before the parent ids existed (e.g. a --base-h5 + # release input) already carries trusted eligibility inputs. Add the + # two id columns and touch nothing else: recomputing the whole stage + # here would overwrite carried values with raw-column re-derivations. + return _with_parent_ids_appended(frame) stage_person = person.copy(deep=True) stage_person[_PERSON_WEIGHT_COLUMN] = frame.resolve_weights("person").values @@ -460,6 +498,19 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> f"US eligibility-inputs stage output does not cover every " f"person for {column!r}." ) + # Where the relationships are measured, the count and the ids come from + # the same pointers, so they must agree exactly. Downstream frames (ACS + # rows with imputed counts and zero pointers, clones of the person who + # held id 0) only satisfy the weaker gate identity. + derived = aligned.reset_index() + mismatched = _count_identity_mismatches(derived) + if mismatched: + raise SourceRuntimeError( + "US eligibility-inputs derivation produced own_children_in_household " + "counts that disagree with the resolved parent ids for " + f"{mismatched} person(s); both read the same PEPAR pointers, so this " + "is a derivation defect." + ) tables = {entity: frame.table(entity).copy() for entity in frame.entities} bool_columns = ("is_disabled", "is_blind", "is_full_time_college_student") @@ -482,6 +533,75 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> ) +def _with_parent_ids_appended(frame: Frame) -> Frame: + """Add ``parent_1_id``/``parent_2_id`` without touching any other column. + + The ids are resolved from the raw ASEC pointers when the person table + still carries them (``PEPAR1``/``PEPAR2``, ``PH_SEQ``, ``A_LINENO``) and + are ``0`` ("unknown") otherwise, which PolicyEngine-US reads as "use the + count-based proxy" — the behavior such a frame had before this column + existed. + """ + + person = frame.table("person") + if all(column in person.columns for column in _PARENT_ID_SOURCE_COLUMNS): + resolved = _parent_person_ids(person) + else: + resolved = { + column: np.zeros(len(person), dtype=np.int64) + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + } + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + for column, values in resolved.items(): + tables["person"][column] = np.asarray(values, dtype=np.int64) + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + metadata=frame.metadata, + ) + + +def _pointed_at_counts( + person_id: np.ndarray, resolved: dict[str, np.ndarray] +) -> np.ndarray: + """How many people's parent ids name each person (by row).""" + + position_by_id = { + identifier: index for index, identifier in enumerate(person_id.tolist()) + } + counts = np.zeros(len(person_id), dtype=np.float64) + for column_ids in resolved.values(): + for value in column_ids[column_ids > 0].tolist(): + position = position_by_id.get(int(value)) + if position is not None: + counts[position] += 1.0 + return counts + + +def _count_identity_mismatches(person: pd.DataFrame) -> int: + """Rows whose child count differs from the number of ids naming them. + + Exact on a freshly derived CPS frame, except for the one person holding + ``person_id`` 0, whom a 0-means-unknown pointer cannot name. + """ + + person_id = pd.to_numeric(person[_PERSON_ID_COLUMN], errors="raise").to_numpy( + dtype=np.int64 + ) + resolved = { + column: pd.to_numeric(person[column], errors="raise").to_numpy(dtype=np.int64) + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + } + observed = pd.to_numeric( + person["own_children_in_household"], errors="coerce" + ).to_numpy(dtype=np.float64) + expected = _pointed_at_counts(person_id, resolved) + return int(((observed != expected) & (person_id != 0)).sum()) + + def us_eligibility_inputs_summary(frame: Frame) -> dict[str, object]: """Weighted eligibility-share summary for gates and release manifests.""" @@ -617,60 +737,70 @@ def _parent_id_invariant_failures(person: pd.DataFrame) -> list[str]: if household.isna().any(): failures.append(f"{household_column!r} carries non-numeric values.") else: + # Exact integers: household ids can run past 2**53, where a float + # comparison would call two different households equal. + household_ids = household.to_numpy(dtype=np.int64) household_by_id = dict( - zip(person_id.tolist(), household.to_numpy().tolist(), strict=True) + zip(person_id.tolist(), household_ids.tolist(), strict=True) ) + linked_households: set[int] = set() for column, column_ids in resolved.items(): - pointed = column_ids > 0 - if not pointed.any(): - continue - parent_household = np.array( - [ - household_by_id.get(int(value), np.nan) - for value in column_ids[pointed] - ], - dtype=np.float64, - ) - offsite = parent_household != household.to_numpy()[pointed] - if offsite.any(): + pointed = np.flatnonzero(column_ids > 0) + offsite = 0 + for row in pointed.tolist(): + parent_household = household_by_id.get(int(column_ids[row])) + # A name no person carries is reported above as unknown. + if parent_household is None: + continue + if parent_household != int(household_ids[row]): + offsite += 1 + linked_households.add(int(household_ids[row])) + if offsite: failures.append( - f"{column}: {int(offsite.sum())} pointer(s) name a " - "person outside the pointing person's household; a " - "parent id is only ever a co-resident." + f"{column}: {offsite} pointer(s) name a person outside " + "the pointing person's household; a parent id is only " + "ever a co-resident." + ) + if linked_households: + in_linked = np.isin(household_ids, list(linked_households)) + linked_ids = person_id[in_linked] + out_of_range = (linked_ids < 0) | (linked_ids > _INT32_MAX) + if out_of_range.any(): + failures.append( + f"{int(out_of_range.sum())} person_id(s) in households " + f"that carry a parent link fall outside [0, {_INT32_MAX}]. " + "PolicyEngine-US stores parent and person ids as int32, " + "so such ids would wrap and could name the wrong member " + "or collide with the 0 sentinel; renumber person ids " + "(and both parent-id columns with them) before export." ) - expected = np.zeros(len(person), dtype=np.float64) - position_by_id = { - identifier: index for index, identifier in enumerate(person_id.tolist()) - } - for column_ids in resolved.values(): - for value in column_ids[column_ids > 0].tolist(): - position = position_by_id.get(int(value)) - if position is not None: - expected[position] += 1.0 + # Every id naming a person counts toward that person's own children, so + # the count can never be smaller. It can be larger: ACS rows carry an + # imputed count with pointers declared unknown, and the person who held + # id 0 (unnameable) keeps their count after a clone renumbers them. The + # exact equality holds where relationships are measured and is checked + # when the stage derives them. + pointed_at_by = _pointed_at_counts(person_id, resolved) observed = ( pd.to_numeric(person["own_children_in_household"], errors="coerce") .fillna(-1.0) .to_numpy(dtype=np.float64) ) - # The person holding id 0 cannot be named by a 0-means-unknown pointer, - # so their count is unverifiable here by construction. It is reported as - # ``pointers_unnameable_at_person_id_zero`` rather than silently passing. - mismatched = (observed != expected) & (person_id != 0) - if mismatched.any(): + overcounted = pointed_at_by > observed + if overcounted.any(): examples = [ { "person_id": int(person_id[index]), "own_children_in_household": float(observed[index]), - "pointed_at_by": float(expected[index]), + "pointed_at_by": float(pointed_at_by[index]), } - for index in np.flatnonzero(mismatched)[:5] + for index in np.flatnonzero(overcounted)[:5] ] failures.append( - f"own_children_in_household disagrees with the parent pointers " - f"for {int(mismatched.sum())} person(s): the count must equal the " - f"number of people whose parent id names them. Examples: " - f"{examples}." + f"{int(overcounted.sum())} person(s) are named as a parent by more " + "people than own_children_in_household counts; a parent id can " + f"only name a person's own co-resident child. Examples: {examples}." ) return failures @@ -685,11 +815,12 @@ def us_eligibility_inputs_signal_gate(frame: Frame) -> GateResult: It also asserts the parent-id identities from :func:`_parent_id_invariant_failures`: every resolved pointer names a - co-resident person, nobody parents themselves, and - ``own_children_in_household`` equals the number of people whose parent - ids name that person (microcosm#884). Those are exact identities, so - they fail closed on a stale or donated id rather than on an unusual - population. + co-resident person, nobody parents themselves, nobody is named as a + parent by more people than ``own_children_in_household`` counts, and + every id in a linked household survives PolicyEngine-US's int32 storage + (microcosm#884). Those hold for every valid frame, so they fail closed + on a stale or donated id rather than on an unusual population. The + parent ids themselves may be all zero ("unknown"). """ person = frame.table("person") @@ -709,6 +840,8 @@ def us_eligibility_inputs_signal_gate(frame: Frame) -> GateResult: summary = us_eligibility_inputs_summary(frame) for column, count in summary["unique_counts"].items(): + if column not in US_ELIGIBILITY_INPUTS_NONCONSTANT_PERSON_COLUMNS: + continue if count < 2: failures.append( f"{column}: constant column (one observed value) — the " diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py b/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py index 79b780ebdf..2aa0ba66ff 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py @@ -1,43 +1,15 @@ -"""US weighted per-column input-mass helper. +"""US alias for the shared weighted per-column input-mass helper. -The computation was promoted to :mod:`microcosm.build.input_mass` when the UK -terminal battery adopted the same #278 input-mass parity gate (#609); it is -schema-driven and carries no country logic. This module remains so existing US -call sites keep their import path, and now applies the one US-specific -subtraction the shared helper cannot make: the person-referencing id columns. - -``input_mass_totals`` treats every numeric non-structural column as a measured -layer and reports its weighted sum, but its structural set is only the entity -ids and memberships. ``parent_1_id`` is neither: it is an arbitrary integer -label whose weighted sum means nothing, and it moves whenever person ids are -renumbered — so comparing it against a reference release built on a different -id space would fail the parity gate for no reason (microcosm#884). +The implementation was promoted to :mod:`microcosm.build.input_mass` when the +UK terminal battery adopted the same #278 input-mass parity gate (#609); the +computation is schema-driven and never carried US-specific logic. This module +remains so existing US call sites keep their import path. """ from __future__ import annotations -from collections.abc import Iterable, Mapping - from microcosm.build.input_mass import input_mass_totals -from microcosm.build.us_runtime.puf_support import US_PERSON_REFERENCE_ID_COLUMNS -from microcosm.frame import Frame - -__all__ = ["US_INPUT_MASS_EXCLUDED_COLUMNS", "us_input_mass_totals"] - -#: Columns whose weighted sum is not a mass. See the module docstring. -US_INPUT_MASS_EXCLUDED_COLUMNS: frozenset[str] = frozenset( - US_PERSON_REFERENCE_ID_COLUMNS -) - -def us_input_mass_totals( - frame: Frame, - *, - columns: Iterable[str] | None = None, -) -> Mapping[str, float]: - """Weighted per-column totals, less the person-referencing id columns.""" +__all__ = ["us_input_mass_totals"] - totals = dict(input_mass_totals(frame, columns=columns)) - for column in US_INPUT_MASS_EXCLUDED_COLUMNS: - totals.pop(column, None) - return totals +us_input_mass_totals = input_mass_totals diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/puf_capital_gains_tail.py b/packages/microcosm-build/src/microcosm/build/us_runtime/puf_capital_gains_tail.py index 995795bf51..07fb18c9ec 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/puf_capital_gains_tail.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/puf_capital_gains_tail.py @@ -27,6 +27,7 @@ from microcosm.build.us_runtime.puf_support import ( PUF_DONOR_SOURCE_ADJUSTED_GROSS_INCOME_COLUMN, PUF_TAX_DETAIL_DEFAULT_TAX_UNIT_OUTPUTS, + US_PERSON_REFERENCE_ID_COLUMNS, ) from microcosm.build.us_runtime.support_provenance import ( PUF_TAX_DETAIL_CLONE_INDEX, @@ -1592,6 +1593,23 @@ def _assign_tail_donors( return result +def _shift_person_references(person: pd.DataFrame, id_multiplier: int) -> None: + """Shift parent-id columns with ``person_id`` in place, keeping ``0``. + + A parent id names a ``person_id``, so a clone that shifts ``person_id`` + must shift it too or the clone's child names the source arm's parent; + ``0`` means "unknown" and must not become a valid-looking id + (microcosm#884). + """ + + for reference in US_PERSON_REFERENCE_ID_COLUMNS: + if reference in person.columns: + values = person[reference].to_numpy(dtype=np.int64) + person[reference] = np.where( + values == 0, 0, values + int(id_multiplier) + ).astype(np.int64) + + def _clone_and_transfer( frame: Frame, assignments: pd.DataFrame, @@ -1662,6 +1680,7 @@ def _clone_and_transfer( clone[membership] = ( clone[membership].to_numpy(dtype=np.int64) + id_multiplier ) + _shift_person_references(clone, id_multiplier) clone[clone_index] = 2 combined = pd.concat([table, clone], ignore_index=True) if combined[primary].duplicated().any(): diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py b/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py index f0839ea840..6d609532a9 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/puf_support.py @@ -2314,16 +2314,21 @@ def _remap_person_reference_ids( ``0`` means "no such person" on these columns, so it must not be shifted into ``clone_index * id_multiplier`` — a valid-looking id naming a - stranger. + stranger. A missing value (a row, such as a preassembled PUF tax-detail + person, that never had a parent pointer to resolve) is that same + "unknown" and is written as ``0``. """ + known = pd.to_numeric(pd.Series(values), errors="coerce").fillna(0) shifted = _remap_ids( - values, + known.to_numpy(), clone_index=clone_index, id_multiplier=id_multiplier, ) - original = _validated_integral_ids(values, label="PUF support person references") - return np.where(original == 0, 0, shifted) + original = _validated_integral_ids( + known.to_numpy(), label="PUF support person references" + ) + return np.where(original == 0, 0, shifted).astype(np.int64) def _validate_channels(channels: Sequence[str]) -> tuple[str, ...]: diff --git a/packages/microcosm-build/tests/test_spec_engine_coverage_tool.py b/packages/microcosm-build/tests/test_spec_engine_coverage_tool.py index b6f65a195a..972ae5da7f 100644 --- a/packages/microcosm-build/tests/test_spec_engine_coverage_tool.py +++ b/packages/microcosm-build/tests/test_spec_engine_coverage_tool.py @@ -88,21 +88,21 @@ def test_us_coverage_is_exact_complete_and_honest( assert_coverage_complete(coverage_report) assert coverage_report["status"] == "pass" fields = coverage_report["field_usage"] - assert fields["configuration_field_count"] == 42_180 - assert fields["authored_normative_field_count"] == 32_400 + assert fields["configuration_field_count"] == 42_184 + assert fields["authored_normative_field_count"] == 32_404 assert fields["resolved_binding_field_count"] == 9_780 - assert fields["consumed_field_count"] == 42_180 + assert fields["consumed_field_count"] == 42_184 assert fields["unused_field_count"] == 0 assert fields["multiple_primary_use_field_count"] == 0 assert fields["claim_count"] == 49 assert fields["mode_counts"] == { - "legacy_behavior": 14_006, + "legacy_behavior": 14_010, "compiler_semantic": 27_717, "front_end_validation": 354, "identity_only": 103, } assert fields["generation0_effect_counts"] == { - "legacy_behavior": 38_494, + "legacy_behavior": 38_498, "no_generation0_effect": 3_686, } diff --git a/packages/microcosm-build/tests/test_spec_engine_field_usage.py b/packages/microcosm-build/tests/test_spec_engine_field_usage.py index 4d468b5135..c5dbbe3032 100644 --- a/packages/microcosm-build/tests/test_spec_engine_field_usage.py +++ b/packages/microcosm-build/tests/test_spec_engine_field_usage.py @@ -92,22 +92,22 @@ def _mutated_bundle( def test_exact_complete_ledger_has_one_primary_mode_per_pointer(field_ledger) -> None: - assert len(field_ledger.fields) == EXPECTED_CONFIGURATION_FIELD_COUNT == 42_180 + assert len(field_ledger.fields) == EXPECTED_CONFIGURATION_FIELD_COUNT == 42_184 assert field_ledger.source_counts == { - "authored": 32_400, + "authored": 32_404, "resolved_bindings": 9_780, } assert field_ledger.mode_counts == { - "legacy_behavior": 14_006, + "legacy_behavior": 14_010, "compiler_semantic": 27_717, "front_end_validation": 354, "identity_only": 103, } assert field_ledger.generation0_effect_counts == { - "legacy_behavior": 38_494, + "legacy_behavior": 38_498, "no_generation0_effect": 3_686, } - assert len({field.pointer for field in field_ledger.fields}) == 42_180 + assert len({field.pointer for field in field_ledger.fields}) == 42_184 def test_eligibility_concepts_are_validation_not_generation0_behavior( diff --git a/packages/microcosm-build/tests/test_spec_engine_loader.py b/packages/microcosm-build/tests/test_spec_engine_loader.py index 14488eed65..679d91d7c1 100644 --- a/packages/microcosm-build/tests/test_spec_engine_loader.py +++ b/packages/microcosm-build/tests/test_spec_engine_loader.py @@ -236,7 +236,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 == ( - "dff847698ad9db6347b3f5dd4a80932b3716439f490e5bfcd4488aa8ef297d59" + "6af478ff88b3f65b5f06bfa4345f601fe887bfb8b5a6809fc3fc3127a4733922" ) second_root = _rich_minimal(tmp_path / "xy", note="second", store="local:b") diff --git a/packages/microcosm-build/tests/test_us_eligibility_inputs.py b/packages/microcosm-build/tests/test_us_eligibility_inputs.py index 6f29146b4f..f1c02a89ef 100644 --- a/packages/microcosm-build/tests/test_us_eligibility_inputs.py +++ b/packages/microcosm-build/tests/test_us_eligibility_inputs.py @@ -571,17 +571,26 @@ def test_self_parenting_fails(self) -> None: assert not gate.passed assert any("their own parent" in failure for failure in gate.failures) - def test_count_disagreeing_with_the_pointers_fails(self) -> None: + def test_count_below_the_pointers_naming_a_person_fails(self) -> None: frame = with_us_eligibility_inputs( _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD ) person = frame.table("person").copy() - person.loc[person.index[_PARENT_INDEX], "own_children_in_household"] = 7.0 + person.loc[person.index[_PARENT_INDEX], "own_children_in_household"] = 0.0 gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) assert not gate.passed - assert any( - "disagrees with the parent pointers" in failure for failure in gate.failures + assert any("named as a parent by more" in failure for failure in gate.failures) + + def test_count_above_the_pointers_passes(self) -> None: + # ACS rows carry an imputed count with pointers declared unknown, and + # a cloned id-0 parent keeps a count no pointer can reach. + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD ) + person = frame.table("person").copy() + person.loc[person.index[_PARENT_INDEX], "own_children_in_household"] = 7.0 + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + assert not any("named as a parent" in failure for failure in gate.failures) def test_null_parent_id_fails(self) -> None: frame = with_us_eligibility_inputs( @@ -631,3 +640,100 @@ def test_everyone_disabled_fails_the_share_band(self) -> None: gate = us_eligibility_inputs_signal_gate(frame) assert not gate.passed assert any("disabled share" in failure for failure in gate.failures) + + +class TestParentIdReworkRegressions: + """Review of microcosm#1032: existing frames, exact ids, int32 consumers.""" + + def test_legacy_frame_gets_only_the_parent_ids_appended(self) -> None: + # A --base-h5 input carries trusted eligibility values that differ from + # a raw re-derivation; adding the new columns must not rewrite them. + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").copy() + person.loc[person.index[_VETERAN_INDEX], "veterans_benefits"] = 24_000.0 + person = person.drop(columns=list(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS)) + legacy = _replace_person(frame, person) + + result = with_us_eligibility_inputs(legacy, seed=0, time_period=TIME_PERIOD) + out = result.table("person") + + assert out["veterans_benefits"].iloc[_VETERAN_INDEX] == 24_000.0 + for column in person.columns: + pd.testing.assert_series_equal(out[column], person[column]) + # The raw pointers are still present, so the ids resolve. + assert int(out["parent_1_id"].iloc[_CHILD_INDEX]) == int( + out["person_id"].iloc[_PARENT_INDEX] + ) + + def test_legacy_frame_without_raw_pointers_gets_unknown_ids(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").drop( + columns=[*US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, "PEPAR1", "PEPAR2"] + ) + result = with_us_eligibility_inputs( + _replace_person(frame, person), seed=0, time_period=TIME_PERIOD + ) + out = result.table("person") + + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + assert out[column].dtype == np.int64 + assert (out[column] == 0).all() + # All-zero ids mean "unknown", which the gate accepts. + assert us_eligibility_inputs_signal_gate(result).passed + + def test_household_ids_past_2_53_are_compared_exactly(self) -> None: + big = 10**16 + frame = with_us_eligibility_inputs( + _us_frame( + [ + {"PH_SEQ": 1, "A_LINENO": 1, "person_household_id": big}, + { + "PH_SEQ": 1, + "A_LINENO": 2, + "PEPAR1": 1, + "A_AGE": 8, + "person_household_id": big, + }, + {"PH_SEQ": 2, "A_LINENO": 1, "person_household_id": big + 1}, + { + "PH_SEQ": 2, + "A_LINENO": 2, + "PEPAR1": 1, + "A_AGE": 8, + "person_household_id": big + 1, + }, + ] + ), + seed=0, + time_period=TIME_PERIOD, + ) + person = frame.table("person").copy() + # Swap the two children's parents: counts still match, homes do not. + first, second = person["parent_1_id"].iloc[1], person["parent_1_id"].iloc[3] + person.loc[person.index[1], "parent_1_id"] = second + person.loc[person.index[3], "parent_1_id"] = first + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + + assert any( + "outside the pointing person's household" in failure + for failure in gate.failures + ) + + def test_linked_household_ids_must_fit_int32(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").copy() + shift = 2**32 + person["person_id"] = person["person_id"] + shift + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + person[column] = np.where(person[column] > 0, person[column] + shift, 0) + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + + assert any("int32" in failure for failure in gate.failures) + # The same links with int32-safe ids pass. + assert us_eligibility_inputs_signal_gate(frame).passed diff --git a/packages/microcosm-build/tests/test_us_input_mass.py b/packages/microcosm-build/tests/test_us_input_mass.py index 6c0babb717..1af8c3ddd3 100644 --- a/packages/microcosm-build/tests/test_us_input_mass.py +++ b/packages/microcosm-build/tests/test_us_input_mass.py @@ -94,3 +94,23 @@ def test_column_restriction_filters_raw_source_columns() -> None: ) assert set(totals) == {"student_loan_interest"} + + +def test_person_reference_columns_are_not_a_mass() -> None: + frame = _frame() + person = frame.table("person").copy() + person["parent_1_id"] = np.arange(len(person), dtype="int64") + 7 + person["parent_2_id"] = 0 + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables["person"] = person + with_ids = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + ) + + totals = us_input_mass_totals(with_ids) + + assert "parent_1_id" not in totals + assert "parent_2_id" not in totals + assert totals == us_input_mass_totals(frame) diff --git a/packages/microcosm-build/tests/test_us_multispine_pool.py b/packages/microcosm-build/tests/test_us_multispine_pool.py index cf86d2c2bb..fbba65a302 100644 --- a/packages/microcosm-build/tests/test_us_multispine_pool.py +++ b/packages/microcosm-build/tests/test_us_multispine_pool.py @@ -3445,19 +3445,20 @@ def test_row_sensitive_prefix_exposes_clone_first_defects( time_period=2024, acs_rent_donor=donor, ) - old_order = multispine_pool_module.with_us_eligibility_inputs( - old_order, - seed=0, - time_period=2024, - ) - person = old_order.table("person") source_id = support_source_id_column("person") heads = person["is_household_head"].eq(True) rent_variants = person.loc[heads].groupby(source_id)["pre_subsidy_rent"].nunique() assert rent_variants.eq(2).all() - parents = person["A_LINENO"].eq(1) - assert person.loc[parents, "own_children_in_household"].eq(2.0).all() + # Clone-first, the raw (PH_SEQ, A_LINENO) pointers are duplicated. The + # count used to double silently; resolving the pointers to parent ids + # (microcosm#884) now refuses the ambiguous line outright. + with pytest.raises(SourceRuntimeError, match="repeats line"): + multispine_pool_module.with_us_eligibility_inputs( + old_order, + seed=0, + time_period=2024, + ) def test_every_source_operator_output_has_a_pool_owner() -> None: diff --git a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py index 6131cb700c..fcb91d08d9 100644 --- a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py +++ b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py @@ -2506,7 +2506,7 @@ def capture_equality(expected: object, actual: object) -> None: "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "f830589331dad583398485de5724dace0714d8526fb63b37f81cd783b1cb8f30", + "spec_sha256": "1cf1f44d9a11458f59ea7d75dce68303816cab3e66668a6c6f6ecfe62d56f8dc", }, } @@ -7201,3 +7201,46 @@ def test_local_artifact_reference_never_embeds_host_absolute_paths( for reference in (repo_reference, home_reference, outside_reference): assert not reference.startswith("local:///") assert "local://Users/" not in reference + + +def test_post_operator_checkpoint_without_parent_ids_is_stale( + pool_tool: ModuleType, +) -> None: + """microcosm#884: the byte-stable legacy identity cannot see new outputs.""" + + from microcosm.build.us_runtime.eligibility_inputs import ( + US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS, + ) + + def manifest(columns: list[str]) -> dict[str, object]: + return { + "frame_schema": { + "entities": { + "person": [{"name": name, "dtype": "int64"} for name in columns] + } + } + } + + before = [ + "person_id", + *[c for c in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS if "parent" not in c], + ] + after = ["person_id", *US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS] + + assert pool_tool._checkpoint_missing_person_outputs( + manifest(before), stage="simulated" + ) == ["parent_1_id", "parent_2_id"] + assert pool_tool._checkpoint_missing_person_outputs( + manifest(before), stage="transferred" + ) == ["parent_1_id", "parent_2_id"] + # Checkpoints written before the source operators run are not bound. + assert ( + pool_tool._checkpoint_missing_person_outputs( + manifest(before), stage="assembled" + ) + == [] + ) + assert ( + pool_tool._checkpoint_missing_person_outputs(manifest(after), stage="simulated") + == [] + ) diff --git a/packages/microcosm-build/tests/test_us_puf_capital_gains_tail.py b/packages/microcosm-build/tests/test_us_puf_capital_gains_tail.py index f2fb44a06c..4bf43b3318 100644 --- a/packages/microcosm-build/tests/test_us_puf_capital_gains_tail.py +++ b/packages/microcosm-build/tests/test_us_puf_capital_gains_tail.py @@ -1090,3 +1090,35 @@ def dropping_assign(*args, **kwargs): monkeypatch.setattr(mod, "_assign_tail_donors", dropping_assign) with pytest.raises(ValueError, match="bijection"): mod.transfer_puf_capital_gains_tail(frame, donor, seed=7) + + +def test_shift_person_references_moves_ids_and_keeps_the_zero_sentinel() -> None: + person = pd.DataFrame( + { + "person_id": np.asarray([11, 12, 13], dtype=np.int64), + "parent_1_id": np.asarray([0, 11, 11], dtype=np.int64), + "parent_2_id": np.asarray([0, 0, 12], dtype=np.int64), + } + ) + tail_module._shift_person_references(person, 10) + + assert person["parent_1_id"].tolist() == [0, 21, 21] + assert person["parent_2_id"].tolist() == [0, 0, 22] + assert person["parent_1_id"].dtype == np.int64 + + +def test_tail_clone_shifts_person_references_on_the_cloned_people(monkeypatch) -> None: + """microcosm#884: the clone path must route cloned people through the shift.""" + + calls: list[tuple[int, int]] = [] + real = tail_module._shift_person_references + + def spy(person: pd.DataFrame, id_multiplier: int) -> None: + calls.append((len(person), int(id_multiplier))) + real(person, id_multiplier) + + monkeypatch.setattr(tail_module, "_shift_person_references", spy) + _pre_652_all_adequate_reference_frame() + + assert len(calls) == 1 + assert calls[0][0] > 0 and calls[0][1] > 0 diff --git a/tools/build_us_multispine_pool.py b/tools/build_us_multispine_pool.py index 7cc27764cc..70820fb68c 100644 --- a/tools/build_us_multispine_pool.py +++ b/tools/build_us_multispine_pool.py @@ -116,6 +116,9 @@ CURRENT_CONGRESSIONAL_DISTRICT_VINTAGE, load_congressional_district_vintage_crosswalk, ) +from microcosm.build.us_runtime.eligibility_inputs import ( + US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS, +) from microcosm.build.us_runtime.h5_io import ( US_MULTISPINE_AGREEMENT_DIAGNOSTICS_ARTIFACT_KIND, US_MULTISPINE_POOL_H5_ARTIFACT_KIND, @@ -2447,6 +2450,21 @@ def _load(self, stage: str) -> MultispinePoolCheckpoint | None: f"{expected_identity_sha256}." ) return None + missing_outputs = _checkpoint_missing_person_outputs(manifest, stage=stage) + if missing_outputs: + # The retiring pipeline's identity is byte-stable and does not + # bind the source-stage outputs, so a checkpoint written before + # an output existed still matches it. Reusing it would skip the + # stage that adds the column (microcosm#884); recompute instead. + self._attempts[stage] = { + "load_status": "missing_person_outputs", + "ignored_checkpoint": {"missing_person_columns": missing_outputs}, + } + print( + f"Ignored stale pool checkpoint {stage!r} at {path}: " + f"person columns {missing_outputs} are missing." + ) + return None if manifest.get("identity_sha256") != expected_identity_sha256: raise ValueError(f"{stage} checkpoint manifest identity digest changed") checkpoint_receipt = manifest.get("checkpoint") @@ -2918,6 +2936,33 @@ def _frame_schema_payload(frame: Frame) -> dict[str, object]: } +#: Checkpoint stages written after the pre-clone source operators ran, whose +#: person table must therefore carry every eligibility output. +_POST_SOURCE_OPERATOR_CHECKPOINT_STAGES = frozenset({"transferred", "simulated"}) + + +def _checkpoint_missing_person_outputs( + manifest: Mapping[str, object], *, stage: str +) -> list[str]: + """Eligibility outputs a post-operator checkpoint's recorded schema lacks.""" + + if stage not in _POST_SOURCE_OPERATOR_CHECKPOINT_STAGES: + return [] + schema = manifest.get("frame_schema") + entities = schema.get("entities") if isinstance(schema, Mapping) else None + person = entities.get("person") if isinstance(entities, Mapping) else None + if not isinstance(person, list): + return [] + present = { + str(column.get("name")) for column in person if isinstance(column, Mapping) + } + return [ + column + for column in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS + if column not in present + ] + + def _validate_checkpoint_frame( frame: Frame, *, From 813978feb75d7943e4f04623932448fcd171046d Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 27 Sep 2026 02:33:06 -0400 Subject: [PATCH 5/7] Register the parent ids as stored non-variable columns; make them optional Round-2 review of #1032: - Main's stored-input contract (#1031) refuses a stored column that is neither a variable of the certified engine (policyengine-us 2.2.1) nor registered. parent_1_id/parent_2_id now have register entries naming their producer and the removal condition (the pinned engine defining them, which policyengine-us#9406 does); the producer-binding tests cover them. - Existing frames with the pre-#884 outputs (a --base-h5 input, an existing pool) pass through untouched instead of having ids appended: resolving raw pointers there keyed on (PH_SEQ, A_LINENO) pairs that repeat after cloning and could export ACS reference-person links. The release gate treats the ids as optional and asserts their identities only when both are present, so existing pools still feed releases. - A vacuous test assertion now asserts the gate passes. Co-Authored-By: Claude Opus 5.5 --- .../build/us_runtime/eligibility_inputs.py | 86 +++++++------------ .../tests/test_us_eligibility_inputs.py | 42 +++++---- .../tests/test_us_stored_input_register.py | 10 ++- .../src/microcosm/data/stored_inputs.py | 15 ++++ 4 files changed, 75 insertions(+), 78 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py index 36a50e11a2..ef27ddeace 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/eligibility_inputs.py @@ -87,9 +87,9 @@ remap fails closed instead of exporting a stale pointer. Existing-frame note: a frame that already carries the pre-#884 outputs -with signal (a ``--base-h5`` release input) gets only the two id columns -appended — resolved from the raw pointers if it still has them, ``0`` -otherwise — and no other column is recomputed. +with signal (a ``--base-h5`` release input, an existing pool) passes through +untouched, without parent ids; the gate asserts the parent-id identities only +on frames that carry them. """ from __future__ import annotations @@ -185,13 +185,6 @@ #: :data:`US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS` is load-bearing. _PARENT_POINTER_COLUMNS: tuple[str, ...] = ("PEPAR1", "PEPAR2") -#: Raw columns the parent-id resolution needs on top of ``person_id``. -_PARENT_ID_SOURCE_COLUMNS: tuple[str, ...] = ( - *_PARENT_POINTER_COLUMNS, - "PH_SEQ", - "A_LINENO", -) - #: The frame's person identifier. Not an ASEC source column: it is the #: structural id the stage aligns its output on, so it is already present. _PERSON_ID_COLUMN = "person_id" @@ -469,17 +462,16 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> have_legacy = all( column in person.columns for column in _LEGACY_OUTPUT_COLUMNS ) and _disabled_carries_signal(person) - have_parent_ids = all( - column in person.columns for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS - ) - if have_legacy and have_parent_ids: - return frame if have_legacy: - # A frame built before the parent ids existed (e.g. a --base-h5 - # release input) already carries trusted eligibility inputs. Add the - # two id columns and touch nothing else: recomputing the whole stage - # here would overwrite carried values with raw-column re-derivations. - return _with_parent_ids_appended(frame) + # A frame that already carries trusted eligibility inputs passes + # through untouched, with or without the parent ids. A frame built + # before they existed (a --base-h5 input, an existing pool) keeps + # exactly what it had: recomputing here would overwrite carried values + # with raw-column re-derivations, and resolving raw pointers on a + # cloned or pooled frame would key on (PH_SEQ, A_LINENO) pairs that + # are no longer unique there. Without the ids, PolicyEngine-US uses + # its count-based proxy, as before microcosm#884. + return frame stage_person = person.copy(deep=True) stage_person[_PERSON_WEIGHT_COLUMN] = frame.resolve_weights("person").values @@ -533,37 +525,6 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> ) -def _with_parent_ids_appended(frame: Frame) -> Frame: - """Add ``parent_1_id``/``parent_2_id`` without touching any other column. - - The ids are resolved from the raw ASEC pointers when the person table - still carries them (``PEPAR1``/``PEPAR2``, ``PH_SEQ``, ``A_LINENO``) and - are ``0`` ("unknown") otherwise, which PolicyEngine-US reads as "use the - count-based proxy" — the behavior such a frame had before this column - existed. - """ - - person = frame.table("person") - if all(column in person.columns for column in _PARENT_ID_SOURCE_COLUMNS): - resolved = _parent_person_ids(person) - else: - resolved = { - column: np.zeros(len(person), dtype=np.int64) - for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS - } - tables = {entity: frame.table(entity).copy() for entity in frame.entities} - for column, values in resolved.items(): - tables["person"][column] = np.asarray(values, dtype=np.int64) - return Frame( - tables, - frame.schema, - {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, - frame.strata, - mass_log=frame.mass_log, - metadata=frame.metadata, - ) - - def _pointed_at_counts( person_id: np.ndarray, resolved: dict[str, np.ndarray] ) -> np.ndarray: @@ -825,10 +786,11 @@ def us_eligibility_inputs_signal_gate(frame: Frame) -> GateResult: person = frame.table("person") failures: list[str] = [] + # The parent ids are optional: a frame built before them (or without raw + # pointers) carries none, and PolicyEngine-US then uses its count-based + # proxy. When present, their identities are asserted below. missing = [ - column - for column in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS - if column not in person.columns + column for column in _LEGACY_OUTPUT_COLUMNS if column not in person.columns ] if missing: return GateResult( @@ -867,7 +829,21 @@ def us_eligibility_inputs_signal_gate(frame: Frame) -> GateResult: failures.append( f"{label} {share:.3f} outside plausibility band [{low}, {high}]." ) - failures.extend(_parent_id_invariant_failures(person)) + parent_columns = [ + column + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + if column in person.columns + ] + if parent_columns and len(parent_columns) != len( + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + ): + failures.append( + f"person columns {parent_columns} are present without their pair; " + f"the parent ids {list(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS)} are " + "written together." + ) + elif parent_columns: + failures.extend(_parent_id_invariant_failures(person)) return GateResult( name="eligibility_inputs_signal", passed=not failures, diff --git a/packages/microcosm-build/tests/test_us_eligibility_inputs.py b/packages/microcosm-build/tests/test_us_eligibility_inputs.py index f1c02a89ef..3fd5c6c2e2 100644 --- a/packages/microcosm-build/tests/test_us_eligibility_inputs.py +++ b/packages/microcosm-build/tests/test_us_eligibility_inputs.py @@ -590,7 +590,7 @@ def test_count_above_the_pointers_passes(self) -> None: person = frame.table("person").copy() person.loc[person.index[_PARENT_INDEX], "own_children_in_household"] = 7.0 gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) - assert not any("named as a parent" in failure for failure in gate.failures) + assert gate.passed, gate.failures def test_null_parent_id_fails(self) -> None: frame = with_us_eligibility_inputs( @@ -645,9 +645,11 @@ def test_everyone_disabled_fails_the_share_band(self) -> None: class TestParentIdReworkRegressions: """Review of microcosm#1032: existing frames, exact ids, int32 consumers.""" - def test_legacy_frame_gets_only_the_parent_ids_appended(self) -> None: - # A --base-h5 input carries trusted eligibility values that differ from - # a raw re-derivation; adding the new columns must not rewrite them. + def test_legacy_frame_passes_through_untouched(self) -> None: + # A --base-h5 input or an existing pool carries trusted eligibility + # values that differ from a raw re-derivation, and its raw + # (PH_SEQ, A_LINENO) keys may repeat after cloning; it must come out + # exactly as it went in, without parent ids. frame = with_us_eligibility_inputs( _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD ) @@ -657,33 +659,29 @@ def test_legacy_frame_gets_only_the_parent_ids_appended(self) -> None: legacy = _replace_person(frame, person) result = with_us_eligibility_inputs(legacy, seed=0, time_period=TIME_PERIOD) - out = result.table("person") - assert out["veterans_benefits"].iloc[_VETERAN_INDEX] == 24_000.0 - for column in person.columns: - pd.testing.assert_series_equal(out[column], person[column]) - # The raw pointers are still present, so the ids resolve. - assert int(out["parent_1_id"].iloc[_CHILD_INDEX]) == int( - out["person_id"].iloc[_PARENT_INDEX] + assert result is legacy + assert result.table("person")["veterans_benefits"].iloc[_VETERAN_INDEX] == ( + 24_000.0 ) - def test_legacy_frame_without_raw_pointers_gets_unknown_ids(self) -> None: + def test_gate_accepts_a_frame_without_parent_ids(self) -> None: frame = with_us_eligibility_inputs( _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD ) person = frame.table("person").drop( - columns=[*US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, "PEPAR1", "PEPAR2"] - ) - result = with_us_eligibility_inputs( - _replace_person(frame, person), seed=0, time_period=TIME_PERIOD + columns=list(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS) ) - out = result.table("person") + assert us_eligibility_inputs_signal_gate(_replace_person(frame, person)).passed - for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: - assert out[column].dtype == np.int64 - assert (out[column] == 0).all() - # All-zero ids mean "unknown", which the gate accepts. - assert us_eligibility_inputs_signal_gate(result).passed + def test_gate_refuses_one_parent_id_without_its_pair(self) -> None: + frame = with_us_eligibility_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person").drop(columns=["parent_2_id"]) + gate = us_eligibility_inputs_signal_gate(_replace_person(frame, person)) + assert not gate.passed + assert any("without their pair" in failure for failure in gate.failures) def test_household_ids_past_2_53_are_compared_exactly(self) -> None: big = 10**16 diff --git a/packages/microcosm-build/tests/test_us_stored_input_register.py b/packages/microcosm-build/tests/test_us_stored_input_register.py index 9bfc43ac0a..107f8da81d 100644 --- a/packages/microcosm-build/tests/test_us_stored_input_register.py +++ b/packages/microcosm-build/tests/test_us_stored_input_register.py @@ -26,6 +26,9 @@ from microcosm.build.us_runtime import acs_pums, acs_transfer from microcosm.build.us_runtime import puf_capital_gains_tail as puf_tail from microcosm.build.us_runtime.base_pool import spine_column +from microcosm.build.us_runtime.eligibility_inputs import ( + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, +) from microcosm.build.us_runtime.operator_boundary import _ACS_NATIVE_INPUT_CONTRACTS from microcosm.build.us_runtime.support_provenance import ( spine_source_id_column, @@ -170,9 +173,10 @@ def test_every_register_entry_is_a_named_producer_column(): | tail | _ACS_NATIVE_AMOUNTS | {"puma_geoid"} + | set(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS) ) assert len(_ACS_NATIVE_AMOUNTS) == 6 - assert len(US_STORED_NON_VARIABLE_COLUMNS) == 30 + 6 + 6 + 6 + 1 + assert len(US_STORED_NON_VARIABLE_COLUMNS) == 30 + 6 + 6 + 6 + 1 + 2 def test_every_register_reason_names_where_its_column_comes_from(): @@ -201,6 +205,10 @@ def test_every_register_reason_names_where_its_column_comes_from(): TAX_UNIT_FILING_STATUS_COLUMN: ( "microcosm.frame.units.TAX_UNIT_FILING_STATUS_COLUMN" ), + **{ + column: "US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS" + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + }, } for column, owner in expected_owner.items(): assert owner in US_STORED_NON_VARIABLE_COLUMNS[column], column diff --git a/packages/microcosm-data/src/microcosm/data/stored_inputs.py b/packages/microcosm-data/src/microcosm/data/stored_inputs.py index 19a4c11f18..93c68f05d8 100644 --- a/packages/microcosm-data/src/microcosm/data/stored_inputs.py +++ b/packages/microcosm-data/src/microcosm/data/stored_inputs.py @@ -407,6 +407,21 @@ "puf_capital_gains_tail_transfer_weight": _PUF_TAIL.format( detail="the donor weight carried by the transfer" ), + **{ + column: ( + f"The person_id of the co-resident parent the person's CPS " + f"ASEC {pointer} pointer names, or 0 when unknown, written by " + "microcosm.build.us_runtime.eligibility_inputs " + "(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS; microcosm#884). The " + "certified engine does not define it yet, so it is ignored " + "there; policyengine-us#9406 adds it as an input. Remove this " + "entry when the pinned engine defines the variable." + ) + for column, pointer in ( + ("parent_1_id", "PEPAR1"), + ("parent_2_id", "PEPAR2"), + ) + }, } ) From dd372e109801b86d7adb9a1aba86b86287c84ee6 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 27 Sep 2026 02:45:31 -0400 Subject: [PATCH 6/7] Rest the parent-id register entries on their producer, not an examined file Round-3 review of #1032: two microcosm-data tests require every register entry to be stored by an H5 file examined for #1026, and no examined file stores the parent ids yet. Following the *_spine precedent, the tests name them as producer-bound entries (bound to eligibility_inputs by test_us_stored_input_register.py), and the register docstring says so. Also fixes a gate assertion that filtered on text no message contains. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_us_eligibility_inputs.py | 4 +++- .../src/microcosm/data/stored_inputs.py | 6 ++++++ .../tests/test_stored_inputs.py | 20 +++++++++++++++---- 3 files changed, 25 insertions(+), 5 deletions(-) diff --git a/packages/microcosm-build/tests/test_us_eligibility_inputs.py b/packages/microcosm-build/tests/test_us_eligibility_inputs.py index 3fd5c6c2e2..c9cd5075d0 100644 --- a/packages/microcosm-build/tests/test_us_eligibility_inputs.py +++ b/packages/microcosm-build/tests/test_us_eligibility_inputs.py @@ -625,10 +625,12 @@ def test_person_id_zero_cannot_be_named_and_is_reported(self) -> None: assert person["parent_1_id"].iloc[1] == 0 summary = us_eligibility_inputs_summary(frame) assert summary["pointers_unnameable_at_person_id_zero"] == 1 + # The person holding id 0 has a child no pointer can name; the count + # bound (pointers naming a person <= their count) still holds. failures = [ failure for failure in us_eligibility_inputs_signal_gate(frame).failures - if "parent pointers" in failure + if "parent" in failure ] assert failures == [] diff --git a/packages/microcosm-data/src/microcosm/data/stored_inputs.py b/packages/microcosm-data/src/microcosm/data/stored_inputs.py index 93c68f05d8..f1ad592d76 100644 --- a/packages/microcosm-data/src/microcosm/data/stored_inputs.py +++ b/packages/microcosm-data/src/microcosm/data/stored_inputs.py @@ -265,6 +265,12 @@ #: - the ACS local-area release ``populace-us-2024-buildo-acs-local- #: 767312d60-20260923T074941Z`` (the ``*_spine`` tags as well). #: +#: One pair postdates those files: ``parent_1_id`` and ``parent_2_id`` +#: (microcosm#884). No examined file stores them yet, so they rest on their +#: live producer, ``eligibility_inputs.US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS``, +#: which ``test_us_stored_input_register.py`` binds, and leave the register +#: once the pinned engine defines them. +#: #: Two model-named non-variable columns that the published default, its #: receipt child and the ACS local-area release store are deliberately absent, #: because each is a retired engine input rather than metadata: diff --git a/packages/microcosm-data/tests/test_stored_inputs.py b/packages/microcosm-data/tests/test_stored_inputs.py index 34d3a34646..878d59420f 100644 --- a/packages/microcosm-data/tests/test_stored_inputs.py +++ b/packages/microcosm-data/tests/test_stored_inputs.py @@ -257,9 +257,17 @@ def test_the_shipped_register_is_model_named_and_every_entry_has_a_reason(): assert "medicare_part_b_premiums" not in US_STORED_NON_VARIABLE_COLUMNS +#: Register entries that postdate the files examined for #1026 and rest on a +#: live producer instead: the parent ids (microcosm#884), bound to +#: eligibility_inputs.US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS by +#: test_us_stored_input_register.py (microcosm-data cannot import the build). +_PRODUCER_BOUND_ENTRIES = frozenset({"parent_1_id", "parent_2_id"}) + + def test_every_register_entry_is_stored_by_a_file_examined_for_1026(): """The register holds evidence, not guesses: each entry is a column one of - the examined H5 files actually stores.""" + the examined H5 files actually stores, except the producer-bound entries + that postdate those files.""" stored = { column @@ -268,7 +276,9 @@ def test_every_register_entry_is_stored_by_a_file_examined_for_1026(): for column in columns } - assert set(US_STORED_NON_VARIABLE_COLUMNS) <= stored + assert _PRODUCER_BOUND_ENTRIES <= set(US_STORED_NON_VARIABLE_COLUMNS) + assert set(US_STORED_NON_VARIABLE_COLUMNS) - _PRODUCER_BOUND_ENTRIES <= stored + assert not _PRODUCER_BOUND_ENTRIES & stored def test_the_examined_files_store_no_state_code_column(): @@ -1219,8 +1229,10 @@ def stored_by(*names): "family_spine", "marital_unit_spine", } - unbacked = set(US_STORED_NON_VARIABLE_COLUMNS) - stored_by( - _REHEARSAL_EXPORT, _STACKED_POOL + unbacked = ( + set(US_STORED_NON_VARIABLE_COLUMNS) + - _PRODUCER_BOUND_ENTRIES + - stored_by(_REHEARSAL_EXPORT, _STACKED_POOL) ) assert unbacked == spine_tags assert unbacked <= stored_by(_ACS_LOCAL_RELEASE) From 4f152b6483265052a2b8114f911deea645bd4d47 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 27 Sep 2026 03:57:44 -0400 Subject: [PATCH 7/7] Narrow the resume guard to pre-#884 checkpoints The guard treated any transferred/simulated checkpoint lacking an eligibility output as stale, so the checkpoint round-trip tests' synthetic fixtures (which carry no eligibility columns) were rebuilt instead of resumed, and it still required the parent ids after they became optional. It now marks stale only a checkpoint that carries the pre-#884 eligibility outputs but not the parent ids: the shape main writes, where resuming would build without ids that a fresh run produces. Anything else is left to the existing identity and schema checks. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_us_multispine_pool_tool.py | 8 ++++++++ tools/build_us_multispine_pool.py | 19 +++++++++++++++++-- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py index fcb91d08d9..25d87009f7 100644 --- a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py +++ b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py @@ -7244,3 +7244,11 @@ def manifest(columns: list[str]) -> dict[str, object]: pool_tool._checkpoint_missing_person_outputs(manifest(after), stage="simulated") == [] ) + # A checkpoint with no eligibility outputs at all (a synthetic fixture) is + # left to the identity and schema checks. + assert ( + pool_tool._checkpoint_missing_person_outputs( + manifest(["person_id"]), stage="simulated" + ) + == [] + ) diff --git a/tools/build_us_multispine_pool.py b/tools/build_us_multispine_pool.py index 70820fb68c..828cceffdc 100644 --- a/tools/build_us_multispine_pool.py +++ b/tools/build_us_multispine_pool.py @@ -118,6 +118,7 @@ ) from microcosm.build.us_runtime.eligibility_inputs import ( US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS, + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, ) from microcosm.build.us_runtime.h5_io import ( US_MULTISPINE_AGREEMENT_DIAGNOSTICS_ARTIFACT_KIND, @@ -2944,7 +2945,14 @@ def _frame_schema_payload(frame: Frame) -> dict[str, object]: def _checkpoint_missing_person_outputs( manifest: Mapping[str, object], *, stage: str ) -> list[str]: - """Eligibility outputs a post-operator checkpoint's recorded schema lacks.""" + """Parent-id columns a pre-#884 post-operator checkpoint lacks. + + A checkpoint written before microcosm#884 carries the other eligibility + outputs but not the parent ids; resuming from it would build without + them where a fresh run would build with them. Only that shape is stale: + a checkpoint with no eligibility outputs at all is left to the existing + identity and schema checks. + """ if stage not in _POST_SOURCE_OPERATOR_CHECKPOINT_STAGES: return [] @@ -2956,9 +2964,16 @@ def _checkpoint_missing_person_outputs( present = { str(column.get("name")) for column in person if isinstance(column, Mapping) } - return [ + legacy = [ column for column in US_ELIGIBILITY_INPUTS_OUTPUT_COLUMNS + if column not in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + ] + if not all(column in present for column in legacy): + return [] + return [ + column + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS if column not in present ]