From b13dc8c9559056932603d048bd33170d09e6195a Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 22 Sep 2026 10:38:29 -0400 Subject: [PATCH 1/4] Declare the US SPM measurement universe from the source record type MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit policyengine-us#9462 makes the SPM measurement universe a source declaration: a dataset supplies `spm_unit_spm_universe_status` for every SPM unit, and the model never infers scope from composition. Microcosm had no producer for that input. Add `us_runtime.spm_universe_source`. ACS housing units (`TYPEHUGQ == 1`) and every ASEC record are `INCLUDED`; ACS group quarters (`TYPEHUGQ` in {2, 3}) are `OUTSIDE`. Both arms come from Census SEHSD-WP2020-09 pages 6-7 and they are not symmetric: the ACS exclusion is a data limitation Census states explicitly, while the CPS ASEC sample frame already equals the CPS SPM universe, so the ASEC ruling is made at the spine and carries its own refusal if a vintage ever arrives with a record-type field to re-derive it against. Eleven named refusals fail closed rather than guess. Two earn their keep beyond the obvious: `SPM_UNIVERSE_UNIT_SPANS_KINDS`, natively unreachable but the point at which a bad merge would otherwise fabricate a majority kind; and `SPM_UNIVERSE_DEGRADED_PARTITION`, which proves the ASEC arm's SPM partition is the native `SPM_ID` one rather than the documented household fallback that any single missing `SPM_ID` triggers frame-wide. That check is ACS-exempt on purpose: ACS PUMS supplies no `SPM_ID`, so one unit per household is the native ACS partition, not a degradation. Never an inference. The status is read from the source's own record type, fixed before any composition is computed. A zero-adult *housing* unit is `INCLUDED` — marking it `OUTSIDE` would brand a data defect as out-of-universe and make the universe move whenever the role column moved. Both the anti-inference case and invariance to age and role are pinned by test. The status is never a calibration target, never a selection criterion, never a gate, and deliberately never a `spine_agreement` surface: it is meant to differ by spine, which is its entire content. Not wired. No build path calls it, the export adapter is untouched, the enrichment lane does not carry it, and no consumer changes. Registry entries moved: the spine-blindness inventory (classification plus the provenance-owner allowlist, both load-bearing) and the CI test grouping, which resolves to `us-qs` by suffix. `source_enrichment.PRODUCER_SOURCE_FILES` is deliberately not touched; see docs/us-spm-measurement-universe.md. Co-Authored-By: Claude Opus 5 --- ...s-spm-measurement-universe-source.added.md | 1 + docs/us-spm-measurement-universe.md | 153 ++++ .../build/us_runtime/spm_universe_source.py | 781 ++++++++++++++++++ .../tests/test_us_spine_blindness.py | 7 + .../tests/test_us_spm_universe_source.py | 647 +++++++++++++++ 5 files changed, 1589 insertions(+) create mode 100644 changelog.d/us-spm-measurement-universe-source.added.md create mode 100644 docs/us-spm-measurement-universe.md create mode 100644 packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py create mode 100644 packages/microcosm-build/tests/test_us_spm_universe_source.py diff --git a/changelog.d/us-spm-measurement-universe-source.added.md b/changelog.d/us-spm-measurement-universe-source.added.md new file mode 100644 index 0000000000..fa8fffd676 --- /dev/null +++ b/changelog.d/us-spm-measurement-universe-source.added.md @@ -0,0 +1 @@ +Add `us_runtime.spm_universe_source`, the source-owned producer for policyengine-us#9462's `spm_unit_spm_universe_status`: ACS housing units and every ASEC record are `INCLUDED`, ACS group quarters are `OUTSIDE`, and eleven named refusals fail closed rather than guess. Declared from the source's own record type, never from a unit's composition. Not wired into any build path and not exported. diff --git a/docs/us-spm-measurement-universe.md b/docs/us-spm-measurement-universe.md new file mode 100644 index 0000000000..00d9755198 --- /dev/null +++ b/docs/us-spm-measurement-universe.md @@ -0,0 +1,153 @@ +# The US SPM measurement universe + +PolicyEngine/policyengine-us#9462 makes the SPM measurement universe a +**source declaration**. A dataset supplies `spm_unit_spm_universe_status` ∈ +{`INCLUDED`, `OUTSIDE`, `UNRESOLVED`} for every SPM unit; `OUTSIDE` units get +missing SPM amounts and poverty indicators and never reach the threshold +provider, and an undeclared dataset is `UNRESOLVED` and raises +`SPM_UNIVERSE_REQUIRED`. The engine never infers scope from a unit's +composition, its tenure, its age or its geography. + +This page covers the Microcosm side. The full design — the source evidence, +the registry survey, the consumer enumeration and the open questions — is the +design note `docs/us-spm-measurement-universe-producer.md` on branch +`us-spm-measurement-universe-producer-design`. + +## The producer + +`microcosm.build.us_runtime.spm_universe_source` declares the universe from +each record's own source record type, never from what the record contains: + +| Source record | Status | +| --- | --- | +| ACS housing unit (`TYPEHUGQ == 1`) | `INCLUDED` | +| ACS group quarters (`TYPEHUGQ` ∈ {2, 3}) | `OUTSIDE` | +| ASEC record | `INCLUDED` | + +Both arms come from Census SEHSD-WP2020-09, pages 6–7, and they are not +symmetric. The ACS exclusion is a *data limitation* Census states explicitly: +the public ACS PUMS cannot distinguish noninstitutional group quarters, so the +ACS SPM sample is limited to people living in households, and Microcosm's ACS +spine inherits exactly that limitation. The ASEC ruling is the opposite shape: +the CPS ASEC sample frame already equals the CPS SPM universe, so there is no +ASEC record outside it to declare. The ASEC ruling is therefore made at the +spine rather than per record, and it carries its own refusal — if a frame ever +arrives carrying an ASEC household record-type field, the producer refuses +rather than applying a ruling that was made on the premise that no such field +is read. + +`UNRESOLVED` is the engine's *absence* sentinel, not a producer value. A +producer that cannot decide refuses. + +### Never an inference + +The ACS problem was discovered as SPM units with no classified adult, and the +tempting wrong design is to mark zero-adult units `OUTSIDE`. That would brand +a genuine household-unit data defect as out-of-universe and hide it, and it +would make the universe move whenever the role column moved. The input is +`TYPEHUGQ`, which the source fixes before any composition is computed. +`test_a_zero_adult_housing_unit_is_included` pins this, and +`test_the_declaration_is_invariant_to_age_and_role` pins it in both +directions. + +### Never a target, never a gate, never an agreement surface + +The status is not a calibration target, not a selection criterion, and not a +release gate. Poverty in this repository is a comparison only. The status is +also deliberately **not** a `spine_agreement` surface: it is *intended* to +differ by spine — that is its entire content — so registering it there would +assert that the ACS and ASEC arms should agree about a fact they are defined +to disagree about. + +`OUTSIDE` changes no weight, no membership, and no record's presence in the +file. The fiscal population is unchanged; only its SPM measurability is +declared. + +### Refusals + +Every refusal is a named, fail-closed `ValueError`: + +| Code | Condition | +| --- | --- | +| `SPM_UNIVERSE_COLUMN_EXISTS` | the SPM unit table already carries a declaration | +| `SPM_UNIVERSE_UNDECLARED_SPINE` | a provenance channel with no universe ruling, an untagged frame with no caller declaration, or a caller declaration over a frame that already declares | +| `SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND` | an ACS household whose `TYPEHUGQ` ∉ {1, 2, 3}, missing included | +| `SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND` | a non-ACS household carrying an ACS record type | +| `SPM_UNIVERSE_MIXED_HOUSEHOLD` | one household id resolving to more than one kind | +| `SPM_UNIVERSE_GQ_MULTI_PERSON` | an ACS group-quarters placeholder with more than one native person | +| `SPM_UNIVERSE_ORPHAN_MEMBER` | a membership id matching no owning entity row | +| `SPM_UNIVERSE_EMPTY_UNIT` | an SPM unit with no member | +| `SPM_UNIVERSE_UNIT_SPANS_KINDS` | one SPM unit whose members sit in households of different kinds | +| `SPM_UNIVERSE_DEGRADED_PARTITION` | the ASEC arm's SPM partition is not the native one | +| `SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED` | the frame carries an ASEC record-type field the spine-level ruling was not derived against | + +`SPM_UNIVERSE_UNIT_SPANS_KINDS` is natively unreachable — ACS group-quarters +households are one-person placeholders — and earns its keep the moment a +future spine, a donor merge or a clone-index bug puts a group-quarters person +into a multi-person unit. Silently picking a majority kind there would be a +fabricated universe decision. + +`SPM_UNIVERSE_DEGRADED_PARTITION` closes the largest silent-failure mode in +the area. `assign_us_unit_structure` uses the native `SPM_ID` when it is fully +present and falls back to `household_id` otherwise, and *any* missing `SPM_ID` +triggers that fallback for the whole frame — while `asec_pool` deliberately +tolerates a missing `SPM_ID` when globalizing. Declaring `INCLUDED` over a +degraded partition would attest to a unit structure that does not exist. The +check is ASEC-only on purpose: ACS PUMS supplies no `SPM_ID` at all, so the +ACS arm reaches `assign_us_unit_structure` without one and takes the household +fallback by construction — there, one SPM unit per household *is* the native +partition and carries no information. + +## What is wired, and what is not + +**Wired.** The producer, its refusals, and its tests. It is classified in the +spine-blindness inventory as a reviewed non-registry module and as a +source-spine provenance owner (the provenance tag is its input, not something +it must be blind to). + +**Not wired, deliberately.** + +- **No build path calls it.** Neither `tools/build_us_acs_local_release.py`, + `tools/build_us_fiscal_refresh_release.py` nor the legacy multispine base + invokes it. Adding a call site is separate work. +- **No export.** `microcosm.frame.adapters.policyengine_us` is untouched, so + the column reaches no release H5. The adapter change is blocked upstream: + see the design note's §4 and Q-A. +- **No enrichment-lane carry.** `microcosm.data.h5_enrichment` writes one + Boolean person column at the HDF type level; carrying an `spm_unit`-entity + string column is a separate, larger piece of work (design note §3). +- **No consumer change.** `check_spm_composition` still grades every SPM unit + with no universe predicate, and `reform_validation` still computes the 104 + state SPM rates over every person. Making `_person_rate` nullable-safe is a + separate PR, numerically inert on the pinned engine. +- **No producer source pin.** The module is deliberately *not* added to + `source_enrichment.PRODUCER_SOURCE_FILES`: that tuple is the SPM-role + enrichment producer's own reviewed source closure, and this module is not in + it until the enrichment lane carries the column. See "Open questions" below. + +## Open questions + +These are the design note's, carried here so they are answerable from the +repository. The note states each in full. + +- **Q-A — the formula-owned collision.** The engine variable as drafted cannot + be exported by Microcosm at all, for a reason unrelated to the pin. Resolve + upstream with a source-declaration marker, ask upstream to move the + household default, or add a Microcosm-side allowlist? +- **Q-B — sequencing.** Ship the held-back frame column now, wait for the + engine release, or both? +- **Q-C — group-quarters housing.** ACS group-quarters records are drawing a + housing-receipt flag today with no housing unit. Force the flags `False` on + the group-quarters mask, or supply an explicit reported-subsidy path? This + is a defect on `main` independent of the universe rule. +- **Q-D — the published-rate discontinuity.** The 104 state SPM rates move + when outside persons leave the denominator, correctly but visibly. +- **Q-E — the scope of the ASEC ruling.** Encoded here as a named, cited + constant with its own refusal, per the note's recommendation. + +## Upstream + +- Engine PR: PolicyEngine/policyengine-us#9462, fixing #9461. +- The dataset must carry `spm_unit_spm_universe_status` on `spm_unit` for + **every year** SPM is requested; a single-year H5 resolves only its base + year. Encode as member names. diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py new file mode 100644 index 0000000000..90930fc617 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py @@ -0,0 +1,781 @@ +"""Declare the US SPM measurement universe from each source's record type. + +PolicyEngine/policyengine-us#9462 makes the SPM measurement universe a +**source declaration**: a dataset supplies ``spm_unit_spm_universe_status`` in +``{INCLUDED, OUTSIDE, UNRESOLVED}`` for every SPM unit and year, and the model +never infers scope from composition. This module is Microcosm's producer for +that input. + +It is a source declaration, not a model. Nothing here fits, draws, ages, +weights, or imputes: the status is read off ``TYPEHUGQ`` — the ACS PUMS +household record type the source already puts on the frame's household table +(``acs_pums.py`` ``_HOUSEHOLD_FRAME_COLUMNS``) — and off the record's +provenance channel. ``OUTSIDE`` changes no weight, no membership, and no row's +presence in the file; it declares only that the source does not measure that +unit's SPM poverty. + +The rule +-------- + +=============================================== ========== +ACS housing unit (``TYPEHUGQ == 1``) INCLUDED +ACS group quarters (``TYPEHUGQ`` in ``{2, 3}``) OUTSIDE +ASEC record INCLUDED +=============================================== ========== + +Both arms come from Census SEHSD-WP2020-09 pages 6-7 (see +:data:`SPM_UNIVERSE_METHODOLOGY`), and they are not symmetric. The ACS +exclusion is a *data limitation* Census states explicitly — the public PUMS +cannot separate institutional from noninstitutional group quarters, so the ACS +SPM sample is limited to people living in households. The ASEC ruling is the +opposite shape: the CPS ASEC *sample frame* already equals the CPS SPM +universe, so there is no ASEC record outside it to declare. That ruling is +therefore made at the spine, not per record, and carries its own refusal +(:data:`ASEC_RECORD_TYPE_FIELDS`) so that a future ASEC vintage which admits a +record-type field cannot be silently declared in-universe. + +``UNRESOLVED`` is the engine's *absence* sentinel, not a value a producer +emits. A producer that cannot decide refuses; it never writes ``UNRESOLVED``. + +Never an inference +------------------ + +The ACS problem was discovered as SPM units with no classified adult, and it +would be easy — and wrong — to mark zero-adult units ``OUTSIDE``. That would +brand a genuine household-unit data defect as out-of-universe and hide it, and +it would make the universe move whenever the role column moved. The input is +``TYPEHUGQ``, which the source fixes before any composition is computed. The +anti-inference property is pinned by test: a zero-adult *housing* unit is +``INCLUDED``. + +Never a spine-agreement surface +------------------------------- + +A new categorical ``spm_unit`` column looks like something that belongs in +``spine_agreement``. It does not. The universe status is *intended* to differ +by spine — that is its entire content — and registering it there would assert +that the ACS and ASEC arms should agree about a fact they are defined to +disagree about. The status must never become a declared transfer, a derived +transfer, a take-up contract entry or a simulated output, which is what would +be required to admit it to that gate. + +Never a target +-------------- + +The status is not a calibration target, not a selection criterion, and not a +gate. Poverty in this repository is a comparison only. + +Import weight +------------- + +This producer is a leaf: a preflight or a release tool must be able to call it +without dragging the ACS loader into the process. Measured at this head on +this machine, ``import microcosm.build.us_runtime.acs_pums`` costs ~38 s and +``...stacked_spine`` ~90 s (both pull the country engine), so the four channel +tag values below are named here rather than imported from those modules. They +are not a second source of truth: :mod:`test_us_spm_universe_source` parses +each canonical definition out of its defining module with :mod:`ast` — no +import — and fails if this table drifts from it. The one canonical constant +that is free to import (:data:`BASE_ASEC_SUPPORT_CHANNEL`) is imported. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType +from typing import Any + +import numpy as np +import pandas as pd + +from .support_provenance import ( + BASE_ASEC_SUPPORT_CHANNEL, + support_channel_column, + support_clone_index_column, +) + +__all__ = [ + "ACS_ARM", + "ACS_GROUP_QUARTERS_KINDS", + "ACS_HOUSEHOLD_KIND_COLUMN", + "ACS_HOUSEHOLD_KINDS", + "ACS_HOUSING_UNIT_KIND", + "ACS_SPM_OUTSIDE_AUTHORITY", + "ASEC_ARM", + "ASEC_NATIVE_UNIT_COLUMN", + "ASEC_RECORD_TYPE_FIELDS", + "ASEC_SPM_INCLUDED_AUTHORITY", + "INCLUDED", + "OUTSIDE", + "SPM_UNIVERSE_CHANNEL_ARMS", + "SPM_UNIVERSE_METHODOLOGY", + "SPM_UNIVERSE_PROFILE_NAME", + "SPM_UNIVERSE_REFUSALS", + "SPM_UNIVERSE_RULE", + "SPM_UNIVERSE_STATUSES", + "SpmUniverseClassification", + "SpmUniverseSourceResult", + "UNIVERSE_INPUT", + "UNRESOLVED", + "attach_spm_universe_status", + "classify_spm_universe", +] + +#: The engine input this producer supplies (policyengine-us#9462). The name +#: and the three-state vocabulary below are deliberately identical to the +#: graph-native line's ``spm_input_contract``; there is no reason for two +#: spellings of the same declaration. +UNIVERSE_INPUT = "spm_unit_spm_universe_status" + +INCLUDED = "INCLUDED" +OUTSIDE = "OUTSIDE" +#: The engine's absence sentinel. Read back from a defaulted dataset; never +#: written by this producer. +UNRESOLVED = "UNRESOLVED" +SPM_UNIVERSE_STATUSES = (INCLUDED, OUTSIDE, UNRESOLVED) + +#: The two source universes this producer rules on. +ACS_ARM = "acs" +ASEC_ARM = "asec" + +#: The ACS PUMS household record type, as the source stages it. +ACS_HOUSEHOLD_KIND_COLUMN = "TYPEHUGQ" +ACS_HOUSING_UNIT_KIND = 1 +ACS_GROUP_QUARTERS_KINDS = (2, 3) +ACS_HOUSEHOLD_KINDS = (ACS_HOUSING_UNIT_KIND, *ACS_GROUP_QUARTERS_KINDS) + +#: The ASEC native SPM unit id. Its presence and completeness is what proves +#: the frame's SPM partition is the native one rather than the documented +#: household fallback (``microcosm.frame.units.assign_us_unit_structure``). +ASEC_NATIVE_UNIT_COLUMN = "SPM_ID" + +#: ASEC household record-type fields. Microcosm reads none of them at this +#: head, which is why the ASEC ruling is made at the spine. If one ever +#: arrives on a frame, the spine-level ruling has to be re-derived against it +#: rather than silently applied, so its presence refuses. +ASEC_RECORD_TYPE_FIELDS = ("H_TYPE", "HRHTYPE", "H_HHTYPE", "HUNITS", "H_LIVQRT") + +SPM_UNIVERSE_METHODOLOGY = ( + "Census SEHSD-WP2020-09, pages 6-7: all people living in group quarters " + "are excluded from the ACS SPM because the public ACS PUMS cannot " + "distinguish noninstitutional group quarters, while the CPS ASEC sample " + "frame already equals the CPS SPM universe." +) +#: Named authority for the ACS exclusion; identical to the graph-native line's +#: ``acs_spm_scope.ACS_SPM_OUTSIDE_AUTHORITY``. +ACS_SPM_OUTSIDE_AUTHORITY = "outside_acs_household_universe" +#: Named authority for the ASEC ruling. This one has no upstream: it is this +#: producer's own source-universe judgement, made on the Census quote above +#: plus the zero-adult reconciliation ``spm_role_source`` already enforces +#: against the pinned complete Census ASEC files. +ASEC_SPM_INCLUDED_AUTHORITY = "asec_sample_frame_equals_cps_spm_universe" +#: The named analysis scope a release declares when it carries this status. +SPM_UNIVERSE_PROFILE_NAME = "us_acs_household_and_asec_frame_universe_v1" + +SPM_UNIVERSE_RULE = ( + "TYPEHUGQ == 1 -> INCLUDED; TYPEHUGQ in {2, 3} -> OUTSIDE; ASEC record -> INCLUDED" +) + +#: Provenance tag values this producer has a universe ruling for. Anything +#: else refuses; see the module docstring on why these are named rather than +#: imported. +SPM_UNIVERSE_CHANNEL_ARMS: Mapping[str, str] = MappingProxyType( + { + # microcosm.build.us_runtime.acs_pums.ACS_2024_1YR_SPINE + "acs_2024_1yr": ACS_ARM, + # microcosm.build.us_runtime.stacked_spine.ACS_STACKED_SUPPORT_CHANNEL + "acs": ACS_ARM, + # microcosm.build.us_runtime.base_pool.ASEC_PUF_SPINE + "asec_puf": ASEC_ARM, + BASE_ASEC_SUPPORT_CHANNEL: ASEC_ARM, + } +) + +#: The household column naming each row's base spine +#: (``base_pool.spine_column("household")``), then the assembled support +#: channel, in the order the producer looks for them. +HOUSEHOLD_SPINE_COLUMN = "household_spine" +HOUSEHOLD_SUPPORT_CHANNEL_COLUMN = support_channel_column("household") +_CHANNEL_COLUMN_ORDER = (HOUSEHOLD_SPINE_COLUMN, HOUSEHOLD_SUPPORT_CHANNEL_COLUMN) +_PERSON_CLONE_INDEX_COLUMN = support_clone_index_column("person") +_NATIVE_CLONE_INDEX = 0 + +SPM_UNIVERSE_REFUSALS = ( + "SPM_UNIVERSE_COLUMN_EXISTS", + "SPM_UNIVERSE_UNDECLARED_SPINE", + "SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND", + "SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND", + "SPM_UNIVERSE_MIXED_HOUSEHOLD", + "SPM_UNIVERSE_GQ_MULTI_PERSON", + "SPM_UNIVERSE_ORPHAN_MEMBER", + "SPM_UNIVERSE_EMPTY_UNIT", + "SPM_UNIVERSE_UNIT_SPANS_KINDS", + "SPM_UNIVERSE_DEGRADED_PARTITION", + "SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED", +) + +_MAX_REPORTED_EXAMPLES = 5 + + +@dataclass(frozen=True) +class SpmUniverseClassification: + """One frame's universe declaration, plus the evidence it was made from.""" + + #: Member-name strings in ``unit_ids`` order. + status: np.ndarray + #: ``unit_ids`` as classified, in the order given. + unit_ids: np.ndarray + #: ``ACS_ARM``/``ASEC_ARM`` per household row. + household_arm: pd.Series + #: ACS record type per household row; missing on the ASEC arm. + household_kind: pd.Series + provenance: dict[str, Any] + + +@dataclass(frozen=True) +class SpmUniverseSourceResult: + """A frame carrying the declaration, with the classification behind it.""" + + frame: Any + classification: SpmUniverseClassification + + @property + def status(self) -> np.ndarray: + return self.classification.status + + @property + def provenance(self) -> dict[str, Any]: + return self.classification.provenance + + +def _require(condition: bool, code: str, detail: str) -> None: + """Refuse, named and fail-closed, in ``spm_role_source``'s idiom.""" + if not condition: + raise ValueError(f"{code}: {detail}") + + +def _examples(values: Any) -> str: + unique = pd.unique(np.asarray(values, dtype=object)) + shown = ", ".join(repr(value) for value in unique[:_MAX_REPORTED_EXAMPLES]) + if len(unique) > _MAX_REPORTED_EXAMPLES: + shown = f"{shown}, ... ({len(unique)} distinct)" + return shown + + +def _resolve_channel( + household: pd.DataFrame, + *, + declared_channel: str | None, + channel_column: str | None, +) -> tuple[pd.Series, str]: + """Return each household's provenance tag and where it came from. + + Resolution order is explicit and fail-closed: an operator-named column, the + base-spine tag, the assembled support channel, then a caller declaration + for a frame that carries no tag at all. A caller may not declare over a + frame that already declares for itself. + """ + present = [name for name in _CHANNEL_COLUMN_ORDER if name in household.columns] + if channel_column is not None: + _require( + channel_column in household.columns, + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"the household table carries no requested channel column " + f"{channel_column!r}; present: {sorted(household.columns)[:20]}.", + ) + _require( + declared_channel is None, + "SPM_UNIVERSE_UNDECLARED_SPINE", + "declared_channel and channel_column are mutually exclusive.", + ) + return household[channel_column].astype(object), channel_column + if present: + _require( + declared_channel is None, + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"the household table already declares its own provenance in " + f"{present[0]!r}; a caller declaration would override it.", + ) + return household[present[0]].astype(object), present[0] + _require( + declared_channel is not None, + "SPM_UNIVERSE_UNDECLARED_SPINE", + "the household table carries no provenance tag " + f"({' or '.join(_CHANNEL_COLUMN_ORDER)}) and the caller declared no " + "channel, so no universe ruling applies to these rows.", + ) + return ( + pd.Series(declared_channel, index=household.index, dtype=object), + f"declared:{declared_channel}", + ) + + +def _household_arms(channel: pd.Series) -> pd.Series: + arm = channel.map(SPM_UNIVERSE_CHANNEL_ARMS) + unruled = channel[arm.isna()] + _require( + unruled.empty, + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"{len(unruled)} household row(s) carry a provenance channel this " + f"producer has no universe ruling for: {_examples(unruled)}; ruled " + f"channels: {sorted(SPM_UNIVERSE_CHANNEL_ARMS)}.", + ) + return arm.astype(object) + + +def _household_kinds(household: pd.DataFrame, arm: pd.Series) -> pd.Series: + """Validate the ACS record type and its confinement to the ACS arm.""" + if ACS_HOUSEHOLD_KIND_COLUMN in household.columns: + kind = pd.to_numeric( + household[ACS_HOUSEHOLD_KIND_COLUMN], errors="coerce" + ).astype(float) + else: + kind = pd.Series(np.nan, index=household.index, dtype=float) + acs = arm.eq(ACS_ARM).to_numpy() + unknown = acs & ~kind.isin(ACS_HOUSEHOLD_KINDS).to_numpy() + if ACS_HOUSEHOLD_KIND_COLUMN in household.columns: + offending = _examples(household.loc[unknown, ACS_HOUSEHOLD_KIND_COLUMN]) + else: + offending = "" + _require( + not unknown.any(), + "SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND", + f"{int(unknown.sum())} ACS household row(s) carry a " + f"{ACS_HOUSEHOLD_KIND_COLUMN} outside {list(ACS_HOUSEHOLD_KINDS)} " + f"(missing counts as outside): {offending}.", + ) + # TYPEHUGQ == 1 means "ACS housing unit", not "not group quarters": ASEC + # rows carry it as missing by design, so a non-missing value on the ASEC + # arm is a bad merge, not a housing unit. + off_arm = (~acs) & kind.notna().to_numpy() + _require( + not off_arm.any(), + "SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND", + f"{int(off_arm.sum())} non-ACS household row(s) carry an ACS " + f"{ACS_HOUSEHOLD_KIND_COLUMN} value " + f"({_examples(kind[off_arm])}); the ACS record type does not " + "describe them and cannot be read as one.", + ) + return kind + + +def _kind_keys(arm: pd.Series, kind: pd.Series) -> pd.Series: + """One hashable ``(arm, record type)`` per household row.""" + return pd.Series( + [ + (str(a), None if pd.isna(k) else int(k)) + for a, k in zip(arm.to_numpy(), kind.to_numpy(), strict=True) + ], + index=arm.index, + dtype=object, + ) + + +def _assert_one_kind_per_household( + household_ids: pd.Series, kind_keys: pd.Series +) -> None: + """A household id resolving to more than one kind is a bad merge. + + Unreachable through a validated :class:`~microcosm.frame.Frame`, whose + constructor already refuses a duplicated group id. It is reachable when the + producer is handed tables directly, and picking a majority kind there would + be a fabricated universe decision. + """ + distinct = kind_keys.groupby(household_ids.to_numpy(), sort=False).nunique() + mixed = distinct[distinct > 1] + _require( + mixed.empty, + "SPM_UNIVERSE_MIXED_HOUSEHOLD", + f"{len(mixed)} household id(s) resolve to more than one " + f"(arm, record type): {_examples(mixed.index)}.", + ) + + +def _assert_group_quarters_are_single_native_records( + *, + household_ids: pd.Series, + arm: pd.Series, + kind: pd.Series, + person_household_ids: pd.Series, + clone_index: pd.Series | None, +) -> int: + """Mirror the stacked spine's one-native-person-per-GQ-placeholder rule. + + ACS group-quarters rows are one-person ``WGTP == 0`` placeholders at the + source (``acs_pums._occupied_households``), and the stacked assembly + re-asserts it. Clones of such a placeholder are expected and are not + counted here; more than one *native* person in one is not. + """ + gq = (arm.eq(ACS_ARM) & kind.isin(ACS_GROUP_QUARTERS_KINDS)).to_numpy() + gq_ids = pd.Index(household_ids.to_numpy()[gq]).unique() + if len(gq_ids) == 0: + return 0 + native = ( + clone_index.eq(_NATIVE_CLONE_INDEX).to_numpy() + if clone_index is not None + else np.ones(len(person_household_ids), dtype=bool) + ) + counts = ( + pd.Series(1, index=person_household_ids.to_numpy()[native]) + .groupby(level=0) + .sum() + .reindex(gq_ids, fill_value=0) + ) + crowded = counts[counts > 1] + _require( + crowded.empty, + "SPM_UNIVERSE_GQ_MULTI_PERSON", + f"{len(crowded)} ACS group-quarters household row(s) carry more than " + f"one native person: {_examples(crowded.index)}.", + ) + return int(len(gq_ids)) + + +def _assert_membership( + values: pd.Series, owner_ids: np.ndarray, *, column: str, owner: str +) -> None: + """Both directions of a membership join, independently of the frame kernel.""" + owner_index = pd.Index(owner_ids) + orphan = values[~pd.Index(values).isin(owner_index)] + _require( + orphan.empty, + "SPM_UNIVERSE_ORPHAN_MEMBER", + f"{len(orphan)} person row(s) carry a {column!r} value matching no " + f"{owner} id: {_examples(orphan)}.", + ) + if owner == "spm_unit": + referenced = pd.Index(pd.unique(values.to_numpy())) + empty = owner_index[~owner_index.isin(referenced)] + _require( + empty.empty, + "SPM_UNIVERSE_EMPTY_UNIT", + f"{len(empty)} {owner} id(s) have no member in {column!r}: " + f"{_examples(empty)}; an empty unit has no source record to read " + "a universe from.", + ) + + +def _assert_native_asec_partition( + *, + person: pd.DataFrame, + asec_person: np.ndarray, + unit_membership_column: str, +) -> None: + """Refuse a silently degraded SPM partition on the ASEC arm. + + ``assign_us_unit_structure`` uses the native ``SPM_ID`` when it is fully + present and falls back to ``household_id`` otherwise — and *any* missing + ``SPM_ID`` triggers that fallback for the whole frame, while ``asec_pool`` + deliberately tolerates a missing ``SPM_ID`` when globalizing. Declaring + ``INCLUDED`` over a degraded partition would attest to a unit structure + that does not exist, so the ASEC arm must prove its partition is the + native one. + + The check is deliberately ASEC-only. ACS PUMS supplies no ``SPM_ID`` at + all, so the ACS arm reaches ``assign_us_unit_structure`` without one and + takes the household fallback by construction — there, one SPM unit per + household *is* the native partition and carries no information. + """ + if not asec_person.any(): + return + _require( + ASEC_NATIVE_UNIT_COLUMN in person.columns, + "SPM_UNIVERSE_DEGRADED_PARTITION", + f"the person table carries no {ASEC_NATIVE_UNIT_COLUMN!r}, so the ASEC " + "arm's SPM partition cannot be shown to be the native one rather than " + "the documented household fallback.", + ) + native = person.loc[asec_person, ASEC_NATIVE_UNIT_COLUMN] + missing = int(native.isna().sum()) + _require( + missing == 0, + "SPM_UNIVERSE_DEGRADED_PARTITION", + f"{missing} ASEC person row(s) carry a missing " + f"{ASEC_NATIVE_UNIT_COLUMN}; any missing value silently converts every " + "SPM unit in the frame into a household.", + ) + units = person.loc[asec_person, unit_membership_column] + pairs = pd.MultiIndex.from_arrays([native.to_numpy(), units.to_numpy()]) + _require( + len(pairs.unique()) == native.nunique() == units.nunique(), + "SPM_UNIVERSE_DEGRADED_PARTITION", + f"the ASEC arm's {unit_membership_column!r} is not a one-to-one " + f"densification of {ASEC_NATIVE_UNIT_COLUMN} " + f"({native.nunique()} native unit(s), {units.nunique()} frame unit(s), " + f"{len(pairs.unique())} distinct pairing(s)), so the native partition " + "did not survive assembly.", + ) + + +def _assert_asec_record_type_unreviewed( + *, household: pd.DataFrame, person: pd.DataFrame, asec_any: bool +) -> None: + """Guard the spine-level ASEC ruling against a vintage that outgrows it.""" + if not asec_any: + return + found = sorted( + field + for field in ASEC_RECORD_TYPE_FIELDS + if field in household.columns or field in person.columns + ) + _require( + not found, + "SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED", + f"the frame carries ASEC household record-type field(s) {found}. The " + f"ASEC ruling ({ASEC_SPM_INCLUDED_AUTHORITY}) was made at the spine " + "precisely because Microcosm reads none of these; it must be " + "re-derived against the field rather than applied over it.", + ) + + +def classify_spm_universe( + *, + household: pd.DataFrame, + person: pd.DataFrame, + unit_ids: np.ndarray, + household_id_column: str = "household_id", + household_membership_column: str = "person_household_id", + unit_membership_column: str = "person_spm_unit_id", + declared_channel: str | None = None, + channel_column: str | None = None, +) -> SpmUniverseClassification: + """Classify every SPM unit's measurement universe from source record type. + + Args: + household: The household table, carrying its provenance tag and, on + the ACS arm, ``TYPEHUGQ``. + person: The person table, carrying household and SPM-unit membership. + unit_ids: The SPM unit ids to classify, in the order the caller wants + the returned status. + household_id_column: Id column on ``household``. + household_membership_column: Person column linking to ``household``. + unit_membership_column: Person column linking to the SPM unit table. + declared_channel: Universe channel for a frame that carries no + provenance tag of its own. Refused when the frame does carry one. + channel_column: Explicit provenance column to read instead of the + default resolution order. + + Returns: + The per-unit status as member-name strings, the household-level + evidence it was derived from, and a provenance dict. + + Raises: + ValueError: Any of :data:`SPM_UNIVERSE_REFUSALS`. Every refusal is + fail-closed: this producer never resolves an ambiguity by guessing, + and never writes ``UNRESOLVED`` to mean "I could not tell". + """ + for table, columns, label in ( + (household, (household_id_column,), "household"), + (person, (household_membership_column, unit_membership_column), "person"), + ): + missing = sorted(set(columns) - set(table.columns)) + _require( + not missing, + "SPM_UNIVERSE_ORPHAN_MEMBER", + f"the {label} table is missing required column(s) {missing}.", + ) + + unit_ids = np.asarray(unit_ids) + household_ids = household[household_id_column] + channel, channel_source = _resolve_channel( + household, declared_channel=declared_channel, channel_column=channel_column + ) + arm = _household_arms(channel) + kind = _household_kinds(household, arm) + kind_keys = _kind_keys(arm, kind) + _assert_one_kind_per_household(household_ids, kind_keys) + + person_household_ids = person[household_membership_column] + person_units = person[unit_membership_column] + _assert_membership( + person_household_ids, + household_ids.to_numpy(), + column=household_membership_column, + owner="household", + ) + _assert_membership( + person_units, unit_ids, column=unit_membership_column, owner="spm_unit" + ) + + clone_index = ( + pd.to_numeric(person[_PERSON_CLONE_INDEX_COLUMN], errors="coerce") + if _PERSON_CLONE_INDEX_COLUMN in person.columns + else None + ) + group_quarters_households = _assert_group_quarters_are_single_native_records( + household_ids=household_ids, + arm=arm, + kind=kind, + person_household_ids=person_household_ids, + clone_index=clone_index, + ) + + kind_by_household = pd.Series( + kind_keys.to_numpy(), index=pd.Index(household_ids.to_numpy()) + ) + # Every duplicate id resolves to one kind by now, so dropping repeats + # cannot change a lookup; it only keeps ``map`` from fanning rows out. + kind_by_household = kind_by_household[~kind_by_household.index.duplicated()] + person_kind = person_household_ids.map(kind_by_household) + asec_person = np.asarray( + [key[0] == ASEC_ARM for key in person_kind.to_numpy()], dtype=bool + ) + _assert_asec_record_type_unreviewed( + household=household, person=person, asec_any=bool(asec_person.any()) + ) + _assert_native_asec_partition( + person=person, + asec_person=asec_person, + unit_membership_column=unit_membership_column, + ) + + by_unit = pd.DataFrame( + {"unit": person_units.to_numpy(), "kind": person_kind.to_numpy()} + ) + distinct = by_unit.groupby("unit", sort=False)["kind"].nunique() + spanning = distinct[distinct > 1] + _require( + spanning.empty, + "SPM_UNIVERSE_UNIT_SPANS_KINDS", + f"{len(spanning)} SPM unit(s) have members in households of different " + f"(arm, record type): {_examples(spanning.index)}. Picking a majority " + "kind would be a fabricated universe decision.", + ) + + unit_kind = ( + by_unit.groupby("unit", sort=False)["kind"].first().reindex(pd.Index(unit_ids)) + ) + status = np.asarray( + [ + OUTSIDE + if key[0] == ACS_ARM and key[1] in ACS_GROUP_QUARTERS_KINDS + else INCLUDED + for key in unit_kind.to_numpy() + ], + dtype=object, + ) + # A producer that cannot decide refuses above; UNRESOLVED is the engine's + # absence sentinel and must never be emitted. + _require( + UNRESOLVED not in set(status), + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"the classification emitted {UNRESOLVED}, which is the engine's " + "absence sentinel rather than a producer value.", + ) + + arm_counts = arm.value_counts() + kind_counts = { + int(value): int(count) + for value, count in kind.value_counts(dropna=True).items() + } + provenance = { + "declared_variable": UNIVERSE_INPUT, + "declared_entity": "spm_unit", + "declared_period": "YEAR", + "stored_encoding": "enum member name string", + "profile": SPM_UNIVERSE_PROFILE_NAME, + "rule": SPM_UNIVERSE_RULE, + "acs_authority": ACS_SPM_OUTSIDE_AUTHORITY, + "asec_authority": ASEC_SPM_INCLUDED_AUTHORITY, + "methodology": SPM_UNIVERSE_METHODOLOGY, + "evidence_status": ( + "Source record type read from the frame's own household table " + "(ACS TYPEHUGQ), plus a spine-level ruling for the ASEC arm whose " + "sample frame equals the CPS SPM universe. No composition, age, " + "tenure, geography or role is read." + ), + "channel_source": channel_source, + "channel_arms": dict(SPM_UNIVERSE_CHANNEL_ARMS), + "household_rows": int(len(household)), + "households_by_arm": {str(k): int(v) for k, v in arm_counts.items()}, + "households_by_acs_record_type": kind_counts, + "acs_group_quarters_households": group_quarters_households, + "person_rows": int(len(person)), + "spm_units": int(len(unit_ids)), + "units_included": int((status == INCLUDED).sum()), + "units_outside": int((status == OUTSIDE).sum()), + "units_unresolved": 0, + "refusals": list(SPM_UNIVERSE_REFUSALS), + "composition_read": False, + "weights_used": False, + "ages_changed": False, + "is_calibration_target": False, + "is_spine_agreement_surface": False, + "period_handling": ( + "The declaration is a construction fact about the source record, " + "constant across the years a build requests." + ), + } + return SpmUniverseClassification( + status=status, + unit_ids=unit_ids, + household_arm=arm, + household_kind=kind, + provenance=provenance, + ) + + +def attach_spm_universe_status( + frame: Any, + *, + unit_entity: str = "spm_unit", + declared_channel: str | None = None, + channel_column: str | None = None, +) -> SpmUniverseSourceResult: + """Attach :data:`UNIVERSE_INPUT` to ``frame``'s SPM unit table. + + The status is stored as the enum **member name** — ``"OUTSIDE"``, not + ``2``, not ``"SPMUniverseStatus.OUTSIDE"``, and not the label text — + because the export adapter normalizes a stored cell to a member name and + validates it cell-by-cell against the engine's own enum domain. + + Nothing else about the frame changes: no weight, no membership, no row. + + Args: + frame: A validated :class:`~microcosm.frame.Frame`. + unit_entity: The SPM unit entity. + declared_channel: See :func:`classify_spm_universe`. + channel_column: See :func:`classify_spm_universe`. + + Returns: + A new frame carrying the column, and the classification behind it. + + Raises: + ValueError: Any of :data:`SPM_UNIVERSE_REFUSALS`, including + ``SPM_UNIVERSE_COLUMN_EXISTS`` when the unit table already carries + the declaration — a producer overwriting an existing declaration + would silently replace someone else's source ruling. + """ + from microcosm.frame import Frame + + schema = frame.schema + unit_table = frame.table(unit_entity) + _require( + UNIVERSE_INPUT not in unit_table.columns, + "SPM_UNIVERSE_COLUMN_EXISTS", + f"the {unit_entity} table already carries {UNIVERSE_INPUT!r}; this " + "producer never overwrites an existing universe declaration.", + ) + classification = classify_spm_universe( + household=frame.table("household"), + person=frame.table(schema.person_entity), + unit_ids=unit_table[schema.id_column(unit_entity)].to_numpy(), + household_id_column=schema.id_column("household"), + household_membership_column=schema.membership_column("household"), + unit_membership_column=schema.membership_column(unit_entity), + declared_channel=declared_channel, + channel_column=channel_column, + ) + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables.update({name: frame.link(name).copy() for name in frame.links}) + tables[unit_entity][UNIVERSE_INPUT] = classification.status + declared = 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, + ) + return SpmUniverseSourceResult(frame=declared, classification=classification) diff --git a/packages/microcosm-build/tests/test_us_spine_blindness.py b/packages/microcosm-build/tests/test_us_spine_blindness.py index 8462403f0f..256461d84b 100644 --- a/packages/microcosm-build/tests/test_us_spine_blindness.py +++ b/packages/microcosm-build/tests/test_us_spine_blindness.py @@ -107,6 +107,10 @@ "puf_support.py", # Validates provenance at the clone boundary. "spine_agreement.py", # Pre-calibration distribution comparison. "spine_assembly.py", # New pre-operator assembly seam. + # Declares the SPM measurement universe per source arm. The ACS and + # ASEC rulings differ by construction, so the provenance tag is the + # producer's input rather than something it must be blind to. + "spm_universe_source.py", # Stacked-spine pilot (#578 revision): stacking, gap-fill donor # routing, activation authority, the completeness gate, and the # by-origin battery are origin-aware by charter. @@ -297,6 +301,9 @@ # Pinned ASEC role reconstruction and exact parent join; no population # treatment. Remains subject to the all-runtime source-identity scan. "spm_role_source.py", + # Source-owned SPM measurement-universe declaration; reads the source + # record type and no population attribute. Provenance owner above. + "spm_universe_source.py", "stacked_spine.py", # Provenance owner (#578 revision); see owners list. "support_provenance.py", "take_up.py", diff --git a/packages/microcosm-build/tests/test_us_spm_universe_source.py b/packages/microcosm-build/tests/test_us_spm_universe_source.py new file mode 100644 index 0000000000..c2a1f420d1 --- /dev/null +++ b/packages/microcosm-build/tests/test_us_spm_universe_source.py @@ -0,0 +1,647 @@ +"""The SPM measurement-universe producer: the rule, and every refusal. + +Every fixture here is invented. Nothing in this file opens a build artifact, +a pinned source, or a release. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.us_runtime.spm_universe_source import ( + ACS_ARM, + ACS_GROUP_QUARTERS_KINDS, + ACS_HOUSEHOLD_KIND_COLUMN, + ACS_HOUSING_UNIT_KIND, + ASEC_ARM, + ASEC_NATIVE_UNIT_COLUMN, + HOUSEHOLD_SPINE_COLUMN, + HOUSEHOLD_SUPPORT_CHANNEL_COLUMN, + INCLUDED, + OUTSIDE, + SPM_UNIVERSE_CHANNEL_ARMS, + SPM_UNIVERSE_REFUSALS, + UNIVERSE_INPUT, + UNRESOLVED, + attach_spm_universe_status, + classify_spm_universe, +) +from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights + +_US_RUNTIME = ( + Path(__file__).resolve().parents[1] / "src" / "microcosm" / "build" / "us_runtime" +) +_ACS_CHANNEL = "acs_2024_1yr" +_ASEC_CHANNEL = "asec_puf" + + +def _tables( + persons: list[dict[str, object]], + *, + channel_column: str = HOUSEHOLD_SPINE_COLUMN, + household_extra: dict[int, dict[str, object]] | None = None, + person_extra_columns: dict[str, list[object]] | None = None, + extra_household_rows: list[dict[str, object]] | None = None, + extra_unit_ids: list[int] | None = None, +) -> tuple[pd.DataFrame, pd.DataFrame, np.ndarray]: + """Household, person and SPM unit ids from a list of invented persons. + + Each person dict carries ``hid`` (household), ``unit`` (SPM unit), + ``chan`` (provenance tag), and optionally ``kind`` (``TYPEHUGQ``), + ``spm_id`` (the ASEC native unit id) and ``clone`` (clone index). + """ + prows = [] + households: dict[int, dict[str, object]] = {} + for index, spec in enumerate(persons, start=1): + hid = int(spec["hid"]) + row: dict[str, object] = { + "person_id": index, + "person_household_id": hid, + "person_spm_unit_id": int(spec["unit"]), + "person_tax_unit_id": hid, + "person_family_id": hid, + "person_marital_unit_id": index, + "person_support_clone_index": int(spec.get("clone", 0)), + } + if "spm_id" in spec: + row[ASEC_NATIVE_UNIT_COLUMN] = spec["spm_id"] + prows.append(row) + household = {"household_id": hid, channel_column: spec["chan"]} + if "kind" in spec: + household[ACS_HOUSEHOLD_KIND_COLUMN] = spec["kind"] + households.setdefault(hid, household) + person = pd.DataFrame(prows) + if person_extra_columns: + for name, values in person_extra_columns.items(): + person[name] = values + household_rows = [households[hid] for hid in sorted(households)] + if household_extra: + for row in household_rows: + row.update(household_extra.get(int(row["household_id"]), {})) + if extra_household_rows: + household_rows.extend(extra_household_rows) + household = pd.DataFrame(household_rows) + if ACS_HOUSEHOLD_KIND_COLUMN in household.columns: + household[ACS_HOUSEHOLD_KIND_COLUMN] = household[ + ACS_HOUSEHOLD_KIND_COLUMN + ].astype(object) + unit_ids = np.asarray( + sorted({int(spec["unit"]) for spec in persons} | set(extra_unit_ids or [])) + ) + return household, person, unit_ids + + +def _classify(persons: list[dict[str, object]], **kwargs) -> dict[int, str]: + table_kwargs = { + key: kwargs.pop(key) + for key in ( + "channel_column", + "household_extra", + "person_extra_columns", + "extra_household_rows", + "extra_unit_ids", + ) + if key in kwargs + } + household, person, unit_ids = _tables(persons, **table_kwargs) + result = classify_spm_universe( + household=household, person=person, unit_ids=unit_ids, **kwargs + ) + return dict(zip(result.unit_ids.tolist(), result.status.tolist(), strict=True)) + + +def _frame(persons: list[dict[str, object]], **kwargs) -> Frame: + household, person, unit_ids = _tables(persons, **kwargs) + household_ids = household["household_id"].to_numpy() + return Frame( + { + "person": person, + "household": household, + "tax_unit": pd.DataFrame({"tax_unit_id": household_ids}), + "spm_unit": pd.DataFrame({"spm_unit_id": unit_ids}), + "family": pd.DataFrame({"family_id": household_ids}), + "marital_unit": pd.DataFrame( + {"marital_unit_id": person["person_id"].to_numpy()} + ), + }, + US_SCHEMA, + { + "household": Weights( + np.full(len(household_ids), 100.0), WeightKind.CALIBRATED + ) + }, + metadata={"stage": "invented"}, + ) + + +# -------------------------------------------------------------------------- +# The rule +# -------------------------------------------------------------------------- + + +def test_acs_housing_unit_is_included(): + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + ] + ) + assert statuses == {1: INCLUDED} + + +@pytest.mark.parametrize("kind", ACS_GROUP_QUARTERS_KINDS) +def test_acs_group_quarters_are_outside_for_each_code(kind): + statuses = _classify([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": kind}]) + assert statuses == {1: OUTSIDE} + + +def test_asec_records_are_included(): + """The ASEC ruling is spine-level: no record-type field is consulted.""" + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12}, + {"hid": 2, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 13}, + ] + ) + assert statuses == {1: INCLUDED, 2: INCLUDED, 3: INCLUDED} + + +def test_a_zero_adult_housing_unit_is_included(): + """The anti-inference test. + + A housing unit whose only members are children is a household-unit data + defect, not an out-of-universe record. Marking it ``OUTSIDE`` would hide + the defect and make the universe move whenever the role column moved. Only + the record type decides: the identical composition under ``TYPEHUGQ`` 2 is + ``OUTSIDE``, and it is the record type that moved it. + """ + children = [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + ] + ages = {"age": [3, 7], "is_spm_independent_minor_role": [False, False]} + assert _classify(children, person_extra_columns=ages) == {1: INCLUDED} + + group_quarters = [dict(spec, kind=2) for spec in children[:1]] + assert _classify(group_quarters, person_extra_columns={"age": [3]}) == {1: OUTSIDE} + + +def test_the_declaration_is_invariant_to_age_and_role(): + """Composition cannot move the answer, in either direction.""" + persons = [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + ] + children = _classify( + persons, + person_extra_columns={ + "age": [2, 3], + "is_spm_independent_minor_role": [False, False], + }, + ) + adults = _classify( + persons, + person_extra_columns={ + "age": [44, 51], + "is_spm_independent_minor_role": [True, True], + }, + ) + assert children == adults == {1: INCLUDED, 2: OUTSIDE} + + +def test_a_clone_inherits_its_native_rows_status(): + """Cloning a record carries its record type, so it carries its universe.""" + statuses = _classify( + [ + # Native group-quarters placeholder and its clone. + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2, "clone": 1}, + # Native housing unit and its clone. + {"hid": 3, "unit": 3, "chan": _ACS_CHANNEL, "kind": 1, "clone": 0}, + {"hid": 4, "unit": 4, "chan": _ACS_CHANNEL, "kind": 1, "clone": 1}, + ] + ) + assert statuses == {1: OUTSIDE, 2: OUTSIDE, 3: INCLUDED, 4: INCLUDED} + + +def test_a_pooled_frame_declares_each_arm_from_its_own_rule(): + household, person, unit_ids = _tables( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 3}, + {"hid": 3, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 31}, + ] + ) + result = classify_spm_universe( + household=household, person=person, unit_ids=unit_ids + ) + assert result.status.tolist() == [INCLUDED, OUTSIDE, INCLUDED] + assert result.provenance["households_by_arm"] == {ACS_ARM: 2, ASEC_ARM: 1} + assert result.provenance["households_by_acs_record_type"] == {1: 1, 3: 1} + assert result.provenance["acs_group_quarters_households"] == 1 + assert result.provenance["units_included"] == 2 + assert result.provenance["units_outside"] == 1 + assert result.provenance["units_unresolved"] == 0 + + +def test_the_support_channel_column_is_read_when_no_spine_tag_is_present(): + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": "acs", "kind": 1}, + {"hid": 2, "unit": 2, "chan": "asec", "spm_id": 21}, + ], + channel_column=HOUSEHOLD_SUPPORT_CHANNEL_COLUMN, + ) + assert statuses == {1: INCLUDED, 2: INCLUDED} + + +def test_an_untagged_table_accepts_an_explicit_caller_declaration(): + household = pd.DataFrame( + {"household_id": [1], ACS_HOUSEHOLD_KIND_COLUMN: [2]}, + ) + person = pd.DataFrame( + {"person_id": [1], "person_household_id": [1], "person_spm_unit_id": [1]} + ) + result = classify_spm_universe( + household=household, + person=person, + unit_ids=np.asarray([1]), + declared_channel=_ACS_CHANNEL, + ) + assert result.status.tolist() == [OUTSIDE] + assert result.provenance["channel_source"] == f"declared:{_ACS_CHANNEL}" + + +# -------------------------------------------------------------------------- +# Stored encoding +# -------------------------------------------------------------------------- + + +def test_the_status_is_stored_as_the_enum_member_name(): + household, person, unit_ids = _tables( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + ] + ) + status = classify_spm_universe( + household=household, person=person, unit_ids=unit_ids + ).status + assert status.dtype == np.dtype(object) + assert status.tolist() == ["INCLUDED", "OUTSIDE"] + # Negative controls: not the integer index, not repr(), not the label. + assert 2 not in set(status.tolist()) + assert "SPMUniverseStatus.OUTSIDE" not in set(status.tolist()) + assert not any(" " in value for value in status.tolist()) + # UNRESOLVED is the engine's absence sentinel; a producer never emits it. + assert UNRESOLVED not in set(status.tolist()) + + +# -------------------------------------------------------------------------- +# Refusals +# -------------------------------------------------------------------------- + + +def _refuses(code: str): + return pytest.raises(ValueError, match=rf"^{code}: ") + + +def test_refuses_a_preexisting_declaration(): + frame = _frame([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}]) + declared = attach_spm_universe_status(frame).frame + with _refuses("SPM_UNIVERSE_COLUMN_EXISTS"): + attach_spm_universe_status(declared) + + +def test_refuses_a_channel_with_no_universe_ruling(): + with _refuses("SPM_UNIVERSE_UNDECLARED_SPINE"): + _classify([{"hid": 1, "unit": 1, "chan": "puf_tax_detail", "kind": 1}]) + + +def test_refuses_a_frame_with_no_provenance_tag_and_no_declaration(): + household = pd.DataFrame({"household_id": [1], ACS_HOUSEHOLD_KIND_COLUMN: [1]}) + person = pd.DataFrame( + {"person_id": [1], "person_household_id": [1], "person_spm_unit_id": [1]} + ) + with _refuses("SPM_UNIVERSE_UNDECLARED_SPINE"): + classify_spm_universe( + household=household, person=person, unit_ids=np.asarray([1]) + ) + + +def test_refuses_a_caller_declaration_over_a_tagged_frame(): + with _refuses("SPM_UNIVERSE_UNDECLARED_SPINE"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + declared_channel=_ASEC_CHANNEL, + ) + + +@pytest.mark.parametrize("kind", [4, 0, -1, None, float("nan")]) +def test_refuses_an_unknown_acs_record_type(kind): + with _refuses("SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND"): + _classify([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": kind}]) + + +def test_refuses_an_acs_record_type_on_a_non_acs_row(): + """``TYPEHUGQ == 1`` means "ACS housing unit", never "not group quarters".""" + with _refuses("SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "kind": 1, "spm_id": 11}] + ) + + +def test_refuses_a_household_resolving_to_more_than_one_kind(): + household = pd.DataFrame( + { + "household_id": [1, 1], + HOUSEHOLD_SPINE_COLUMN: [_ACS_CHANNEL, _ACS_CHANNEL], + ACS_HOUSEHOLD_KIND_COLUMN: [1, 2], + } + ) + person = pd.DataFrame( + {"person_id": [1], "person_household_id": [1], "person_spm_unit_id": [1]} + ) + with _refuses("SPM_UNIVERSE_MIXED_HOUSEHOLD"): + classify_spm_universe( + household=household, person=person, unit_ids=np.asarray([1]) + ) + + +def test_refuses_a_group_quarters_placeholder_with_two_native_persons(): + with _refuses("SPM_UNIVERSE_GQ_MULTI_PERSON"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + ] + ) + + +def test_a_group_quarters_placeholder_with_a_clone_is_accepted(): + """Negative control for the refusal above: clones are expected.""" + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 1}, + ] + ) + assert statuses == {1: OUTSIDE} + + +def test_refuses_a_member_of_no_declared_unit(): + with _refuses("SPM_UNIVERSE_ORPHAN_MEMBER"): + household, person, _ = _tables( + [{"hid": 1, "unit": 7, "chan": _ACS_CHANNEL, "kind": 1}] + ) + classify_spm_universe( + household=household, person=person, unit_ids=np.asarray([1]) + ) + + +def test_refuses_a_member_of_no_declared_household(): + household, person, unit_ids = _tables( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}] + ) + person["person_household_id"] = [9] + with _refuses("SPM_UNIVERSE_ORPHAN_MEMBER"): + classify_spm_universe(household=household, person=person, unit_ids=unit_ids) + + +def test_refuses_a_unit_with_no_members(): + with _refuses("SPM_UNIVERSE_EMPTY_UNIT"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + extra_unit_ids=[2], + ) + + +def test_refuses_a_unit_spanning_two_record_types(): + """Natively unreachable; a majority vote here would fabricate a universe.""" + with _refuses("SPM_UNIVERSE_UNIT_SPANS_KINDS"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2}, + ] + ) + + +def test_refuses_a_unit_spanning_two_arms(): + with _refuses("SPM_UNIVERSE_UNIT_SPANS_KINDS"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 21}, + ] + ) + + +def test_refuses_an_asec_arm_with_no_native_unit_column(): + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify([{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL}]) + + +def test_refuses_a_missing_native_unit_id_on_the_asec_arm(): + """One missing ``SPM_ID`` degrades every SPM unit in the frame.""" + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 2, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": np.nan}, + ] + ) + + +def test_refuses_an_asec_partition_that_is_not_the_native_one(): + """Two native units collapsed into one frame unit: the fallback's shape.""" + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 12}, + ] + ) + + +def test_the_degraded_partition_check_does_not_fire_on_the_acs_arm(): + """ACS PUMS supplies no ``SPM_ID``; one unit per household is native there.""" + assert _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 1}, + ] + ) == {1: INCLUDED, 2: INCLUDED} + + +@pytest.mark.parametrize("field", ["H_TYPE", "HRHTYPE", "H_HHTYPE", "HUNITS"]) +def test_refuses_an_asec_vintage_carrying_a_record_type_field(field): + with _refuses("SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}], + household_extra={1: {field: 1}}, + ) + + +def test_an_acs_only_frame_is_unaffected_by_the_asec_record_type_guard(): + """Negative control: the guard is scoped to the arm whose ruling it guards.""" + assert _classify( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + household_extra={1: {"H_LIVQRT": 1}}, + ) == {1: INCLUDED} + + +# -------------------------------------------------------------------------- +# The frame wrapper +# -------------------------------------------------------------------------- + + +def test_attach_declares_the_status_and_changes_nothing_else(): + frame = _frame( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + {"hid": 3, "unit": 3, "chan": _ACS_CHANNEL, "kind": 3}, + ] + ) + result = attach_spm_universe_status(frame) + declared = result.frame + + assert declared.table("spm_unit")[UNIVERSE_INPUT].tolist() == [ + INCLUDED, + OUTSIDE, + OUTSIDE, + ] + # The input frame is untouched: this is a new frame, not a mutation. + assert UNIVERSE_INPUT not in frame.table("spm_unit").columns + # No weight, no membership, no row moves. + assert declared.weights_for("household").total == pytest.approx( + frame.weights_for("household").total + ) + for entity in frame.entities: + assert declared.n(entity) == frame.n(entity) + before = frame.table(entity) + after = declared.table(entity) + assert set(after.columns) - set(before.columns) <= {UNIVERSE_INPUT} + pd.testing.assert_frame_equal(after[before.columns], before) + assert declared.metadata == frame.metadata + assert declared.mass_log == frame.mass_log + + +def test_attach_reports_the_classification_it_declared(): + frame = _frame( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + ] + ) + result = attach_spm_universe_status(frame) + assert result.status.tolist() == [INCLUDED, OUTSIDE] + assert result.provenance["declared_variable"] == UNIVERSE_INPUT + assert result.provenance["declared_entity"] == "spm_unit" + assert result.provenance["stored_encoding"] == "enum member name string" + + +def test_the_provenance_states_what_the_producer_is_not(): + frame = _frame([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}]) + provenance = attach_spm_universe_status(frame).provenance + assert provenance["weights_used"] is False + assert provenance["ages_changed"] is False + assert provenance["composition_read"] is False + assert provenance["is_calibration_target"] is False + assert provenance["is_spine_agreement_surface"] is False + + +# -------------------------------------------------------------------------- +# Constants parity and refusal-code completeness +# -------------------------------------------------------------------------- + + +def _module_constant(module: str, name: str) -> str: + """Read one module-level string constant without importing the module. + + ``acs_pums`` and ``stacked_spine`` both pull the country engine and cost + tens of seconds to import; parsing is exact and free. + """ + tree = ast.parse((_US_RUNTIME / module).read_text()) + for node in tree.body: + if isinstance(node, ast.Assign) and any( + isinstance(target, ast.Name) and target.id == name + for target in node.targets + ): + assert isinstance(node.value, ast.Constant), (module, name) + return node.value.value + raise AssertionError(f"{module} declares no module-level {name}") + + +@pytest.mark.parametrize( + ("module", "constant", "arm"), + [ + ("acs_pums.py", "ACS_2024_1YR_SPINE", ACS_ARM), + ("stacked_spine.py", "ACS_STACKED_SUPPORT_CHANNEL", ACS_ARM), + ("base_pool.py", "ASEC_PUF_SPINE", ASEC_ARM), + ("support_provenance.py", "BASE_ASEC_SUPPORT_CHANNEL", ASEC_ARM), + ], +) +def test_every_ruled_channel_matches_its_canonical_definition(module, constant, arm): + value = _module_constant(module, constant) + assert SPM_UNIVERSE_CHANNEL_ARMS.get(value) == arm, ( + f"{module}:{constant} == {value!r}, which this producer rules as " + f"{SPM_UNIVERSE_CHANNEL_ARMS.get(value)!r} rather than {arm!r}" + ) + + +def test_the_ruling_table_declares_nothing_beyond_the_canonical_channels(): + canonical = { + _module_constant("acs_pums.py", "ACS_2024_1YR_SPINE"), + _module_constant("stacked_spine.py", "ACS_STACKED_SUPPORT_CHANNEL"), + _module_constant("base_pool.py", "ASEC_PUF_SPINE"), + _module_constant("support_provenance.py", "BASE_ASEC_SUPPORT_CHANNEL"), + } + assert set(SPM_UNIVERSE_CHANNEL_ARMS) == canonical + + +def test_the_spine_column_name_matches_its_canonical_factory(): + tree = ast.parse((_US_RUNTIME / "base_pool.py").read_text()) + factory = next( + node + for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "spine_column" + ) + template = next( + node for node in ast.walk(factory) if isinstance(node, ast.JoinedStr) + ) + suffix = "".join( + part.value for part in template.values if isinstance(part, ast.Constant) + ) + assert HOUSEHOLD_SPINE_COLUMN == f"household{suffix}" + + +def test_every_refusal_the_module_raises_is_declared(): + """No refusal code may exist only inside a message string.""" + source = (_US_RUNTIME / "spm_universe_source.py").read_text() + tree = ast.parse(source) + raised = set() + for node in ast.walk(tree): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "_require" + and len(node.args) >= 2 + and isinstance(node.args[1], ast.Constant) + ): + raised.add(node.args[1].value) + assert raised, "the producer raises no named refusal at all" + assert raised <= set(SPM_UNIVERSE_REFUSALS), sorted( + raised - set(SPM_UNIVERSE_REFUSALS) + ) + assert len(SPM_UNIVERSE_REFUSALS) == len(set(SPM_UNIVERSE_REFUSALS)) + # SPM_UNIVERSE_COLUMN_EXISTS is raised by the frame wrapper, every other + # declared code by the classifier; together they must cover the tuple. + assert raised == set(SPM_UNIVERSE_REFUSALS) From 119effed95719212de027938f29a4d573d57cc93 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 22 Sep 2026 19:06:58 -0400 Subject: [PATCH 2/4] Key the ASEC native-partition check on the support-clone copy The PUF support clone (puf_support._clone_entity_table) deep-copies the person table and remaps only id and membership columns, so clone copy 1 carries the native SPM_ID under new SPM unit ids. Keyed on SPM_ID alone, SPM_UNIVERSE_DEGRADED_PARTITION refused every support-cloned ASEC frame, including the real operator's output on invented rows. The native unit key is now (clone copy, SPM_ID); degraded copies still refuse. A missing person clone index is now read as the native copy in both the ASEC partition check and the one-native-person group-quarters check, so a pool that lost the index on one arm can no longer pass the GQ check vacuously. SPM_UNIVERSE_COLUMN_EXISTS now covers every entity table, not only the SPM unit table. Tests: the real clone operator's output (requires_us), clone-copy negative controls, missing-index cases, channel precedence and explicit channel_column refusals, and every ASEC record-type field on both tables. Docs record #976's merge and Max's 2026-09-22 rulings; the ASEC authority constant records the sign-off. Co-Authored-By: Claude Opus 5.5 --- docs/us-spm-measurement-universe.md | 43 +++- .../build/us_runtime/spm_universe_source.py | 102 +++++++--- .../tests/test_us_spm_universe_source.py | 192 +++++++++++++++++- 3 files changed, 295 insertions(+), 42 deletions(-) diff --git a/docs/us-spm-measurement-universe.md b/docs/us-spm-measurement-universe.md index 00d9755198..61be3efba5 100644 --- a/docs/us-spm-measurement-universe.md +++ b/docs/us-spm-measurement-universe.md @@ -69,7 +69,7 @@ Every refusal is a named, fail-closed `ValueError`: | Code | Condition | | --- | --- | -| `SPM_UNIVERSE_COLUMN_EXISTS` | the SPM unit table already carries a declaration | +| `SPM_UNIVERSE_COLUMN_EXISTS` | any entity table already carries a declaration | | `SPM_UNIVERSE_UNDECLARED_SPINE` | a provenance channel with no universe ruling, an untagged frame with no caller declaration, or a caller declaration over a frame that already declares | | `SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND` | an ACS household whose `TYPEHUGQ` ∉ {1, 2, 3}, missing included | | `SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND` | a non-ACS household carrying an ACS record type | @@ -78,7 +78,7 @@ Every refusal is a named, fail-closed `ValueError`: | `SPM_UNIVERSE_ORPHAN_MEMBER` | a membership id matching no owning entity row | | `SPM_UNIVERSE_EMPTY_UNIT` | an SPM unit with no member | | `SPM_UNIVERSE_UNIT_SPANS_KINDS` | one SPM unit whose members sit in households of different kinds | -| `SPM_UNIVERSE_DEGRADED_PARTITION` | the ASEC arm's SPM partition is not the native one | +| `SPM_UNIVERSE_DEGRADED_PARTITION` | the ASEC arm's SPM partition is not the native one: `SPM_ID` absent, missing, or not one-to-one with the frame's units per support-clone copy | | `SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED` | the frame carries an ASEC record-type field the spine-level ruling was not derived against | `SPM_UNIVERSE_UNIT_SPANS_KINDS` is natively unreachable — ACS group-quarters @@ -98,6 +98,16 @@ ACS arm reaches `assign_us_unit_structure` without one and takes the household fallback by construction — there, one SPM unit per household *is* the native partition and carries no information. +The native unit key is `(support-clone copy, SPM_ID)`, not `SPM_ID` alone. +The PUF support clone (`puf_support._clone_entity_table`) deep-copies the +person table and remaps only the id and membership columns, so clone copy 1 +carries the native `SPM_ID` under new SPM unit ids; keyed on `SPM_ID` alone, +every support-cloned ASEC frame would read as degraded. +`test_the_real_support_clone_operator_output_is_accepted` runs the real clone +operator on invented rows to pin that. A missing clone index is read as the +native copy, which can only add collisions, so it never relaxes this check or +the one-native-person group-quarters check. + ## What is wired, and what is not **Wired.** The producer, its refusals, and its tests. It is classified in the @@ -116,14 +126,17 @@ it must be blind to). - **No enrichment-lane carry.** `microcosm.data.h5_enrichment` writes one Boolean person column at the HDF type level; carrying an `spm_unit`-entity string column is a separate, larger piece of work (design note §3). -- **No consumer change.** `check_spm_composition` still grades every SPM unit - with no universe predicate, and `reform_validation` still computes the 104 - state SPM rates over every person. Making `_person_rate` nullable-safe is a - separate PR, numerically inert on the pinned engine. +- **No consumer change here.** `check_spm_composition` still grades every SPM + unit with no universe predicate. `reform_validation._person_rate` already + reads the poverty indicator as nullable (#976, merged 2026-09-22), so a + person with a missing indicator leaves both numerator and denominator; on + the pinned engine, where `in_poverty` is Boolean, nothing is missing. - **No producer source pin.** The module is deliberately *not* added to `source_enrichment.PRODUCER_SOURCE_FILES`: that tuple is the SPM-role - enrichment producer's own reviewed source closure, and this module is not in - it until the enrichment lane carries the column. See "Open questions" below. + enrichment producer's own reviewed source closure, and + `validate_source_enrichment_candidate` refuses a recorded inventory whose key + set differs from it. This module is not in it until the enrichment lane + carries the column (ruled 2026-09-22; see "Rulings" below). ## Open questions @@ -143,7 +156,19 @@ repository. The note states each in full. - **Q-D — the published-rate discontinuity.** The 104 state SPM rates move when outside persons leave the denominator, correctly but visibly. - **Q-E — the scope of the ASEC ruling.** Encoded here as a named, cited - constant with its own refusal, per the note's recommendation. + constant with its own refusal, per the note's recommendation; signed off + 2026-09-22 (below). + +## Rulings (Max Ghenis, 2026-09-22) + +- `spm_universe_source.py` stays out of + `source_enrichment.PRODUCER_SOURCE_FILES`. +- The missing-`SPM_ID` refusal on the ASEC arm stays strict: an absent or + partly missing `SPM_ID` refuses rather than falling back. +- `puf_tax_detail` stays unruled: a household whose only provenance tag is + that clone-operator channel refuses with `SPM_UNIVERSE_UNDECLARED_SPINE`. +- The ASEC spine judgement is signed off as the named, cited constant + `ASEC_SPM_INCLUDED_AUTHORITY` (Census SEHSD-WP2020-09, pages 6–7). ## Upstream diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py index 90930fc617..521ba57624 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py @@ -69,9 +69,11 @@ ------------- This producer is a leaf: a preflight or a release tool must be able to call it -without dragging the ACS loader into the process. Measured at this head on -this machine, ``import microcosm.build.us_runtime.acs_pums`` costs ~38 s and -``...stacked_spine`` ~90 s (both pull the country engine), so the four channel +without dragging the ACS loader into the process. Both +``microcosm.build.us_runtime.acs_pums`` and ``...stacked_spine`` pull the +country engine: re-measured in separate processes on 2026-09-22 after merging +main, each import took ~36-38 s on a warm cache (the authoring lane measured +~90 s for ``stacked_spine`` cold). So the four channel tag values below are named here rather than imported from those modules. They are not a second source of truth: :mod:`test_us_spm_universe_source` parses each canonical definition out of its defining module with :mod:`ast` — no @@ -124,7 +126,8 @@ #: The engine input this producer supplies (policyengine-us#9462). The name #: and the three-state vocabulary below are deliberately identical to the -#: graph-native line's ``spm_input_contract``; there is no reason for two +#: graph-native line's ``spm_input_contract`` (unmerged native-SPM branches, +#: first at commit 774a90091; not on main); there is no reason for two #: spellings of the same declaration. UNIVERSE_INPUT = "spm_unit_spm_universe_status" @@ -163,12 +166,13 @@ "frame already equals the CPS SPM universe." ) #: Named authority for the ACS exclusion; identical to the graph-native line's -#: ``acs_spm_scope.ACS_SPM_OUTSIDE_AUTHORITY``. +#: ``acs_spm_scope.ACS_SPM_OUTSIDE_AUTHORITY`` (same unmerged branches). ACS_SPM_OUTSIDE_AUTHORITY = "outside_acs_household_universe" #: Named authority for the ASEC ruling. This one has no upstream: it is this #: producer's own source-universe judgement, made on the Census quote above #: plus the zero-adult reconciliation ``spm_role_source`` already enforces -#: against the pinned complete Census ASEC files. +#: against the pinned complete Census ASEC files. Signed off as this named, +#: cited constant by Max Ghenis on 2026-09-22 (microcosm#977). ASEC_SPM_INCLUDED_AUTHORITY = "asec_sample_frame_equals_cps_spm_universe" #: The named analysis scope a release declares when it carries this status. SPM_UNIVERSE_PROFILE_NAME = "us_acs_household_and_asec_frame_universe_v1" @@ -392,30 +396,49 @@ def _assert_one_kind_per_household( ) +def _person_clone_copies(person: pd.DataFrame) -> np.ndarray: + """Each person's support-clone copy, with a missing index read as native. + + A support clone deep-copies every column of a table and remaps only the + structural id and membership columns + (``puf_support._clone_entity_table``), so clone copy ``k`` carries the + native row's source fields under new ids. Reading an absent or missing + clone index as the native copy can only make the checks that use it + stricter: it merges rows *into* the native copy, where they can collide + with native rows, and never separates rows the truth would merge. + """ + if _PERSON_CLONE_INDEX_COLUMN not in person.columns: + return np.full(len(person), _NATIVE_CLONE_INDEX, dtype=float) + return ( + pd.to_numeric(person[_PERSON_CLONE_INDEX_COLUMN], errors="coerce") + .fillna(_NATIVE_CLONE_INDEX) + .to_numpy(dtype=float) + ) + + def _assert_group_quarters_are_single_native_records( *, household_ids: pd.Series, arm: pd.Series, kind: pd.Series, person_household_ids: pd.Series, - clone_index: pd.Series | None, + clone_copies: np.ndarray, ) -> int: """Mirror the stacked spine's one-native-person-per-GQ-placeholder rule. ACS group-quarters rows are one-person ``WGTP == 0`` placeholders at the source (``acs_pums._occupied_households``), and the stacked assembly re-asserts it. Clones of such a placeholder are expected and are not - counted here; more than one *native* person in one is not. + counted here; more than one *native* person in one is not. A person + whose clone index is missing counts as native (see + :func:`_person_clone_copies`), so a pool that lost the index on one arm + cannot pass this check vacuously. """ gq = (arm.eq(ACS_ARM) & kind.isin(ACS_GROUP_QUARTERS_KINDS)).to_numpy() gq_ids = pd.Index(household_ids.to_numpy()[gq]).unique() if len(gq_ids) == 0: return 0 - native = ( - clone_index.eq(_NATIVE_CLONE_INDEX).to_numpy() - if clone_index is not None - else np.ones(len(person_household_ids), dtype=bool) - ) + native = clone_copies == _NATIVE_CLONE_INDEX counts = ( pd.Series(1, index=person_household_ids.to_numpy()[native]) .groupby(level=0) @@ -461,6 +484,7 @@ def _assert_native_asec_partition( person: pd.DataFrame, asec_person: np.ndarray, unit_membership_column: str, + clone_copies: np.ndarray, ) -> None: """Refuse a silently degraded SPM partition on the ASEC arm. @@ -472,6 +496,13 @@ def _assert_native_asec_partition( that does not exist, so the ASEC arm must prove its partition is the native one. + A support clone copies ``SPM_ID`` verbatim and gives each copy new unit + ids (see :func:`_person_clone_copies`), so every clone copy is its own + instance of the native partition. The native unit key is therefore + ``(clone copy, SPM_ID)``, and the frame's units must be a one-to-one + densification of that key. Keying on ``SPM_ID`` alone would read every + support-cloned ASEC frame as degraded. + The check is deliberately ASEC-only. ACS PUMS supplies no ``SPM_ID`` at all, so the ACS arm reaches ``assign_us_unit_structure`` without one and takes the household fallback by construction — there, one SPM unit per @@ -495,16 +526,21 @@ def _assert_native_asec_partition( f"{ASEC_NATIVE_UNIT_COLUMN}; any missing value silently converts every " "SPM unit in the frame into a household.", ) - units = person.loc[asec_person, unit_membership_column] - pairs = pd.MultiIndex.from_arrays([native.to_numpy(), units.to_numpy()]) + units = person.loc[asec_person, unit_membership_column].to_numpy() + copies = clone_copies[asec_person] + native_units = len(pd.MultiIndex.from_arrays([copies, native.to_numpy()]).unique()) + frame_units = len(pd.unique(units)) + pairings = len( + pd.MultiIndex.from_arrays([copies, native.to_numpy(), units]).unique() + ) _require( - len(pairs.unique()) == native.nunique() == units.nunique(), + pairings == native_units == frame_units, "SPM_UNIVERSE_DEGRADED_PARTITION", f"the ASEC arm's {unit_membership_column!r} is not a one-to-one " - f"densification of {ASEC_NATIVE_UNIT_COLUMN} " - f"({native.nunique()} native unit(s), {units.nunique()} frame unit(s), " - f"{len(pairs.unique())} distinct pairing(s)), so the native partition " - "did not survive assembly.", + f"densification of (clone copy, {ASEC_NATIVE_UNIT_COLUMN}) " + f"({native_units} native unit(s), {frame_units} frame unit(s), " + f"{pairings} distinct pairing(s)), so the native partition did not " + "survive assembly.", ) @@ -598,17 +634,13 @@ def classify_spm_universe( person_units, unit_ids, column=unit_membership_column, owner="spm_unit" ) - clone_index = ( - pd.to_numeric(person[_PERSON_CLONE_INDEX_COLUMN], errors="coerce") - if _PERSON_CLONE_INDEX_COLUMN in person.columns - else None - ) + clone_copies = _person_clone_copies(person) group_quarters_households = _assert_group_quarters_are_single_native_records( household_ids=household_ids, arm=arm, kind=kind, person_household_ids=person_household_ids, - clone_index=clone_index, + clone_copies=clone_copies, ) kind_by_household = pd.Series( @@ -628,6 +660,7 @@ def classify_spm_universe( person=person, asec_person=asec_person, unit_membership_column=unit_membership_column, + clone_copies=clone_copies, ) by_unit = pd.DataFrame( @@ -743,18 +776,25 @@ def attach_spm_universe_status( Raises: ValueError: Any of :data:`SPM_UNIVERSE_REFUSALS`, including - ``SPM_UNIVERSE_COLUMN_EXISTS`` when the unit table already carries - the declaration — a producer overwriting an existing declaration - would silently replace someone else's source ruling. + ``SPM_UNIVERSE_COLUMN_EXISTS`` when any entity table already + carries the declaration — a producer overwriting an existing + declaration would silently replace someone else's source ruling, + and a copy on another entity would collide with it under the + frame's global column-name rule. """ from microcosm.frame import Frame schema = frame.schema unit_table = frame.table(unit_entity) + carriers = [ + entity + for entity in frame.entities + if UNIVERSE_INPUT in frame.table(entity).columns + ] _require( - UNIVERSE_INPUT not in unit_table.columns, + not carriers, "SPM_UNIVERSE_COLUMN_EXISTS", - f"the {unit_entity} table already carries {UNIVERSE_INPUT!r}; this " + f"the {carriers} table(s) already carry {UNIVERSE_INPUT!r}; this " "producer never overwrites an existing universe declaration.", ) classification = classify_spm_universe( diff --git a/packages/microcosm-build/tests/test_us_spm_universe_source.py b/packages/microcosm-build/tests/test_us_spm_universe_source.py index c2a1f420d1..28a05812c2 100644 --- a/packages/microcosm-build/tests/test_us_spm_universe_source.py +++ b/packages/microcosm-build/tests/test_us_spm_universe_source.py @@ -20,6 +20,7 @@ ACS_HOUSING_UNIT_KIND, ASEC_ARM, ASEC_NATIVE_UNIT_COLUMN, + ASEC_RECORD_TYPE_FIELDS, HOUSEHOLD_SPINE_COLUMN, HOUSEHOLD_SUPPORT_CHANNEL_COLUMN, INCLUDED, @@ -48,6 +49,7 @@ def _tables( person_extra_columns: dict[str, list[object]] | None = None, extra_household_rows: list[dict[str, object]] | None = None, extra_unit_ids: list[int] | None = None, + without_support_metadata: bool = False, ) -> tuple[pd.DataFrame, pd.DataFrame, np.ndarray]: """Household, person and SPM unit ids from a list of invented persons. @@ -66,7 +68,10 @@ def _tables( "person_tax_unit_id": hid, "person_family_id": hid, "person_marital_unit_id": index, - "person_support_clone_index": int(spec.get("clone", 0)), + # ``"clone": None`` models a pool that lost the index on one arm. + "person_support_clone_index": ( + np.nan if spec.get("clone", 0) is None else int(spec.get("clone", 0)) + ), } if "spm_id" in spec: row[ASEC_NATIVE_UNIT_COLUMN] = spec["spm_id"] @@ -76,6 +81,8 @@ def _tables( household[ACS_HOUSEHOLD_KIND_COLUMN] = spec["kind"] households.setdefault(hid, household) person = pd.DataFrame(prows) + if without_support_metadata: + person = person.drop(columns=["person_support_clone_index"]) if person_extra_columns: for name, values in person_extra_columns.items(): person[name] = values @@ -261,6 +268,57 @@ def test_the_support_channel_column_is_read_when_no_spine_tag_is_present(): assert statuses == {1: INCLUDED, 2: INCLUDED} +def test_the_spine_tag_takes_precedence_over_the_support_channel(): + """A PUF clone's support channel names its operator role, not its spine.""" + statuses = _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}], + household_extra={1: {HOUSEHOLD_SUPPORT_CHANNEL_COLUMN: "puf_tax_detail"}}, + ) + assert statuses == {1: INCLUDED} + + +def test_an_explicit_channel_column_overrides_the_default_order(): + household, person, unit_ids = _tables( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2}], + household_extra={1: {"operator_channel": _ACS_CHANNEL}}, + ) + result = classify_spm_universe( + household=household.drop(columns=[HOUSEHOLD_SPINE_COLUMN]).assign( + **{HOUSEHOLD_SUPPORT_CHANNEL_COLUMN: "puf_tax_detail"} + ), + person=person, + unit_ids=unit_ids, + channel_column="operator_channel", + ) + assert result.status.tolist() == [OUTSIDE] + assert result.provenance["channel_source"] == "operator_channel" + + +@pytest.mark.parametrize( + ("channel_column", "declared_channel", "reason"), + [ + ("operator_channel", None, "no requested channel column"), + (HOUSEHOLD_SPINE_COLUMN, _ACS_CHANNEL, "mutually exclusive"), + ], +) +def test_refuses_a_bad_explicit_channel_column( + channel_column, declared_channel, reason +): + household, person, unit_ids = _tables( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}] + ) + with pytest.raises( + ValueError, match=rf"^SPM_UNIVERSE_UNDECLARED_SPINE: .*{reason}" + ): + classify_spm_universe( + household=household, + person=person, + unit_ids=unit_ids, + channel_column=channel_column, + declared_channel=declared_channel, + ) + + def test_an_untagged_table_accepts_an_explicit_caller_declaration(): household = pd.DataFrame( {"household_id": [1], ACS_HOUSEHOLD_KIND_COLUMN: [2]}, @@ -319,6 +377,15 @@ def test_refuses_a_preexisting_declaration(): attach_spm_universe_status(declared) +def test_refuses_a_declaration_already_on_another_entity_table(): + frame = _frame( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + person_extra_columns={UNIVERSE_INPUT: [INCLUDED]}, + ) + with _refuses("SPM_UNIVERSE_COLUMN_EXISTS"): + attach_spm_universe_status(frame) + + def test_refuses_a_channel_with_no_universe_ruling(): with _refuses("SPM_UNIVERSE_UNDECLARED_SPINE"): _classify([{"hid": 1, "unit": 1, "chan": "puf_tax_detail", "kind": 1}]) @@ -470,6 +537,119 @@ def test_refuses_an_asec_partition_that_is_not_the_native_one(): ) +def test_a_support_cloned_asec_frame_keeps_its_native_partition(): + """Each support-clone copy is its own instance of the native partition. + + ``puf_support._clone_entity_table`` deep-copies the person table and remaps + only the id and membership columns, so clone copy 1 carries the native + ``SPM_ID`` under new unit ids. Keying the check on ``SPM_ID`` alone read + every support-cloned ASEC frame as degraded. + """ + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12, "clone": 0}, + {"hid": 2, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 21, "clone": 0}, + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 1}, + {"hid": 11, "unit": 12, "chan": _ASEC_CHANNEL, "spm_id": 12, "clone": 1}, + {"hid": 12, "unit": 13, "chan": _ASEC_CHANNEL, "spm_id": 21, "clone": 1}, + ] + ) + assert statuses == dict.fromkeys((1, 2, 3, 11, 12, 13), INCLUDED) + + +@pytest.mark.parametrize( + "clone_rows", + [ + pytest.param( + [ + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 12}, + ], + id="two-native-units-collapsed-inside-the-clone-copy", + ), + pytest.param( + [ + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 11, "unit": 12, "chan": _ASEC_CHANNEL, "spm_id": 11}, + ], + id="one-native-unit-split-inside-the-clone-copy", + ), + ], +) +def test_the_clone_copy_key_still_refuses_a_degraded_clone_copy(clone_rows): + """Negative control: keying on the copy does not relax the check within it.""" + native = [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12, "clone": 0}, + ] + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify(native + [dict(row, clone=1) for row in clone_rows]) + + +def test_a_missing_clone_index_is_read_as_native_on_the_asec_arm(): + """A clone row that lost its index collides with its native row: refused.""" + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + { + "hid": 11, + "unit": 11, + "chan": _ASEC_CHANNEL, + "spm_id": 11, + "clone": None, + }, + ] + ) + + +def test_a_missing_clone_index_counts_as_native_in_group_quarters(): + """A lost clone index cannot make the one-person GQ check pass vacuously.""" + with _refuses("SPM_UNIVERSE_GQ_MULTI_PERSON"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": None}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": None}, + ] + ) + + +@pytest.mark.requires_us +def test_the_real_support_clone_operator_output_is_accepted(): + """The copy-key rule holds on ``clone_us_frame_for_puf_support`` itself. + + Invented rows. Imported lazily: the operator's module pulls the country + engine, which the rest of this file never needs. + """ + from microcosm.build.us_runtime.puf_support import ( + clone_us_frame_for_puf_support, + ) + + frame = _frame( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12}, + {"hid": 2, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 21}, + ], + without_support_metadata=True, + ) + cloned = clone_us_frame_for_puf_support(frame) + person = cloned.table("person") + # The operator's shape this rule depends on: SPM_ID copied verbatim onto + # a second copy whose SPM units are new ids. + assert sorted(person["person_support_clone_index"].unique().tolist()) == [0, 1] + by_copy = person.groupby("person_support_clone_index") + assert by_copy[ASEC_NATIVE_UNIT_COLUMN].apply(sorted).tolist() == [ + [11, 12, 21], + [11, 12, 21], + ] + assert person["person_spm_unit_id"].nunique() == 6 + + result = attach_spm_universe_status(cloned) + assert result.status.tolist() == [INCLUDED] * 6 + + def test_the_degraded_partition_check_does_not_fire_on_the_acs_arm(): """ACS PUMS supplies no ``SPM_ID``; one unit per household is native there.""" assert _classify( @@ -480,7 +660,7 @@ def test_the_degraded_partition_check_does_not_fire_on_the_acs_arm(): ) == {1: INCLUDED, 2: INCLUDED} -@pytest.mark.parametrize("field", ["H_TYPE", "HRHTYPE", "H_HHTYPE", "HUNITS"]) +@pytest.mark.parametrize("field", ASEC_RECORD_TYPE_FIELDS) def test_refuses_an_asec_vintage_carrying_a_record_type_field(field): with _refuses("SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED"): _classify( @@ -489,6 +669,14 @@ def test_refuses_an_asec_vintage_carrying_a_record_type_field(field): ) +def test_refuses_an_asec_record_type_field_on_the_person_table(): + with _refuses("SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}], + person_extra_columns={"H_TYPE": [1]}, + ) + + def test_an_acs_only_frame_is_unaffected_by_the_asec_record_type_guard(): """Negative control: the guard is scoped to the arm whose ruling it guards.""" assert _classify( From f07d0f0e94d5ef830475d6fb1e43557b03268595 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 22 Sep 2026 21:18:21 -0400 Subject: [PATCH 3/4] Refuse an undecidable support-clone index instead of reading it as native When the person table carries person_support_clone_index, a missing, non-finite, negative or non-integral value now raises SPM_UNIVERSE_INVALID_CLONE_INDEX before the group-quarters and degraded-partition checks run. Reading a missing index as the native copy could merge two clone copies that share one frame unit into a single key and let a genuine cross-copy collision pass the partition check (gate round 1 counterexample). The maintained readers of this column already refuse such values. A frame with no clone-index column stays all-native. Co-Authored-By: Claude Opus 5.5 --- docs/us-spm-measurement-universe.md | 11 ++-- .../build/us_runtime/spm_universe_source.py | 38 ++++++++----- .../tests/test_us_spm_universe_source.py | 53 ++++++++++++++++--- 3 files changed, 80 insertions(+), 22 deletions(-) diff --git a/docs/us-spm-measurement-universe.md b/docs/us-spm-measurement-universe.md index 61be3efba5..b1c4f9beee 100644 --- a/docs/us-spm-measurement-universe.md +++ b/docs/us-spm-measurement-universe.md @@ -80,6 +80,7 @@ Every refusal is a named, fail-closed `ValueError`: | `SPM_UNIVERSE_UNIT_SPANS_KINDS` | one SPM unit whose members sit in households of different kinds | | `SPM_UNIVERSE_DEGRADED_PARTITION` | the ASEC arm's SPM partition is not the native one: `SPM_ID` absent, missing, or not one-to-one with the frame's units per support-clone copy | | `SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED` | the frame carries an ASEC record-type field the spine-level ruling was not derived against | +| `SPM_UNIVERSE_INVALID_CLONE_INDEX` | the person table carries a support-clone index that is missing, non-finite, negative or non-integral, so a row's clone copy is undecidable | `SPM_UNIVERSE_UNIT_SPANS_KINDS` is natively unreachable — ACS group-quarters households are one-person placeholders — and earns its keep the moment a @@ -104,9 +105,13 @@ person table and remaps only the id and membership columns, so clone copy 1 carries the native `SPM_ID` under new SPM unit ids; keyed on `SPM_ID` alone, every support-cloned ASEC frame would read as degraded. `test_the_real_support_clone_operator_output_is_accepted` runs the real clone -operator on invented rows to pin that. A missing clone index is read as the -native copy, which can only add collisions, so it never relaxes this check or -the one-native-person group-quarters check. +operator on invented rows to pin that. When the frame carries a clone-index +column, a missing, non-finite, negative or non-integral value is refused +(`SPM_UNIVERSE_INVALID_CLONE_INDEX`) before either this check or the +one-native-person group-quarters check runs: the row's copy is undecidable, +and reading it as native could merge a genuine cross-copy collision into an +apparent bijection. A frame with no clone-index column was never +support-cloned, and every person in it is native. ## What is wired, and what is not diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py index 521ba57624..7fd990025f 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py @@ -217,6 +217,7 @@ "SPM_UNIVERSE_UNIT_SPANS_KINDS", "SPM_UNIVERSE_DEGRADED_PARTITION", "SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED", + "SPM_UNIVERSE_INVALID_CLONE_INDEX", ) _MAX_REPORTED_EXAMPLES = 5 @@ -397,23 +398,36 @@ def _assert_one_kind_per_household( def _person_clone_copies(person: pd.DataFrame) -> np.ndarray: - """Each person's support-clone copy, with a missing index read as native. + """Each person's support-clone copy; refuses an undecidable clone index. A support clone deep-copies every column of a table and remaps only the structural id and membership columns (``puf_support._clone_entity_table``), so clone copy ``k`` carries the - native row's source fields under new ids. Reading an absent or missing - clone index as the native copy can only make the checks that use it - stricter: it merges rows *into* the native copy, where they can collide - with native rows, and never separates rows the truth would merge. + native row's source fields under new ids. With no clone-index column the + frame was never support-cloned and every person is native. When the column + is present, a missing, non-finite, negative or non-integral value is + undecidable provenance, and the module refuses it rather than guess a copy: + reading it as native could merge a genuine cross-copy collision into an + apparent bijection in the degraded-partition check. The maintained readers + of this column refuse the same values (``acs_transfer`` raises on an + invalid clone index; ``stacked_spine`` and ``puf_support`` parse it with + ``errors="raise"`` or refuse a missing one). """ if _PERSON_CLONE_INDEX_COLUMN not in person.columns: return np.full(len(person), _NATIVE_CLONE_INDEX, dtype=float) - return ( - pd.to_numeric(person[_PERSON_CLONE_INDEX_COLUMN], errors="coerce") - .fillna(_NATIVE_CLONE_INDEX) - .to_numpy(dtype=float) + values = pd.to_numeric( + person[_PERSON_CLONE_INDEX_COLUMN], errors="coerce" + ).to_numpy(dtype=float) + invalid = ~np.isfinite(values) | (values < 0) | (values != np.floor(values)) + _require( + not invalid.any(), + "SPM_UNIVERSE_INVALID_CLONE_INDEX", + f"{int(invalid.sum())} person row(s) carry a missing, non-finite, " + f"negative or non-integral {_PERSON_CLONE_INDEX_COLUMN}; the row's " + "support-clone copy is undecidable, so neither the group-quarters nor " + "the degraded-partition check can be judged.", ) + return values def _assert_group_quarters_are_single_native_records( @@ -430,9 +444,9 @@ def _assert_group_quarters_are_single_native_records( source (``acs_pums._occupied_households``), and the stacked assembly re-asserts it. Clones of such a placeholder are expected and are not counted here; more than one *native* person in one is not. A person - whose clone index is missing counts as native (see - :func:`_person_clone_copies`), so a pool that lost the index on one arm - cannot pass this check vacuously. + whose clone index is missing or invalid is refused before this check runs + (see :func:`_person_clone_copies`), so a pool that lost the index on one + arm cannot pass it vacuously. """ gq = (arm.eq(ACS_ARM) & kind.isin(ACS_GROUP_QUARTERS_KINDS)).to_numpy() gq_ids = pd.Index(household_ids.to_numpy()[gq]).unique() diff --git a/packages/microcosm-build/tests/test_us_spm_universe_source.py b/packages/microcosm-build/tests/test_us_spm_universe_source.py index 28a05812c2..e4718596eb 100644 --- a/packages/microcosm-build/tests/test_us_spm_universe_source.py +++ b/packages/microcosm-build/tests/test_us_spm_universe_source.py @@ -68,9 +68,14 @@ def _tables( "person_tax_unit_id": hid, "person_family_id": hid, "person_marital_unit_id": index, - # ``"clone": None`` models a pool that lost the index on one arm. + # ``"clone": None`` models a pool that lost the index on one arm; + # a float passes through unchanged so invalid indices can be tested. "person_support_clone_index": ( - np.nan if spec.get("clone", 0) is None else int(spec.get("clone", 0)) + np.nan + if spec.get("clone", 0) is None + else spec.get("clone", 0) + if isinstance(spec.get("clone", 0), float) + else int(spec.get("clone", 0)) ), } if "spm_id" in spec: @@ -587,9 +592,9 @@ def test_the_clone_copy_key_still_refuses_a_degraded_clone_copy(clone_rows): _classify(native + [dict(row, clone=1) for row in clone_rows]) -def test_a_missing_clone_index_is_read_as_native_on_the_asec_arm(): - """A clone row that lost its index collides with its native row: refused.""" - with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): +def test_a_missing_clone_index_is_refused_on_the_asec_arm(): + """A clone row that lost its index has an undecidable copy: refused.""" + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): _classify( [ {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, @@ -604,9 +609,9 @@ def test_a_missing_clone_index_is_read_as_native_on_the_asec_arm(): ) -def test_a_missing_clone_index_counts_as_native_in_group_quarters(): +def test_a_missing_clone_index_is_refused_in_group_quarters(): """A lost clone index cannot make the one-person GQ check pass vacuously.""" - with _refuses("SPM_UNIVERSE_GQ_MULTI_PERSON"): + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): _classify( [ {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": None}, @@ -615,6 +620,40 @@ def test_a_missing_clone_index_counts_as_native_in_group_quarters(): ) +def test_a_missing_clone_index_cannot_collapse_a_cross_copy_collision(): + """The gate's counterexample: two copies sharing one frame unit. + + With both indices present the rows are two native keys in one frame unit + and the partition check refuses. Reading the missing indices as native + would merge them into one key and accept; the module refuses first. + """ + rows = [ + {"hid": 1, "unit": 5, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 5, "chan": _ASEC_CHANNEL, "spm_id": 11}, + ] + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify([dict(rows[0], clone=1), dict(rows[1], clone=2)]) + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): + _classify([dict(row, clone=None) for row in rows]) + + +@pytest.mark.parametrize("bad", [-1.0, 0.5, float("inf")]) +def test_an_invalid_clone_index_is_refused(bad): + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + { + "hid": 2, + "unit": 2, + "chan": _ASEC_CHANNEL, + "spm_id": 11, + "clone": bad, + }, + ] + ) + + @pytest.mark.requires_us def test_the_real_support_clone_operator_output_is_accepted(): """The copy-key rule holds on ``clone_us_frame_for_puf_support`` itself. From c9ed12706a0d588ff3ca1017f731ee89dcc9086c Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 22 Sep 2026 22:09:25 -0400 Subject: [PATCH 4/4] Count the twelfth refusal in the changelog fragment Co-Authored-By: Claude Opus 5.5 --- changelog.d/us-spm-measurement-universe-source.added.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/us-spm-measurement-universe-source.added.md b/changelog.d/us-spm-measurement-universe-source.added.md index fa8fffd676..bb9db4f5ba 100644 --- a/changelog.d/us-spm-measurement-universe-source.added.md +++ b/changelog.d/us-spm-measurement-universe-source.added.md @@ -1 +1 @@ -Add `us_runtime.spm_universe_source`, the source-owned producer for policyengine-us#9462's `spm_unit_spm_universe_status`: ACS housing units and every ASEC record are `INCLUDED`, ACS group quarters are `OUTSIDE`, and eleven named refusals fail closed rather than guess. Declared from the source's own record type, never from a unit's composition. Not wired into any build path and not exported. +Add `us_runtime.spm_universe_source`, the source-owned producer for policyengine-us#9462's `spm_unit_spm_universe_status`: ACS housing units and every ASEC record are `INCLUDED`, ACS group quarters are `OUTSIDE`, and twelve named refusals fail closed rather than guess. Declared from the source's own record type, never from a unit's composition. Not wired into any build path and not exported.