From bbb5c6851ecf0c7b5a8ad22a67babb29b1e4b49e Mon Sep 17 00:00:00 2001 From: PavelMakarchuk Date: Thu, 5 Nov 2026 17:47:32 -0500 Subject: [PATCH] Resolve parent ids on frames built before they existed A --base-h5 release input or an existing pool carries the pre-#884 eligibility outputs, so with_us_eligibility_inputs passed it through without parent_1_id / parent_2_id and no published release has them. Such a frame still carries PEPAR1 / PEPAR2, A_LINENO, PERIDNUM and person_household_id, so the stage now resolves the ids from those columns and adds them, leaving every carried value unchanged. The lookup keys on (person_household_id, A_LINENO), because a support clone repeats its source's PH_SEQ and lines in a household of its own. Only rows with CPS evidence point or are pointed at, since ACS rows carry synthesized reference-person links under the PEPAR names. A repeated key or a disagreement with the carried own_children_in_household fails closed. Closes #1185. Co-Authored-By: Claude Opus 5.5 --- ...-parent-ids-from-carried-pointers.added.md | 1 + .../build/us_runtime/eligibility_inputs.py | 193 ++++++++++++++-- .../us/test_us_eligibility_inputs.py | 217 +++++++++++++++++- 3 files changed, 391 insertions(+), 20 deletions(-) create mode 100644 changelog.d/1185-parent-ids-from-carried-pointers.added.md diff --git a/changelog.d/1185-parent-ids-from-carried-pointers.added.md b/changelog.d/1185-parent-ids-from-carried-pointers.added.md new file mode 100644 index 0000000000..a0248b6509 --- /dev/null +++ b/changelog.d/1185-parent-ids-from-carried-pointers.added.md @@ -0,0 +1 @@ +US frames built before `parent_1_id` and `parent_2_id` existed (a `--base-h5` release input, an existing pool) now gain the two ids from the CPS parent pointers they still carry, keyed on the frame's own household so support clones resolve inside their own copy; ACS rows and every other carried column are left as they were. 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 ef27ddeace..2fe082ab3d 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,22 @@ 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, an existing pool) passes through -untouched, without parent ids; the gate asserts the parent-id identities only -on frames that carry them. +with signal (a ``--base-h5`` release input, an existing pool) keeps every +column it carries. If it lacks the parent ids but still carries the raw +pointers, the CPS evidence column and ``person_household_id``, the ids are +resolved from those carried columns and added; nothing else changes +(microcosm#1185). The resolution keys on ``(person_household_id, A_LINENO)``, +not ``(PH_SEQ, A_LINENO)``: a support clone keeps its source's PH_SEQ and line +numbers but sits in its own household, so the frame's household id is what +keeps each pointer inside its own copy of the household. Only rows with CPS +evidence (a non-empty ``PERIDNUM``, the column the pool's ``cps_source`` scope +reads) take part, on both sides of a pointer: ACS rows carry synthesized +reference-person links under the same PEPAR names (``acs_pums``), which +``acs_inputs`` deliberately reports as unknown (0). On the CPS rows the carried +``own_children_in_household`` came from the same pointers, so the stage +requires the two to agree exactly and fails closed if they do not. A frame +without those carried columns passes through untouched, without parent ids; +the gate asserts the parent-id identities only on frames that carry them. """ from __future__ import annotations @@ -189,6 +202,15 @@ #: structural id the stage aligns its output on, so it is already present. _PERSON_ID_COLUMN = "person_id" +#: The frame's own household of each person. Unlike PH_SEQ it differs between +#: a support clone and its source, so carried pointers resolve against it. +_PERSON_HOUSEHOLD_ID_COLUMN = "person_household_id" + +#: The raw CPS ASEC column that marks a CPS-lineage row. ACS rows do not carry +#: it; the multispine pool's ``cps_source`` scope reads the same column +#: (``multispine_pool._CPS_SOURCE_EVIDENCE_COLUMN``). +_CPS_EVIDENCE_COLUMN = "PERIDNUM" + #: 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 @@ -428,13 +450,127 @@ def _disabled_carries_signal(person: pd.DataFrame) -> bool: return values.nunique() > 1 +def _cps_evidence_mask(person: pd.DataFrame) -> np.ndarray: + """Rows with a non-empty CPS evidence value (``PERIDNUM``).""" + + evidence = person[_CPS_EVIDENCE_COLUMN] + available = evidence.notna() + if pd.api.types.is_string_dtype(evidence.dtype) or evidence.dtype == object: + available &= evidence.astype("string").str.strip().ne("").fillna(False) + return available.to_numpy(dtype=bool) + + +def _integral_positive(values: np.ndarray) -> np.ndarray: + """Finite, positive, whole-number entries (a usable line or pointer).""" + + finite = np.isfinite(values) + whole = np.zeros(len(values), dtype=bool) + whole[finite] = values[finite] == np.floor(values[finite]) + return finite & whole & (values > 0) + + +def _parent_ids_from_carried_pointers( + person: pd.DataFrame, +) -> dict[str, np.ndarray] | None: + """Resolve the parent ids of a frame that carries the pre-#884 outputs. + + Returns ``None`` when the frame lacks a column the resolution needs (the + pointers, ``A_LINENO``, ``person_id``, ``person_household_id`` or the CPS + evidence column) or carries a non-numeric household or person id; the + frame then passes through without parent ids, as before microcosm#1185. + + A pointer names the person on the same ``A_LINENO`` in the pointing + person's own ``person_household_id``. Only CPS-evidenced rows point or are + pointed at; every other row reads 0 ("unknown"). Keys are exact int64, so + household ids past 2**53 stay distinct. + + Raises: + SourceRuntimeError: If ``(person_household_id, A_LINENO)`` repeats + among CPS-evidenced rows (an ambiguous pointer), or if the + resolved ids disagree with the carried + ``own_children_in_household`` on a CPS-evidenced row. Both read + the same pointers, so a disagreement means the frame's pointers or + households no longer describe the relationships it was built from. + """ + + required = ( + *_PARENT_POINTER_COLUMNS, + "A_LINENO", + _PERSON_ID_COLUMN, + _PERSON_HOUSEHOLD_ID_COLUMN, + _CPS_EVIDENCE_COLUMN, + ) + if any(column not in person.columns for column in required): + return None + household_values = pd.to_numeric( + person[_PERSON_HOUSEHOLD_ID_COLUMN], errors="coerce" + ) + person_id_values = pd.to_numeric(person[_PERSON_ID_COLUMN], errors="coerce") + if household_values.isna().any() or person_id_values.isna().any(): + return None + household = household_values.to_numpy(dtype=np.int64) + person_id = person_id_values.to_numpy(dtype=np.int64) + cps = _cps_evidence_mask(person) + line = pd.to_numeric(person["A_LINENO"], errors="coerce").to_numpy(dtype=np.float64) + + person_id_by_line: dict[tuple[int, int], int] = {} + for row in np.flatnonzero(cps & _integral_positive(line)).tolist(): + key = (int(household[row]), int(line[row])) + if key in person_id_by_line: + raise SourceRuntimeError( + "US eligibility-inputs parent-id resolution on a carried frame " + "requires (person_household_id, A_LINENO) to identify at most " + f"one CPS person; household {key[0]} repeats line {key[1]}." + ) + person_id_by_line[key] = int(person_id[row]) + + resolved: dict[str, np.ndarray] = {} + for column, pointer_column in zip( + US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS, _PARENT_POINTER_COLUMNS, strict=True + ): + pointer = pd.to_numeric(person[pointer_column], errors="coerce").to_numpy( + dtype=np.float64 + ) + values = np.zeros(len(person), dtype=np.int64) + for row in np.flatnonzero(cps & _integral_positive(pointer)).tolist(): + values[row] = person_id_by_line.get( + (int(household[row]), int(pointer[row])), 0 + ) + resolved[column] = values + + observed = pd.to_numeric( + person["own_children_in_household"], errors="coerce" + ).to_numpy(dtype=np.float64) + expected = _pointed_at_counts(person_id, resolved) + mismatched = cps & (observed != expected) & (person_id != 0) + if mismatched.any(): + examples = [ + { + "person_id": int(person_id[row]), + "own_children_in_household": float(observed[row]), + "pointed_at_by": float(expected[row]), + } + for row in np.flatnonzero(mismatched)[:5].tolist() + ] + raise SourceRuntimeError( + "US eligibility-inputs parent ids resolved on a carried frame " + "disagree with its own_children_in_household for " + f"{int(mismatched.sum())} CPS person(s); both come from the same " + f"PEPAR pointers. Examples: {examples}." + ) + return resolved + + 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 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. + disability distribution passes through untouched (idempotent). A frame + carrying every output column except the parent ids keeps its values and + gains the ids from its carried pointers, when it has the columns + :func:`_parent_ids_from_carried_pointers` reads; otherwise it too passes + through untouched. Any other surface — columns missing, or disability + constant at the engine default — is recomputed from the raw ASEC columns. Args: frame: A US-schema frame whose person table still carries the raw @@ -452,8 +588,10 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> 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, or if - the parent pointers cannot be resolved unambiguously. + SourceRuntimeError: If required raw ASEC columns are missing, if + the parent pointers cannot be resolved unambiguously, or if ids + resolved on a carried frame disagree with its carried + ``own_children_in_household``. """ if frame.schema != US_SCHEMA: @@ -463,15 +601,36 @@ def with_us_eligibility_inputs(frame: Frame, *, seed: int, time_period: int) -> column in person.columns for column in _LEGACY_OUTPUT_COLUMNS ) and _disabled_carries_signal(person) if have_legacy: - # 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 + # A frame that already carries trusted eligibility inputs keeps every + # value it has: recomputing here would overwrite carried values with + # raw-column re-derivations. A frame built before the parent ids + # existed (a --base-h5 input, an existing pool) gains them from its + # carried pointers when it still has them (microcosm#1185). The + # resolution keys on the frame's own household, because a clone or a + # pooled frame repeats (PH_SEQ, A_LINENO) pairs. Without those + # columns, PolicyEngine-US uses its count-based proxy, as before + # microcosm#884. + if any( + column in person.columns + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS + ): + # Ids already present (or one without its pair, which the gate + # refuses) are never rewritten. + return frame + carried = _parent_ids_from_carried_pointers(person) + if carried is None: + return frame + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + for column, values in carried.items(): + tables["person"][column] = values + 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, + ) stage_person = person.copy(deep=True) stage_person[_PERSON_WEIGHT_COLUMN] = frame.resolve_weights("person").values diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_eligibility_inputs.py b/packages/microcosm-build/tests/engine_free/us/test_us_eligibility_inputs.py index c9cd5075d0..e27ac63b7b 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_eligibility_inputs.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_eligibility_inputs.py @@ -649,9 +649,11 @@ class TestParentIdReworkRegressions: 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. + # values that differ from a raw re-derivation. Without the CPS + # evidence column (PERIDNUM) its pointers cannot be told apart from + # ACS synthesized links, so it must come out exactly as it went in, + # without parent ids (TestCarriedFrameParentIds covers frames that + # carry the evidence). frame = with_us_eligibility_inputs( _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD ) @@ -737,3 +739,212 @@ def test_linked_household_ids_must_fit_int32(self) -> None: 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 + + +def _carried(frame: Frame) -> Frame: + """The frame as a pre-#884 build stored it: every output but the ids.""" + + person = frame.table("person").drop( + columns=list(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS) + ) + return _replace_person(frame, person) + + +def _with_cps_evidence(rows: list[dict]) -> list[dict]: + """Rows that carry a CPS PERIDNUM, as every ASEC-lineage row does.""" + + return [{"PERIDNUM": f"cps-{index}", **row} for index, row in enumerate(rows)] + + +def _with_disability_signal(rows: list[dict]) -> list[dict]: + """Add a disabled adult in a household of their own. + + A carried frame is trusted only when ``is_disabled`` carries signal; + otherwise the stage recomputes everything from the raw columns. + """ + + return [ + *rows, + {"PERIDNUM": "cps-disabled", "PH_SEQ": 99, "A_LINENO": 1, "PEDISPHY": 1}, + ] + + +class TestCarriedFrameParentIds: + """microcosm#1185: ids for frames built before microcosm#884.""" + + def test_carried_frame_gains_the_ids_and_keeps_every_other_value(self) -> None: + derived = with_us_eligibility_inputs( + _us_frame(_with_cps_evidence(_plausible_rows())), + seed=0, + time_period=TIME_PERIOD, + ) + carried_person = _carried(derived).table("person").copy() + # A carried value that a raw re-derivation would overwrite. + carried_person.loc[ + carried_person.index[_VETERAN_INDEX], "veterans_benefits" + ] = 24_000.0 + carried = _replace_person(derived, carried_person) + + result = with_us_eligibility_inputs(carried, seed=0, time_period=TIME_PERIOD) + person = result.table("person") + + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + assert person[column].dtype == np.int64 + assert person[column].tolist() == derived.table("person")[column].tolist() + pd.testing.assert_frame_equal( + person.drop(columns=list(US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS)), + carried_person, + ) + assert person["veterans_benefits"].iloc[_VETERAN_INDEX] == 24_000.0 + assert us_eligibility_inputs_signal_gate(result).passed + # A second pass is the identity. + assert with_us_eligibility_inputs(result, seed=0, time_period=TIME_PERIOD) is ( + result + ) + + def test_support_clones_resolve_inside_their_own_household(self) -> None: + # A clone repeats its source's (PH_SEQ, A_LINENO); the frame's own + # household id keeps each pointer inside its copy of the household. + direct = with_us_eligibility_inputs( + _us_frame( + _with_disability_signal( + _with_cps_evidence([{}, {"PEPAR1": 1, "A_AGE": 8}]) + ) + ), + seed=0, + time_period=TIME_PERIOD, + ) + expanded = clone_us_frame_for_puf_support(direct) + person = expanded.table("person") + assert person.duplicated(["PH_SEQ", "A_LINENO"]).any() + + result = with_us_eligibility_inputs( + _carried(expanded), seed=0, time_period=TIME_PERIOD + ) + resolved = result.table("person") + + for column in US_ELIGIBILITY_INPUTS_PARENT_ID_COLUMNS: + assert resolved[column].tolist() == person[column].tolist() + clone_column = support_clone_index_column("person") + for clone_index in sorted(resolved[clone_column].unique()): + arm = resolved[resolved[clone_column] == clone_index] + child = arm[arm["A_LINENO"] == 2].iloc[0] + parent = arm[arm["A_LINENO"] == 1].iloc[0] + assert child["parent_1_id"] == parent["person_id"] + + def test_rows_without_cps_evidence_neither_point_nor_are_named(self) -> None: + # ACS rows carry synthesized reference-person links under the PEPAR + # names (acs_pums); acs_inputs reports them as unknown, and so does + # this resolution. + rows = [ + {"PERIDNUM": "cps-1", "PH_SEQ": 1, "A_LINENO": 1}, + {"PERIDNUM": "cps-2", "PH_SEQ": 1, "A_LINENO": 2, "PEPAR1": 1, "A_AGE": 8}, + {"PERIDNUM": None, "PH_SEQ": 2, "A_LINENO": 1}, + {"PERIDNUM": "", "PH_SEQ": 2, "A_LINENO": 2, "PEPAR1": 1, "A_AGE": 8}, + ] + derived = with_us_eligibility_inputs( + _us_frame(_with_disability_signal(rows)), seed=0, time_period=TIME_PERIOD + ) + result = with_us_eligibility_inputs( + _carried(derived), seed=0, time_period=TIME_PERIOD + ) + person = result.table("person") + + assert person["parent_1_id"].tolist() == [ + 0, + person["person_id"].iloc[0], + 0, + 0, + 0, + ] + assert person["parent_2_id"].tolist() == [0, 0, 0, 0, 0] + + def test_carried_frame_with_large_household_ids_resolves_exactly(self) -> None: + big = 2**53 + rows = _with_cps_evidence( + [ + {"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, + }, + # big + 1 is the same float64 as big; line 1 here must not + # be confused with line 1 above. + {"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, + }, + ] + ) + derived = with_us_eligibility_inputs( + _us_frame(_with_disability_signal(rows)), seed=0, time_period=TIME_PERIOD + ) + result = with_us_eligibility_inputs( + _carried(derived), seed=0, time_period=TIME_PERIOD + ) + person = result.table("person") + + assert person["parent_1_id"].iloc[1] == person["person_id"].iloc[0] + assert person["parent_1_id"].iloc[3] == person["person_id"].iloc[2] + + def test_repeated_household_line_is_refused(self) -> None: + derived = with_us_eligibility_inputs( + _us_frame( + _with_disability_signal( + _with_cps_evidence( + [ + {"PH_SEQ": 1, "A_LINENO": 1}, + {"PH_SEQ": 2, "A_LINENO": 1, "person_household_id": 1}, + ] + ) + ) + ), + seed=0, + time_period=TIME_PERIOD, + ) + with pytest.raises(SourceRuntimeError, match="carried frame .* repeats line 1"): + with_us_eligibility_inputs( + _carried(derived), seed=0, time_period=TIME_PERIOD + ) + + def test_count_that_disagrees_with_the_pointers_is_refused(self) -> None: + derived = with_us_eligibility_inputs( + _us_frame( + _with_disability_signal( + _with_cps_evidence([{}, {"PEPAR1": 1, "A_AGE": 8}]) + ) + ), + seed=0, + time_period=TIME_PERIOD, + ) + person = _carried(derived).table("person").copy() + person.loc[person.index[0], "own_children_in_household"] = 2.0 + with pytest.raises(SourceRuntimeError, match="disagree"): + with_us_eligibility_inputs( + _replace_person(derived, person), seed=0, time_period=TIME_PERIOD + ) + + def test_existing_parent_ids_are_never_rewritten(self) -> None: + derived = with_us_eligibility_inputs( + _us_frame( + _with_disability_signal( + _with_cps_evidence([{}, {"PEPAR1": 1, "A_AGE": 8}]) + ) + ), + seed=0, + time_period=TIME_PERIOD, + ) + person = derived.table("person").copy() + person["parent_1_id"] = 0 + zeroed = _replace_person(derived, person) + assert ( + with_us_eligibility_inputs(zeroed, seed=0, time_period=TIME_PERIOD) + is zeroed + )