diff --git a/docs/evidence/spec-engine/us-f0-coverage.json b/docs/evidence/spec-engine/us-f0-coverage.json index 9e3634d5ce..e9214c8ecd 100644 --- a/docs/evidence/spec-engine/us-f0-coverage.json +++ b/docs/evidence/spec-engine/us-f0-coverage.json @@ -1656,13 +1656,13 @@ "compiler_ir.node_slices" ], "expected": { - "map_sha256": "a6feafaf62ed99b4f3fa9fee32e986f83919aed7335848df6470db6ecd17987d", - "protocol_sha256": "45f82554869c7a966b3c12b71edab43a6c887806b0fc6e70cd7598e62f375922" + "map_sha256": "18594e59d7b060265ac55afa41314adab640579e6ccef929123f39bd0bad5ee0", + "protocol_sha256": "56f832aae97f7b8c938f332a2ea52f1a3b067e31dc504d4d60717c970ef615c8" }, "failures": [], "observed": { - "map_sha256": "a6feafaf62ed99b4f3fa9fee32e986f83919aed7335848df6470db6ecd17987d", - "protocol_sha256": "45f82554869c7a966b3c12b71edab43a6c887806b0fc6e70cd7598e62f375922" + "map_sha256": "18594e59d7b060265ac55afa41314adab640579e6ccef929123f39bd0bad5ee0", + "protocol_sha256": "56f832aae97f7b8c938f332a2ea52f1a3b067e31dc504d4d60717c970ef615c8" }, "status": "covered" }, @@ -1677,7 +1677,7 @@ "compiler_ir.seed_stream_map" ], "expected": { - "implementation_sha256": "45f82554869c7a966b3c12b71edab43a6c887806b0fc6e70cd7598e62f375922", + "implementation_sha256": "56f832aae97f7b8c938f332a2ea52f1a3b067e31dc504d4d60717c970ef615c8", "protocol": "legacy-v1", "streams": [ "build_model", @@ -1698,7 +1698,7 @@ }, "failures": [], "observed": { - "implementation_sha256": "45f82554869c7a966b3c12b71edab43a6c887806b0fc6e70cd7598e62f375922", + "implementation_sha256": "56f832aae97f7b8c938f332a2ea52f1a3b067e31dc504d4d60717c970ef615c8", "protocol": "legacy-v1", "streams": [ "build_model", @@ -2599,7 +2599,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "1c46dda9cff4e53e5a44c6502a46fa695d02d92731f23d618616a892071d4841" + "spec_sha256": "3667e30ebfdac4d893e3a111bfba998dc7acf29b0da05360eee521d3e144d08a" } }, "report_schema_version": 3, @@ -2609,7 +2609,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "1c46dda9cff4e53e5a44c6502a46fa695d02d92731f23d618616a892071d4841" + "spec_sha256": "3667e30ebfdac4d893e3a111bfba998dc7acf29b0da05360eee521d3e144d08a" }, "status": "pass" } diff --git a/docs/us-release-build-rule.md b/docs/us-release-build-rule.md index af39120c49..8ee87102cf 100644 --- a/docs/us-release-build-rule.md +++ b/docs/us-release-build-rule.md @@ -145,6 +145,37 @@ on every arm after loading the base. It runs the other checks unchanged on the whole base. With `--release-manifest`, it also requires that release to record `build.selection_source` as `{"enabled": false}`. +**23 September 2026:** without a selection source the release runs its stages +on every support copy the base carries, including the capital-gains own-tail +copy (clone index 2) of each tail household. That copy keeps its source IDs and +the `puf_tax_detail` channel, so the base from `build_us_puf_support_base.py` +holds two PUF-role rows for each tail source person (42,336 on base +`cb1bd1e6…`). SIPP Head Start and voluntary filing identified a copy by source +ID and role on bases without a raw spine ID, so they refused that base. Release +stages now identify a copy by source ID and clone index +(`support_copy_rank_series`), whether or not the base carries a raw spine ID. +A repeated pair is still refused. Source-level decisions fan out to the tail +copy from the lowest surviving clone index. Row-level imputations treat the +tail copy as a PUF-role row, as the base's own post-transfer stages do. +Only a base without a raw spine ID may lack clone indices. An assembled table +(one with a raw spine ID) must carry both its support channel and its clone +index. `require_assembled_support_provenance` refuses one that lacks either. +The metadata-presence check includes the raw spine ID, so losing both columns +cannot make an assembled table look historical. `support_role_series` calls +the validator before resolving roles; this covers the prior-year and SSI +summaries before occurrence pairing while preserving their attested source +bytes and the bundle digest. Head Start, voluntary filing and copy ranking +also validate explicitly before their fallbacks. Missing provenance refuses +imputation and fails or raises from the signal gates. With clone indices the +gates compare every copy of a source unit, including repeated +`(source ID, clone index)` copies; imputations refuse repeated pairs. + +Temporary source-kernel projections validate assembled provenance before +removing the channel, clone index and raw spine ID together. They retain the +stable source IDs, and their outputs are compared with or merged into the +original assembled table. This lets historical source kernels consume the +projection without treating intentionally removed provenance as corruption. + `--dense-default-dataset` is diagnostic only. A release build leaves it unset, so the default is the sparse dataset that runs on standard machines. diff --git a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py index 6228dd7295..61854eb5b9 100644 --- a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py +++ b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py @@ -359,8 +359,8 @@ "late_schedule": "e59c019d3d454eac99ac0ac209b6c5b6faaf9bdfcaeee18c36a25be19bf7da2f", "ownership": "5f64f0aac49e2313177564f71876bffc8c81b3ded4df701e70930e60e9c98356", "primary_tuples": "987b501c695e31f45521c4a178528f75ab3df22c09bc407b182213b2de99ee57", - "seed_map": "a6feafaf62ed99b4f3fa9fee32e986f83919aed7335848df6470db6ecd17987d", - "seed_protocol": "45f82554869c7a966b3c12b71edab43a6c887806b0fc6e70cd7598e62f375922", + "seed_map": "18594e59d7b060265ac55afa41314adab640579e6ccef929123f39bd0bad5ee0", + "seed_protocol": "56f832aae97f7b8c938f332a2ea52f1a3b067e31dc504d4d60717c970ef615c8", "source_manifest": "a87278445d928ea20e15935a8cc07312d6e819e34409b04e8c814f17910d828f", "take_up": "9522ce40f7dea569312dd7a7beb474e5a5afadd2f1e10cb5876534c3ef623d35", "tail": "ac92829c88a1a4fb6460d61190918d5d99c6c377fc8dd8f62f02b332d09bf59c", diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/prior_year_income.py b/packages/microcosm-build/src/microcosm/build/us_runtime/prior_year_income.py index 899b01ea0f..73594082cf 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/prior_year_income.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/prior_year_income.py @@ -40,8 +40,8 @@ run_source_stage, ) from microcosm.build.us_runtime.support_provenance import ( - has_assembled_support_metadata, has_support_role_metadata, + support_clone_index_column, support_role_series, without_support_role_metadata, ) @@ -793,9 +793,12 @@ def _share(mask: np.ndarray) -> float: } ) group_columns = ["_source_id"] - if has_support_role_metadata( - person, entity="person" - ) and not has_assembled_support_metadata(person, entity="person"): + # With clone provenance every copy of a source person, the + # capital-gains own-tail copy included, must agree. Only channel-only + # fixtures, which predate clone indices, pair copies by occurrence. + if has_support_role_metadata(person, entity="person") and ( + support_clone_index_column("person") not in person + ): clone_work["_role"] = support_role_series( person, entity="person" ).to_numpy() diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/sipp_head_start.py b/packages/microcosm-build/src/microcosm/build/us_runtime/sipp_head_start.py index 5d45b4764f..0708b27ae2 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/sipp_head_start.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/sipp_head_start.py @@ -35,9 +35,9 @@ from microcosm.build.gates import GateResult from microcosm.build.source_manifest import SourceStageSpec, load_source_manifest from microcosm.build.us_runtime.support_provenance import ( - has_assembled_support_metadata, has_support_role_metadata, - support_clone_index_column, + require_assembled_support_provenance, + support_copy_rank_series, support_role_series, ) from microcosm.build.us_runtime.voluntary_filing import ( @@ -103,8 +103,6 @@ _OUTPUT = US_SIPP_HEAD_START_OUTPUT_COLUMNS[0] _DONOR_WEIGHT_COLUMN = "sipp_weight" _PERSON_SOURCE_ID_COLUMN = "person_source_id" -_ASEC_CHANNEL = "asec" -_PUF_CHANNEL = "puf_tax_detail" _DEFAULT_N_ESTIMATORS = 100 _ELIGIBLE_MIN_AGE = 3 _ELIGIBLE_MAX_AGE = 5 @@ -483,25 +481,37 @@ def _support_group_keys( person: pd.DataFrame, source_id: pd.Series, ) -> tuple[pd.Series, pd.Series]: - """Return source-person keys and validated legacy-compatible roles.""" + """Return source-person keys and each row's validated support-copy rank. + + Every support copy of a source person shares its key. The rank orders the + copies (native first) and, with the source ID, identifies exactly one row: + the capital-gains own-tail copy is a second PUF-role row of its source + person, distinguished from the primary PUF-detail copy by clone index. + + Only a historical table without support metadata ranks every row 0. An + assembled table that lost its provenance columns is refused before that + early return, so a stripped assembled copy cannot become canonical. + """ - if not has_support_role_metadata(person, entity="person"): - return ( - source_id.astype(object), - pd.Series(_ASEC_CHANNEL, index=person.index), - ) try: - roles = support_role_series(person, entity="person") + require_assembled_support_provenance(person, entity="person") + if not has_support_role_metadata(person, entity="person"): + return ( + source_id.astype(object), + pd.Series(0, index=person.index, dtype="int64"), + ) + copy_rank = support_copy_rank_series(person, entity="person") except ValueError as exc: raise ValueError( "US SIPP Head Start found unsupported support channel or " f"clone-role metadata: {exc}" ) from exc - return source_id.astype(object), roles + return source_id.astype(object), copy_rank def _recipient_predictors(frame: Frame) -> tuple[pd.DataFrame, pd.Series, np.ndarray]: person = frame.table("person") + require_assembled_support_provenance(person, entity="person") missing = [ column for column in US_SIPP_HEAD_START_REQUIRED_SOURCE_COLUMNS @@ -563,7 +573,7 @@ def _recipient_predictors(frame: Frame) -> tuple[pd.DataFrame, pd.Series, np.nda ) source_id = _decoded_strings(person[_PERSON_SOURCE_ID_COLUMN]) - source_key, roles = _support_group_keys(person, source_id) + source_key, copy_rank = _support_group_keys(person, source_id) age_unique = ( pd.Series(age, index=person.index).groupby(source_key, sort=False).nunique() ) @@ -576,55 +586,22 @@ def _recipient_predictors(frame: Frame) -> tuple[pd.DataFrame, pd.Series, np.nda order = pd.DataFrame(index=person.index) order["source_id"] = source_id order["source_key"] = source_key - if has_assembled_support_metadata(person, entity="person"): - clone_column = support_clone_index_column("person") - clone_indices = pd.to_numeric(person[clone_column], errors="raise").astype( - "int64" + order["copy_rank"] = copy_rank + duplicate_copies = order.duplicated(["source_id", "copy_rank"], keep=False) + if duplicate_copies.any(): + bad = ( + order.loc[duplicate_copies, ["source_id", "copy_rank"]] + .drop_duplicates() + .itertuples(index=False, name=None) ) - clone_rows = pd.DataFrame( - {"source_id": source_id, "clone_index": clone_indices} - ) - duplicate_clones = clone_rows.duplicated( - ["source_id", "clone_index"], - keep=False, - ) - if duplicate_clones.any(): - bad = ( - clone_rows.loc[ - duplicate_clones, - ["source_id", "clone_index"], - ] - .drop_duplicates() - .itertuples(index=False, name=None) - ) - raise ValueError( - "US SIPP Head Start assembled source units carry duplicated " - f"clone-index rows; invalid source clone(s): {list(bad)[:5]}." - ) - order["clone_priority"] = clone_indices - priority_column = "clone_priority" - else: - role_rows = pd.DataFrame({"source_id": source_id, "role": roles}) - duplicate_roles = role_rows.duplicated( - ["source_id", "role"], - keep=False, + raise ValueError( + "US SIPP Head Start source units carry duplicated support copies; " + f"invalid (source, copy rank) pair(s): {list(bad)[:5]}." ) - if duplicate_roles.any(): - bad = ( - role_rows.loc[duplicate_roles, ["source_id", "role"]] - .drop_duplicates() - .itertuples(index=False, name=None) - ) - raise ValueError( - "US SIPP Head Start source units carry duplicated same-role " - f"rows; invalid source role(s): {list(bad)[:5]}." - ) - order["role_priority"] = roles.map({_ASEC_CHANNEL: 0, _PUF_CHANNEL: 1}) - priority_column = "role_priority" order["person_key"] = person["person_id"].astype(str) canonical_index = ( order.sort_values( - ["source_id", priority_column, "person_key"], + ["source_id", "copy_rank", "person_key"], kind="mergesort", ) .drop_duplicates("source_key", keep="first") @@ -670,6 +647,7 @@ def impute_us_sipp_head_start( if frame.schema != US_SCHEMA: raise ValueError("US SIPP Head Start imputation requires the US schema.") + require_assembled_support_provenance(frame.table("person"), entity="person") if n_estimators < 1: raise ValueError("n_estimators must be positive") required = [ @@ -805,11 +783,15 @@ def us_sipp_head_start_summary(frame: Frame) -> dict[str, object]: ) role_invalid = False roles: pd.Series | None = None - if has_support_role_metadata(person, entity="person"): - try: + try: + # Flag an assembled table missing a provenance column before the + # metadata check could read it as a historical table without support + # roles. Its copies are still grouped by source ID below. + require_assembled_support_provenance(person, entity="person") + if has_support_role_metadata(person, entity="person"): roles = support_role_series(person, entity="person") - except ValueError: - role_invalid = True + except ValueError: + role_invalid = True clone_groups = clone_mismatches = 0 if not provenance_missing: source_ids = _decoded_strings(person[_PERSON_SOURCE_ID_COLUMN]) @@ -861,12 +843,22 @@ def us_sipp_head_start_signal_gate(frame: Frame) -> GateResult: """Require a nondefault, in-domain, source-clone-consistent take-up flag.""" person = frame.table("person") + provenance_failures: tuple[str, ...] = () + try: + require_assembled_support_provenance(person, entity="person") + except ValueError as exc: + # Retain the summary's clone diagnostics when output is available, + # and report invalid provenance even on the missing-output path. + provenance_failures = (str(exc),) if _OUTPUT not in person: return GateResult( name="sipp_head_start_signal", passed=False, - failures=(f"person.{_OUTPUT}: missing",), - details={"missing": [_OUTPUT]}, + failures=(f"person.{_OUTPUT}: missing", *provenance_failures), + details={ + "missing": [_OUTPUT], + "support_channel_invalid": bool(provenance_failures), + }, ) try: summary = us_sipp_head_start_summary(frame) 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 7fd990025f..7edb9673ad 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 @@ -93,6 +93,8 @@ from .support_provenance import ( BASE_ASEC_SUPPORT_CHANNEL, + _validated_clone_indices, + require_assembled_support_provenance, support_channel_column, support_clone_index_column, ) @@ -400,34 +402,23 @@ def _assert_one_kind_per_household( def _person_clone_copies(person: pd.DataFrame) -> np.ndarray: """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. 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). + Historical input without a clone-index column retains the native-copy + fallback. Assembled input must carry complete support provenance before + that fallback is considered. Shared int64 validation rejects invalid + values and preserves distinct large integer copy indices, so the native + partition check does not merge copies through a float conversion. """ - if _PERSON_CLONE_INDEX_COLUMN not in person.columns: - return np.full(len(person), _NATIVE_CLONE_INDEX, 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 + try: + require_assembled_support_provenance(person, entity="person") + if _PERSON_CLONE_INDEX_COLUMN not in person.columns: + return np.full(len(person), _NATIVE_CLONE_INDEX, dtype=float) + return _validated_clone_indices( + person, + _PERSON_CLONE_INDEX_COLUMN, + owner="SPM universe support provenance", + ) + except ValueError as exc: + _require(False, "SPM_UNIVERSE_INVALID_CLONE_INDEX", str(exc)) def _assert_group_quarters_are_single_native_records( diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/ssi_disability_criteria.py b/packages/microcosm-build/src/microcosm/build/us_runtime/ssi_disability_criteria.py index fa74927642..aeb3f794f9 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/ssi_disability_criteria.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/ssi_disability_criteria.py @@ -45,8 +45,8 @@ from microcosm.build.gates import GateResult from microcosm.build.source_manifest import SourceStageSpec, load_source_manifest from microcosm.build.us_runtime.support_provenance import ( - has_assembled_support_metadata, has_support_role_metadata, + support_clone_index_column, support_role_series, ) from microcosm.build.us_runtime.voluntary_filing import ( @@ -1188,7 +1188,9 @@ def us_ssi_disability_criteria_summary(frame: Frame) -> dict[str, object]: "value": values, } ) - if has_assembled_support_metadata(person, entity="person"): + # With clone provenance every copy of a source person, the + # capital-gains own-tail copy included, joins one divergence group. + if support_clone_index_column("person") in person: clone_groups = ["source_id"] else: clone_table["source_occurrence"] = clone_table.groupby( diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py b/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py index ec5a51eaf2..c2bbb134fd 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py @@ -13,18 +13,21 @@ __all__ = [ "BASE_ASEC_SUPPORT_CHANNEL", "PERSON_SUPPORT_CHANNEL_COLUMN", + "PUF_CAPITAL_GAINS_TAIL_CLONE_INDEX", "PUF_TAX_DETAIL_CLONE_INDEX", "PUF_TAX_DETAIL_SUPPORT_CHANNEL", "SPINE_ASSEMBLY_MANIFEST_KEY", "has_assembled_support_metadata", "has_support_role_metadata", "puf_tax_detail_clone_mask", + "require_assembled_support_provenance", "spine_assembly_manifest", "spine_assembly_receipt", "spine_provenance_counts", "spine_source_id_column", "support_channel_column", "support_clone_index_column", + "support_copy_rank_series", "support_role_series", "support_gate_source_channel_series", "support_source_id_column", @@ -37,8 +40,13 @@ PERSON_SUPPORT_CHANNEL_COLUMN = "person_support_channel" PUF_TAX_DETAIL_SUPPORT_CHANNEL = "puf_tax_detail" PUF_TAX_DETAIL_CLONE_INDEX = 1 +PUF_CAPITAL_GAINS_TAIL_CLONE_INDEX = 2 SPINE_ASSEMBLY_MANIFEST_KEY = "us_spine_assembly_manifest" _SPINE_ASSEMBLY_MANIFEST_VERSION = 1 +_INT64_MAX = int(np.iinfo(np.int64).max) +# 2**63 is exactly representable in float64 and is the first float past the +# int64 range, so every integral float strictly below it casts exactly. +_INT64_FLOAT_EXCLUSIVE_BOUND = float(2**63) class _ProvenanceSchema(Protocol): @@ -174,20 +182,11 @@ def validate_assembly_provenance( f"channel(s) {unknown}." ) - numeric_clone_indices = pd.to_numeric( - table[clone_index_column], - errors="coerce", + clone_indices = _validated_clone_indices( + table, + clone_index_column, + owner=f"{boundary}: assembly manifest provenance", ) - clone_indices = numeric_clone_indices.to_numpy(dtype=np.float64) - if ( - numeric_clone_indices.isna().any() - or (clone_indices < 0.0).any() - or not np.equal(clone_indices, np.floor(clone_indices)).all() - ): - raise ValueError( - f"{boundary}: assembly manifest provenance column " - f"{clone_index_column!r} must contain nonnegative integers." - ) expected_by_channel = raw_counts.get(entity) if ( not isinstance(expected_by_channel, Mapping) @@ -197,7 +196,7 @@ def validate_assembly_provenance( f"{boundary}: assembly manifest row counts for {entity!r} " "do not exactly cover its declared channels." ) - native = clone_indices == 0.0 + native = clone_indices == 0 actual_counts = { channel: int( np.count_nonzero( @@ -290,10 +289,14 @@ def spine_provenance_counts( for entity in frame.entities: table = frame.table(entity) source = table[support_channel_column(entity)].astype(str) - clone_index = pd.to_numeric( - table[support_clone_index_column(entity)], - errors="raise", - ).astype("int64") + clone_index = pd.Series( + _validated_clone_indices( + table, + support_clone_index_column(entity), + owner=f"{boundary}: assembly manifest provenance", + ), + index=table.index, + ) observed_clone_indices = sorted(int(value) for value in clone_index.unique()) counts[entity] = { "rows": int(len(table)), @@ -320,10 +323,16 @@ def has_support_role_metadata( *, entity: str, ) -> bool: - """Return whether clone-role or legacy support-role metadata is present.""" + """Return whether assembly or legacy support-role metadata is present. + + A raw spine ID still marks an assembled table after either role column + is lost. Role readers must validate it instead of taking a historical + no-metadata fallback. + """ return ( - support_clone_index_column(entity) in table + has_assembled_support_metadata(table, entity=entity) + or support_clone_index_column(entity) in table or support_channel_column(entity) in table ) @@ -343,6 +352,61 @@ def has_assembled_support_metadata( return spine_source_id_column(entity) in table +def require_assembled_support_provenance( + table: pd.DataFrame, + *, + entity: str, +) -> None: + """Refuse an assembled table whose support provenance is incomplete. + + Multispine assembly writes the raw spine-record ID together with the + assembly-unique source ID, support channel and clone index. Source IDs + must be complete so grouping cannot silently drop an unkeyed copy. + On an assembled table the channel names a physical source (for example + ``asec`` or ``acs``), not an operator + role, so only the clone index can tell a native row from its donor + copies. An assembled table that has lost a required column must not fall back + to the logic for historical tables without clone indices: channel-only + role ranks, ``(source, role)`` occurrence pairing, or one row per source + ID. That fallback hides copies that disagree. + + :func:`has_support_role_metadata` recognizes the raw spine ID even when + both role columns are missing, routing role readers through this check in + :func:`support_role_series`. Consumers with earlier fallbacks also call + it at entry. Historical tables without a raw spine ID pass unchanged. + """ + + if not has_assembled_support_metadata(table, entity=entity): + return + missing = [ + column + for column in ( + support_channel_column(entity), + support_clone_index_column(entity), + support_source_id_column(entity), + ) + if column not in table + ] + if missing: + raise ValueError( + "assembled support metadata requires " + + " and ".join(repr(column) for column in missing) + + f" alongside {spine_source_id_column(entity)!r}: an assembled " + "table's channel names a physical source rather than a support " + "copy, so without complete source, channel and clone-index provenance its " + "copies cannot be ranked, paired or compared." + ) + source_column = support_source_id_column(entity) + missing_sources = table[source_column].isna() + if missing_sources.any(): + raise ValueError( + "assembled support metadata requires complete non-null " + f"{source_column!r} alongside {spine_source_id_column(entity)!r}; " + f"found {int(missing_sources.sum())} missing source ID(s), so its " + "copies cannot be ranked, paired or compared." + ) + + def spine_source_id_column(entity: str) -> str: """Return the entity-prefixed raw spine-record ID metadata column.""" @@ -376,12 +440,19 @@ def without_support_role_metadata( *, entity: str, ) -> pd.DataFrame: - """Copy a table without source-channel or clone-role metadata.""" + """Project validated support rows into a historical source-kernel input. + + Remove the assembly discriminator with the role columns so deliberately + stripped projections are not mistaken for incomplete assembled tables. + Stable source IDs remain available for reconciliation and output merging. + """ + require_assembled_support_provenance(table, entity=entity) return table.drop( columns=[ support_channel_column(entity), support_clone_index_column(entity), + spine_source_id_column(entity), ], errors="ignore", ).copy(deep=True) @@ -397,9 +468,14 @@ def support_role_series( Native records have the ASEC-compatible role and every donor-detail clone has the PUF-compatible role. Clone provenance takes precedence. Legacy fixtures without clone indices may use the two historical role labels in - their support-channel column. + their support-channel column. An assembled table's channel names a + physical source, never a role, so an assembled table missing either + provenance column is refused first + (:func:`require_assembled_support_provenance`); its channel is never read + as a historical role. """ + require_assembled_support_provenance(table, entity=entity) clone_index_column = support_clone_index_column(entity) if clone_index_column not in table: channel_column = support_channel_column(entity) @@ -429,19 +505,11 @@ def support_role_series( index=table.index, name=f"{entity}_support_role", ) - numeric = pd.to_numeric(table[clone_index_column], errors="coerce") - if numeric.isna().any(): - raise ValueError( - f"PUF support metadata column {clone_index_column!r} must be integral." - ) - clone_indices = numeric.to_numpy(dtype=np.float64) - if (clone_indices < 0).any() or not np.equal( - clone_indices, np.floor(clone_indices) - ).all(): - raise ValueError( - f"PUF support metadata column {clone_index_column!r} must contain " - "nonnegative integers." - ) + clone_indices = _validated_clone_indices( + table, + clone_index_column, + owner="PUF support metadata", + ) channel_column = support_channel_column(entity) if channel_column not in table: raise ValueError( @@ -493,6 +561,66 @@ def support_role_series( ) +def support_copy_rank_series( + table: pd.DataFrame, + *, + entity: str, +) -> pd.Series: + """Rank each support copy within its source unit, canonical copy first. + + Clone provenance is authoritative whenever it is present, on assembled and + historical PUF-support frames alike: the rank is the validated clone index. + Historical ranks are restricted to the produced indices 0, 1 and 2; + assembled ranks retain the full nonnegative int64 domain. + Index 0 is the native record, 1 its primary PUF-detail copy, and 2 the + capital-gains own-tail copy, which splits weight off a PUF-detail household + while keeping that household's source IDs. A historical PUF-support base + therefore carries two PUF-role copies of each tail source unit that differ + only in clone index, so a ``(source ID, role)`` key cannot identify a copy. + Channel-only fixtures predate clone indices and cannot express a tail copy; + they rank their two historical roles ASEC 0 and PUF 1. + + A source unit may carry at most one row per rank. Source-unit operators + refuse a repeated ``(source ID, rank)`` pair as a genuinely duplicated + support copy instead of choosing between its rows. + + An assembled table missing its channel or clone index is refused first + (:func:`require_assembled_support_provenance`): its channel names a + physical source, so it cannot fall back to role ranks. Only a channel-only + historical table ranks by role. + """ + + require_assembled_support_provenance(table, entity=entity) + roles = support_role_series(table, entity=entity) + clone_index_column = support_clone_index_column(entity) + if clone_index_column in table: + ranks = _validated_clone_indices( + table, + clone_index_column, + owner="PUF support metadata", + ) + if not has_assembled_support_metadata(table, entity=entity): + allowed = ( + 0, + PUF_TAX_DETAIL_CLONE_INDEX, + PUF_CAPITAL_GAINS_TAIL_CLONE_INDEX, + ) + invalid = ~np.isin(ranks, allowed) + if invalid.any(): + raise ValueError( + "historical support clone indices must be one of 0, 1, 2; " + f"{clone_index_column!r} contains unsupported value(s) " + f"{np.unique(ranks[invalid]).tolist()}." + ) + else: + ranks = np.where( + roles.eq(BASE_ASEC_SUPPORT_CHANNEL).to_numpy(), + 0, + PUF_TAX_DETAIL_CLONE_INDEX, + ).astype(np.int64) + return pd.Series(ranks, index=table.index, name=f"{entity}_support_copy_rank") + + def support_gate_source_channel_series( table: pd.DataFrame, *, @@ -535,10 +663,11 @@ def puf_tax_detail_clone_mask( clone_index_column = support_clone_index_column(entity) if clone_index_column not in table: return roles.eq(PUF_TAX_DETAIL_SUPPORT_CHANNEL).to_numpy() - clone_indices = pd.to_numeric( - table[clone_index_column], - errors="raise", - ).to_numpy(dtype=np.int64) + clone_indices = _validated_clone_indices( + table, + clone_index_column, + owner="PUF support metadata", + ) return clone_indices == PUF_TAX_DETAIL_CLONE_INDEX @@ -547,6 +676,52 @@ def _require_entity_name(entity: str) -> None: raise ValueError("entity must be a non-empty string.") +def _validated_clone_indices( + table: pd.DataFrame, + column: str, + *, + owner: str, +) -> np.ndarray: + """Return a clone-index column as int64 after proving every value casts. + + This is the module's only clone-index conversion, and validation precedes + the cast. NumPy maps NaN, infinities and out-of-range floats to an + arbitrary int64 (INT64_MIN or INT64_MAX, by platform) with at most a + RuntimeWarning, so an unchecked cast would turn malformed provenance into + a plausible clone role or copy rank. Every value must be a finite, + nonnegative integer representable as int64. + """ + + values = table[column] + numeric = pd.to_numeric(values, errors="coerce") + integer_dtype = pd.api.types.is_integer_dtype(numeric.dtype) + if integer_dtype: + # Compare integer dtypes exactly (including nullable and unsigned + # ones); a float64 view would round values past 2**53 before the + # representability check. + valid = ( + (numeric.ge(0) & numeric.le(_INT64_MAX)).fillna(False).to_numpy(dtype=bool) + ) + else: + as_float = numeric.to_numpy(dtype=np.float64, na_value=np.nan) + valid = ( + np.isfinite(as_float) + & (as_float >= 0.0) + & (as_float < _INT64_FLOAT_EXCLUSIVE_BOUND) + & np.equal(as_float, np.floor(as_float)) + ) + if not valid.all(): + invalid = values.to_numpy(dtype=object)[~valid] + raise ValueError( + f"{owner} column {column!r} must contain nonnegative integers " + f"(finite and representable as int64); found {invalid.size} " + f"invalid value(s), e.g. {invalid[:5].tolist()}." + ) + if integer_dtype: + return numeric.to_numpy(dtype=np.int64) + return as_float.astype(np.int64) + + def _json_ready_mapping(value: Mapping[str, Any]) -> dict[str, object]: """Deep-copy frozen receipt values into JSON-compatible containers.""" diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/voluntary_filing.py b/packages/microcosm-build/src/microcosm/build/us_runtime/voluntary_filing.py index b19d32dda1..e15ea5702d 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/voluntary_filing.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/voluntary_filing.py @@ -38,9 +38,10 @@ from microcosm.build.gates import GateResult from microcosm.build.source_manifest import SourceStageSpec, load_source_manifest from microcosm.build.us_runtime.support_provenance import ( - has_assembled_support_metadata, has_support_role_metadata, + require_assembled_support_provenance, support_clone_index_column, + support_copy_rank_series, support_role_series, ) from microcosm.frame import Frame @@ -131,7 +132,6 @@ _DONOR_WEIGHT_COLUMN = "tax_unit_weight" _DONOR_SOURCE_KEY_COLUMN = "source_tax_unit_key" _DEFAULT_N_ESTIMATORS = 100 -_BASE_ASEC_SUPPORT_CHANNEL = "asec" _TAX_UNIT_SOURCE_ID_COLUMN = "tax_unit_source_id" _TRUE_SHARE_BAND = (0.45, 0.95) _EXPECTED_READ_PARAMETERS: dict[str, object] = { @@ -721,9 +721,17 @@ def _source_receiver_rows( frame: Frame, receiver: pd.DataFrame, ) -> tuple[pd.DataFrame, pd.Series]: - """Select one ASEC predictor row per source unit and return fan-out keys.""" + """Select one canonical predictor row per source unit and fan-out keys. + + The canonical row is the lowest surviving support-copy rank: the native + ASEC copy when present, then the primary PUF-detail copy, then the + capital-gains own-tail copy. An assembled table missing a provenance + column is refused first, so it can reach neither the copy ranking nor the + one-row-per-source path for tables without support metadata. + """ tax_unit = frame.table("tax_unit") + require_assembled_support_provenance(tax_unit, entity="tax_unit") if _TAX_UNIT_SOURCE_ID_COLUMN in tax_unit: source_ids = tax_unit[_TAX_UNIT_SOURCE_ID_COLUMN] if source_ids.isna().any(): @@ -741,80 +749,32 @@ def _source_receiver_rows( rows["_source_id"] = _decoded_strings(source_ids).to_numpy() rows["_tax_unit_id"] = tax_unit["tax_unit_id"].to_numpy() if has_support_role_metadata(tax_unit, entity="tax_unit"): - rows["_support_role"] = support_role_series( + rows["_copy_rank"] = support_copy_rank_series( tax_unit, entity="tax_unit" ).to_numpy() rows["_source_key"] = rows["_source_id"] - ordered_rows = rows.copy() - if has_assembled_support_metadata(tax_unit, entity="tax_unit"): - clone_column = support_clone_index_column("tax_unit") - ordered_rows["_clone_index"] = pd.to_numeric( - tax_unit[clone_column], errors="raise" - ).to_numpy(dtype=np.int64) - clone_counts = ordered_rows.groupby( - ["_source_id", "_clone_index"], - sort=False, - ).size() - duplicated_clones = clone_counts[clone_counts > 1] - if not duplicated_clones.empty: - bad = duplicated_clones.index.tolist() - raise ValueError( - "US voluntary-filing assembled source units carry " - "duplicated clone-index rows; invalid source clone(s) " - f"{bad[:5]}." - ) - source_rows = ( - ordered_rows.sort_values( - ["_source_id", "_clone_index", "_tax_unit_id"], - kind="stable", - ) - .drop_duplicates("_source_key", keep="first") - .drop(columns="_clone_index") - ) - else: - role_counts = rows.groupby( - ["_source_id", "_support_role"], - sort=False, - ).size() - duplicated_roles = role_counts[role_counts > 1] - if not duplicated_roles.empty: - bad = duplicated_roles.index.tolist() - raise ValueError( - "US voluntary-filing support source units carry duplicated " - f"same-role rows; invalid source role(s) {bad[:5]}." - ) - asec_counts = ( - rows["_support_role"] - .eq(_BASE_ASEC_SUPPORT_CHANNEL) - .groupby(rows["_source_key"]) - .sum() - ) - if asec_counts.gt(1).any(): - bad = asec_counts.index[asec_counts.gt(1)].tolist() - raise ValueError( - "US voluntary-filing support source units carry duplicated " - f"ASEC rows; invalid source unit(s) {bad[:5]}." - ) - # Prefer each unit's ASEC row, but a frozen-support selection may - # legitimately keep only a unit's PUF clone. Clones carry the - # unit's source predictors, so the surviving row predicts - # identically; pick it deterministically by role then tax-unit id. - ordered_rows["_asec_rank"] = ( - ~ordered_rows["_support_role"].eq(_BASE_ASEC_SUPPORT_CHANNEL) - ).astype(int) - source_rows = ( - ordered_rows.sort_values( - [ - "_source_id", - "_asec_rank", - "_support_role", - "_tax_unit_id", - ], - kind="stable", - ) - .drop_duplicates("_source_key", keep="first") - .drop(columns="_asec_rank") + copy_counts = rows.groupby( + ["_source_id", "_copy_rank"], + sort=False, + ).size() + duplicated_copies = copy_counts[copy_counts > 1] + if not duplicated_copies.empty: + bad = duplicated_copies.index.tolist() + raise ValueError( + "US voluntary-filing support source units carry duplicated " + f"support copies; invalid (source, copy rank) pair(s) {bad[:5]}." ) + # Prefer each unit's native copy. A frozen-support selection may keep + # only its PUF-role copies (the primary PUF-detail copy, the + # capital-gains own-tail copy, or both); the lowest surviving copy rank + # then supplies the unit's predictors, deterministically, and its one + # decision fans out to every copy. Beyond IDs, weight and provenance, + # the tail transfer writes only the capital-gains vector onto its + # copy, and none of that vector is a predictor here. + source_rows = rows.sort_values( + ["_source_id", "_copy_rank", "_tax_unit_id"], + kind="stable", + ).drop_duplicates("_source_key", keep="first") else: if rows["_source_id"].duplicated().any(): duplicates = rows.loc[ @@ -993,6 +953,7 @@ def us_voluntary_filing_summary(frame: Frame) -> dict[str, object]: """Return weighted incidence, boolean validity, and clone diagnostics.""" tax_unit = frame.table("tax_unit") + require_assembled_support_provenance(tax_unit, entity="tax_unit") values = pd.to_numeric(tax_unit[_OUTPUT], errors="coerce").to_numpy( dtype=np.float64 ) @@ -1041,19 +1002,30 @@ def us_voluntary_filing_summary(frame: Frame) -> dict[str, object]: "value": values, } ) - if has_assembled_support_metadata(tax_unit, entity="tax_unit"): - clone_groups = ["source_id"] - else: - clone_table["source_occurrence"] = clone_table.groupby( + if support_clone_index_column("tax_unit") not in tax_unit: + role_counts = clone_table.groupby( ["source_id", "role"], sort=False - ).cumcount() - clone_groups = ["source_id", "source_occurrence"] - sizes = clone_table.groupby(clone_groups, sort=False).size() + ).size() + duplicated_roles = role_counts[role_counts > 1] + if not duplicated_roles.empty: + raise ValueError( + "US voluntary-filing support source units carry " + "duplicated same-role rows; invalid source role(s) " + f"{duplicated_roles.index.tolist()[:5]}." + ) + # Every source copy must agree. Channel-only tables have one + # row per source and role, as required by the receiver path. + sizes = clone_table.groupby("source_id", sort=False).size() clone_source_units = int((sizes > 1).sum()) - unique = clone_table.groupby(clone_groups, sort=False)["value"].nunique( + unique = clone_table.groupby("source_id", sort=False)["value"].nunique( dropna=False ) clone_mismatch_source_units = int((unique > 1).sum()) + elif _TAX_UNIT_SOURCE_ID_COLUMN in tax_unit: + source_ids = tax_unit[_TAX_UNIT_SOURCE_ID_COLUMN] + clone_metadata_missing = bool( + source_ids.isna().any() or _decoded_strings(source_ids).duplicated().any() + ) return { "weighted_true_share": float(weights[true].sum()) / total_weight, @@ -1074,6 +1046,7 @@ def us_voluntary_filing_signal_gate(frame: Frame) -> GateResult: """Require nonconstant boolean signal and identical support clones.""" tax_unit = frame.table("tax_unit") + require_assembled_support_provenance(tax_unit, entity="tax_unit") if _OUTPUT not in tax_unit: return GateResult( name="voluntary_filing_signal", @@ -1108,7 +1081,7 @@ def us_voluntary_filing_signal_gate(frame: Frame) -> GateResult: if summary["clone_metadata_missing"]: failures.append( "Voluntary-filing support clones lack complete tax_unit_source_id " - "provenance." + "or support-role provenance." ) if summary["clone_mismatch_source_units"]: failures.append( diff --git a/packages/microcosm-build/tests/test_spec_engine_loader.py b/packages/microcosm-build/tests/test_spec_engine_loader.py index 8abeb60861..f731de99e8 100644 --- a/packages/microcosm-build/tests/test_spec_engine_loader.py +++ b/packages/microcosm-build/tests/test_spec_engine_loader.py @@ -236,7 +236,7 @@ def test_semantic_hash_has_golden_vector_and_surface_separation(tmp_path) -> Non # Pin the domain separator, normalization rules, schema-set receipt, and # exact normative projection as one reviewable golden vector. assert first.spec_sha256 == ( - "a2f0b3cd33fbd16e7f9ab44099de2abc579c71f0700137d949c9299f6ca335cd" + "db3fdab1b4f61040ee3f6c6b4ada65435ab130f67f181cb9e80688d40f4b7a9a" ) second_root = _rich_minimal(tmp_path / "xy", note="second", store="local:b") diff --git a/packages/microcosm-build/tests/test_us_alimony.py b/packages/microcosm-build/tests/test_us_alimony.py index 25aefd15c8..c417225b43 100644 --- a/packages/microcosm-build/tests/test_us_alimony.py +++ b/packages/microcosm-build/tests/test_us_alimony.py @@ -398,6 +398,9 @@ def _stacked_alimony_person() -> pd.DataFrame: return pd.DataFrame( { "person_spine_source_id": source_numbers, + # Spine assembly writes the assembly-unique source ID beside the + # raw spine ID; a support clone keeps its source's ID. + "person_source_id": source_numbers, "person_support_channel": source_channels, "person_support_clone_index": clone_indices, "OI_OFF": codes, diff --git a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py index 1667756066..d79c29ffca 100644 --- a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py +++ b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py @@ -2506,7 +2506,7 @@ def capture_equality(expected: object, actual: object) -> None: "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "1c46dda9cff4e53e5a44c6502a46fa695d02d92731f23d618616a892071d4841", + "spec_sha256": "3667e30ebfdac4d893e3a111bfba998dc7acf29b0da05360eee521d3e144d08a", }, } diff --git a/packages/microcosm-build/tests/test_us_multispine_puf_clone.py b/packages/microcosm-build/tests/test_us_multispine_puf_clone.py index d9b3dc7b1b..fe34817067 100644 --- a/packages/microcosm-build/tests/test_us_multispine_puf_clone.py +++ b/packages/microcosm-build/tests/test_us_multispine_puf_clone.py @@ -2,6 +2,10 @@ from __future__ import annotations +import warnings +from collections.abc import Callable +from typing import Any + import numpy as np import pandas as pd import pytest @@ -21,8 +25,11 @@ support_source_id_column, ) from microcosm.build.us_runtime.support_provenance import ( + require_assembled_support_provenance, spine_assembly_manifest, + support_copy_rank_series, support_gate_source_channel_series, + without_support_role_metadata, ) from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights @@ -208,3 +215,338 @@ def test_support_role_legacy_fallback_is_closed_to_known_roles() -> None: with pytest.raises(ValueError, match="complete support provenance"): support_role_series(legacy, entity="person") assert not has_support_role_metadata(pd.DataFrame(), entity="person") + + +def test_support_copy_rank_uses_clone_index_on_every_frame_kind() -> None: + # A historical PUF-support base with a capital-gains own-tail copy: two + # PUF-role rows of source 7 that only the clone index tells apart. + historical = pd.DataFrame( + { + support_source_id_column("person"): [7, 7, 7, 8, 8], + support_channel_column("person"): [ + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + ], + support_clone_index_column("person"): [0, 1, 2, 0, 1], + }, + index=[10, 11, 12, 13, 14], + ) + assert not has_assembled_support_metadata(historical, entity="person") + assert support_role_series(historical, entity="person").tolist() == [ + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + ] + ranks = support_copy_rank_series(historical, entity="person") + assert ranks.tolist() == [0, 1, 2, 0, 1] + assert ranks.index.tolist() == [10, 11, 12, 13, 14] + assert ranks.dtype == np.int64 + keyed = pd.DataFrame( + {"source": historical[support_source_id_column("person")], "rank": ranks} + ) + assert not keyed.duplicated().any() + + assembled = historical.assign( + **{ + spine_source_id_column("person"): [1, 1, 1, 2, 2], + support_channel_column("person"): ["acs"] * 5, + } + ) + assert support_copy_rank_series(assembled, entity="person").tolist() == [ + 0, + 1, + 2, + 0, + 1, + ] + + channel_only = historical.drop(columns=[support_clone_index_column("person")]) + assert support_copy_rank_series(channel_only, entity="person").tolist() == [ + 0, + 1, + 1, + 0, + 1, + ] + + # Rank inherits the role validator: a clone index must agree with its + # historical channel, and metadata must be present. + inconsistent = historical.copy() + inconsistent.loc[12, support_clone_index_column("person")] = 0 + with pytest.raises(ValueError, match="inconsistent"): + support_copy_rank_series(inconsistent, entity="person") + with pytest.raises(ValueError, match="missing both"): + support_copy_rank_series(pd.DataFrame({"x": [1]}), entity="person") + + +def _assembled_copies(*, channel: bool, clone_index: bool) -> pd.DataFrame: + """Two assembled source people, each with a native and a donor copy.""" + + table = pd.DataFrame( + { + support_source_id_column("tax_unit"): [7, 7, 8, 8], + spine_source_id_column("tax_unit"): [70, 70, 80, 80], + }, + index=[30, 31, 32, 33], + ) + if channel: + table[support_channel_column("tax_unit")] = ["asec"] * 4 + if clone_index: + table[support_clone_index_column("tax_unit")] = [0, 1, 0, 1] + return table + + +@pytest.mark.parametrize( + ("channel", "clone_index", "missing"), + [ + pytest.param( + False, + False, + "'tax_unit_support_channel' and 'tax_unit_support_clone_index'", + id="both_missing", + ), + pytest.param( + True, False, "'tax_unit_support_clone_index'", id="clone_index_missing" + ), + pytest.param(False, True, "'tax_unit_support_channel'", id="channel_missing"), + ], +) +def test_assembled_support_provenance_requires_channel_and_clone_index( + channel: bool, + clone_index: bool, + missing: str, +) -> None: + # Microcosm #992 gate findings: assembled tables that lost a provenance + # column fell back to channel-only role ranks, (source, role) occurrence + # pairing or the no-metadata one-row-per-source path, each of which hid a + # support-copy disagreement. One owner-level validator refuses them all. + # The role reader calls it first, so every role reader refuses, including + # the prior-year and SSI summaries; the copy rank also calls it before its + # role-rank fallback, and the Head Start and voluntary-filing consumers + # call it at entry (pinned in the consumer tests). + table = _assembled_copies(channel=channel, clone_index=clone_index) + assert has_support_role_metadata(table, entity="tax_unit") + message = ( + f"assembled support metadata requires {missing} alongside " + "'tax_unit_spine_source_id': " + ) + for accessor in ( + require_assembled_support_provenance, + support_role_series, + support_copy_rank_series, + puf_tax_detail_clone_mask, + support_gate_source_channel_series, + without_support_role_metadata, + ): + with pytest.raises(ValueError) as refused: + accessor(table, entity="tax_unit") + assert str(refused.value).startswith(message), accessor.__name__ + + +def test_assembled_support_provenance_accepts_complete_and_historical_tables() -> None: + complete = _assembled_copies(channel=True, clone_index=True) + assert require_assembled_support_provenance(complete, entity="tax_unit") is None + assert support_copy_rank_series(complete, entity="tax_unit").tolist() == [ + 0, + 1, + 0, + 1, + ] + roles = support_role_series(complete, entity="tax_unit") + assert roles.tolist() == ["asec", "puf_tax_detail"] * 2 + channels = support_gate_source_channel_series(complete, entity="tax_unit") + assert channels.tolist() == ["asec"] * 4 + # Without a raw spine ID the table is historical: clone-index, channel-only + # and metadata-free layouts are all left to their own validators. + for channel, clone_index in ((True, True), (True, False), (False, False)): + historical = _assembled_copies(channel=channel, clone_index=clone_index).drop( + columns=[spine_source_id_column("tax_unit")] + ) + assert ( + require_assembled_support_provenance(historical, entity="tax_unit") is None + ) + with pytest.raises(ValueError, match="entity must be a non-empty string"): + require_assembled_support_provenance(complete, entity="") + + +def test_support_projection_removes_assembly_marker_without_mutating_input() -> None: + table = _assembled_copies(channel=True, clone_index=True) + original = table.copy(deep=True) + projected = without_support_role_metadata(table, entity="tax_unit") + assert not has_support_role_metadata(projected, entity="tax_unit") + assert not has_assembled_support_metadata(projected, entity="tax_unit") + pd.testing.assert_frame_equal( + projected, table[[support_source_id_column("tax_unit")]] + ) + pd.testing.assert_frame_equal(table, original) + + +_CLONE_INDEX_ACCESSORS = pytest.mark.parametrize( + "accessor", + [support_role_series, support_copy_rank_series, puf_tax_detail_clone_mask], + ids=["role", "copy_rank", "puf_detail_mask"], +) +_SUPPORT_FRAME_KINDS = pytest.mark.parametrize( + "assembled", + [True, False], + ids=["assembled", "historical_puf_support"], +) +_MALFORMED_CLONE_INDEX = ( + r"PUF support metadata column 'person_support_clone_index' must contain " + r"nonnegative integers \(finite and representable as int64\)" +) + + +def _support_copies(clone_indices: Any, *, assembled: bool) -> pd.DataFrame: + """Support copies of source 7: its native row, then PUF-role copies. + + Historical PUF-support frames carry the exact ASEC/PUF roles as channels; + assembled frames carry a raw spine ID and a receipt-declared channel. + """ + + count = len(clone_indices) + table = pd.DataFrame( + { + support_source_id_column("person"): [7] * count, + support_channel_column("person"): [ + BASE_ASEC_SUPPORT_CHANNEL, + *[PUF_TAX_DETAIL_SUPPORT_CHANNEL] * (count - 1), + ], + support_clone_index_column("person"): clone_indices, + }, + index=[20 + position for position in range(count)], + ) + if assembled: + table[spine_source_id_column("person")] = [1] * count + table[support_channel_column("person")] = ["acs"] * count + return table + + +@_CLONE_INDEX_ACCESSORS +@_SUPPORT_FRAME_KINDS +@pytest.mark.parametrize( + "clone_indices", + [ + pytest.param([0, np.inf], id="inf"), + pytest.param([0, -np.inf], id="negative_inf"), + pytest.param([0, np.nan], id="nan"), + pytest.param([0, 1.5], id="non_integer"), + pytest.param([0, -1], id="negative"), + # A finite negative float takes the float branch, not the integer + # one, so it pins that branch's own nonnegativity check. + pytest.param([0.0, -1.0], id="negative_float"), + pytest.param([0, 2.0**63], id="float_past_int64"), + pytest.param([0, "inf"], id="string_inf"), + pytest.param(pd.array([0, pd.NA], dtype="Int64"), id="nullable_missing"), + pytest.param( + np.asarray([0, 2**63], dtype=np.uint64), + id="unsigned_past_int64", + ), + ], +) +def test_clone_index_conversions_reject_malformed_provenance_before_casting( + accessor: Callable[..., object], + assembled: bool, + clone_indices: Any, +) -> None: + # Microcosm #992 gate finding: a raw NumPy cast maps +inf (and any float + # past the int64 range) to an arbitrary int64 with only a RuntimeWarning, + # so [0, inf] once passed Head Start canonical selection as a copy rank. + # Escalating that warning proves the refusal happens before any cast. + table = _support_copies(clone_indices, assembled=assembled) + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + with pytest.raises(ValueError, match=_MALFORMED_CLONE_INDEX): + accessor(table, entity="person") + + +@_SUPPORT_FRAME_KINDS +@pytest.mark.parametrize( + ("clone_indices", "expected_ranks"), + [ + pytest.param(np.asarray([0, 1, 2], dtype=np.int64), [0, 1, 2], id="int64"), + pytest.param(np.asarray([0.0, 1.0, 2.0]), [0, 1, 2], id="float64"), + pytest.param(pd.array([0, 1, 2], dtype="Int64"), [0, 1, 2], id="nullable"), + pytest.param(np.asarray([0, 1, 2], dtype=np.uint64), [0, 1, 2], id="uint64"), + pytest.param(["0", "1", "2"], [0, 1, 2], id="numeric_strings"), + ], +) +def test_clone_index_conversions_leave_valid_encodings_unchanged( + assembled: bool, + clone_indices: Any, + expected_ranks: list[int], +) -> None: + table = _support_copies(clone_indices, assembled=assembled) + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + roles = support_role_series(table, entity="person") + ranks = support_copy_rank_series(table, entity="person") + mask = puf_tax_detail_clone_mask(table, entity="person") + + assert roles.tolist() == [ + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + ] + assert ranks.tolist() == expected_ranks + assert ranks.dtype == np.int64 + assert ranks.index.tolist() == table.index.tolist() + assert mask.tolist() == [False, True, False] + + +@pytest.mark.parametrize("clone_index", [3, 7, 2**53 + 1, 2**62]) +@pytest.mark.parametrize("dtype", [np.int64, np.float64]) +def test_copy_rank_restricts_historical_domain_but_preserves_assembled_indices( + clone_index: int, + dtype: type, +) -> None: + indices = np.asarray([0, 1, clone_index], dtype=dtype) + historical = _support_copies(indices, assembled=False) + with pytest.raises(ValueError, match="historical support clone indices.*0, 1, 2"): + support_copy_rank_series(historical, entity="person") + + assembled = _support_copies(indices, assembled=True) + ranks = support_copy_rank_series(assembled, entity="person") + assert ranks.tolist() == indices.astype(np.int64).tolist() + assert ranks.dtype == np.int64 + + +@pytest.mark.parametrize("entity", US_SCHEMA.entities) +@pytest.mark.parametrize("source_state", ["missing", "nan", "none", "nullable"]) +def test_assembled_provenance_requires_complete_source_ids( + entity: str, + source_state: str, +) -> None: + table = _support_copies([0, 1, 2], assembled=True).rename( + columns=lambda column: column.replace("person_", f"{entity}_", 1) + ) + source = support_source_id_column(entity) + if source_state == "missing": + table = table.drop(columns=source) + else: + table[source] = { + "nan": [np.nan, np.nan, 7], + "none": [None, None, 7], + "nullable": pd.array([pd.NA, pd.NA, 7], dtype="Int64"), + }[source_state] + for reader in ( + require_assembled_support_provenance, + support_role_series, + support_copy_rank_series, + without_support_role_metadata, + ): + with pytest.raises( + ValueError, match=f"assembled support metadata requires.*{source}" + ): + reader(table, entity=entity) + + +def test_assembled_copy_rank_preserves_int64_maximum() -> None: + indices = np.asarray([0, 1, np.iinfo(np.int64).max], dtype=np.int64) + table = _support_copies(indices, assembled=True) + assert support_copy_rank_series(table, entity="person").tolist() == indices.tolist() diff --git a/packages/microcosm-build/tests/test_us_prior_year_income.py b/packages/microcosm-build/tests/test_us_prior_year_income.py index 8cd06d53d4..841cb632a1 100644 --- a/packages/microcosm-build/tests/test_us_prior_year_income.py +++ b/packages/microcosm-build/tests/test_us_prior_year_income.py @@ -47,6 +47,18 @@ from microcosm.frame.adapters.policyengine_us import PolicyEngineUSEngine policyengine_us_installed = importlib.util.find_spec("policyengine_us") is not None + + +def _load_tail_fixtures(): + path = Path(__file__).with_name("us_tail_clone_fixtures.py") + spec = importlib.util.spec_from_file_location("us_tail_clone_fixtures", path) + fixtures = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(fixtures) + return fixtures + + +_TAIL = _load_tail_fixtures() requires_us = pytest.mark.skipif( not policyengine_us_installed, reason="requires the policyengine-us [us] extra (build environment)", @@ -411,10 +423,7 @@ def _with_stack_manifest( return Frame( {entity: frame.table(entity).copy() for entity in frame.entities}, frame.schema, - { - entity: frame.weights_for(entity) - for entity in frame.weighted_entities - }, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, frame.strata, mass_log=frame.mass_log, metadata={"us_stacked_spine_manifest": manifest}, @@ -452,9 +461,9 @@ def test_sampled_rung_scales_only_prior_year_availability_floor() -> None: assert gate.details[ "previous_year_income_available_sampled_match_survival_factor" ] == pytest.approx(0.25) - assert gate.details["previous_year_income_available_applied_floor"] == pytest.approx( - 0.0125 - ) + assert gate.details[ + "previous_year_income_available_applied_floor" + ] == pytest.approx(0.0125) assert gate.details["previous_year_income_available_applied_share_band"] == [ 0.0125, 0.50, @@ -504,7 +513,9 @@ def manifest_bytes(gate) -> bytes: assert full_rung == legacy assert manifest_bytes(full_rung) == manifest_bytes(legacy) - assert not any("applied" in key or "survival_factor" in key for key in full_rung.details) + assert not any( + "applied" in key or "survival_factor" in key for key in full_rung.details + ) def test_legacy_acs_only_sampling_does_not_scale_asec_match_floor() -> None: @@ -576,6 +587,220 @@ def test_clone_availability_checks_all_assembled_clones_and_legacy_pairs() -> No assert legacy_summary["clone_availability_mismatches"] == 0 +def test_clone_availability_checks_the_historical_tail_copy() -> None: + native = _frame( + pd.DataFrame( + { + "self_employment_income_last_year": [10.0, -5.0, 0.0], + "previous_year_income_available": [True, False, True], + } + ) + ) + tailed = _TAIL.with_capital_gains_tail_copies( + clone_us_frame_for_puf_support(native), [1, 3] + ) + person = tailed.table("person") + assert "person_spine_source_id" not in person + assert person["person_support_clone_index"].tolist() == [0, 0, 0, 1, 1, 1, 2, 2] + + agreeing = module.us_prior_year_income_summary(tailed) + assert agreeing["clone_availability_mismatches"] == 0 + + # The own-tail copy is a PUF-role row of its source person: under the old + # (source, role-occurrence) pairing it sat alone and was never compared. + divergent_person = person.copy() + clone_index = divergent_person["person_support_clone_index"] + tail_of_source_one = clone_index.eq(2) & divergent_person["person_source_id"].eq(1) + divergent_person.loc[tail_of_source_one, "previous_year_income_available"] = False + divergent = module._replace_person_table(tailed, divergent_person) + summary = module.us_prior_year_income_summary(divergent) + assert summary["clone_availability_mismatches"] == 1 + gate = us_prior_year_income_signal_gate(divergent) + assert any("1 source person" in failure for failure in gate.failures) + + +def _verdict_flip_person(**overrides: object) -> pd.DataFrame: + """20 source persons with ASEC and PUF copies plus a disagreeing tail copy. + + Source person 1 carries a second PUF-role (own-tail) copy whose + availability disagrees with its siblings. Every other gate check passes, + so the clone comparison alone decides the verdict. + """ + + sources = [source for source in range(1, 21) for _ in range(2)] + [1] + channels = [BASE_ASEC_SUPPORT_CHANNEL, PUF_TAX_DETAIL_SUPPORT_CHANNEL] * 20 + [ + PUF_TAX_DETAIL_SUPPORT_CHANNEL + ] + available = [source <= 6 for source in sources] + available[-1] = False + person = pd.DataFrame( + { + "person_source_id": sources, + "person_support_channel": channels, + "self_employment_income_last_year": [ + (-50.0 if channel == BASE_ASEC_SUPPORT_CHANNEL else 100.0) + if source <= 3 + else 0.0 + for source, channel in zip(sources, channels, strict=True) + ], + "previous_year_income_available": available, + } + ) + for column, values in overrides.items(): + person[column] = values + return person + + +def test_gate_refuses_assembled_frame_without_clone_indices() -> None: + # Microcosm #992 gate finding: with raw spine IDs present but the + # clone-index column gone, the summary paired copies by (source, role) + # occurrence. The tail copy of source 1 sat alone, the mismatch count + # dropped from 1 to 0 and this gate flipped from FAIL to PASS. The + # summary reads roles before it compares copies, and the role reader now + # refuses an assembled table missing either provenance column. + sources = _verdict_flip_person()["person_source_id"].to_numpy() + assembled = _frame(_verdict_flip_person(person_spine_source_id=sources)) + pattern = ( + r"assembled support metadata requires 'person_support_clone_index' " + r"alongside 'person_spine_source_id'" + ) + with pytest.raises(ValueError, match=pattern): + module.us_prior_year_income_summary(assembled) + with pytest.raises(ValueError, match=pattern): + us_prior_year_income_signal_gate(assembled) + + # Same shape with every channel 'asec' (a physical source name): the + # role-label fallback must not read it as a role either. + all_asec = _frame( + _verdict_flip_person( + person_spine_source_id=sources, + person_support_channel=[BASE_ASEC_SUPPORT_CHANNEL] * len(sources), + ) + ) + with pytest.raises(ValueError, match=pattern): + us_prior_year_income_signal_gate(all_asec) + + +def _assembled_pair_person(**columns: object) -> pd.DataFrame: + return pd.DataFrame( + { + "person_source_id": [10, 10, 20, 20], + "person_spine_source_id": [1, 1, 2, 2], + "self_employment_income_last_year": [10.0, 10.0, -5.0, -5.0], + "previous_year_income_available": [False, True, True, True], + **columns, + } + ) + + +@pytest.mark.parametrize( + "columns", + [ + pytest.param({"person_support_channel": ["asec"] * 4}, id="all_asec"), + pytest.param( + {"person_support_channel": ["asec", "puf_tax_detail"] * 2}, + id="role_labels", + ), + pytest.param( + {"person_support_clone_index": [0, 1, 0, 1]}, + id="clone_index_without_channel", + ), + pytest.param({}, id="no_channel_no_clone"), + ], +) +def test_gate_refuses_assembled_pair_without_complete_support_provenance( + columns: dict[str, list[object]], +) -> None: + # Exact sweep regression: IDs [10, 10, 20, 20], availability [F, T, T, T]. + # Without clone indices, occurrence pairing split source 10's disagreeing + # copies. Even without either role column, a spine ID must route through + # the shared validator before any historical fallback can run. + frame = _frame(_assembled_pair_person(**columns)) + with pytest.raises(ValueError, match="assembled support metadata requires"): + module.us_prior_year_income_summary(frame) + with pytest.raises(ValueError, match="assembled support metadata requires"): + us_prior_year_income_signal_gate(frame) + + +def test_assembled_frame_with_complete_provenance_groups_every_copy() -> None: + # Complete assembled provenance: every copy of a source person joins one + # group, whatever its physical channel, so the disagreement is counted. + summary = module.us_prior_year_income_summary( + _frame( + _assembled_pair_person( + person_support_channel=["asec", "acs", "asec", "acs"], + person_support_clone_index=[0, 1, 0, 1], + ) + ) + ) + assert summary["clone_availability_mismatches"] == 1 + + +@pytest.mark.parametrize("source_state", ["missing", "nan", "nullable"]) +def test_prior_year_gate_refuses_assembled_missing_source_ids( + source_state: str, +) -> None: + person = _assembled_pair_person( + person_support_channel=["asec"] * 4, + person_support_clone_index=[0, 1, 0, 1], + ) + if source_state == "missing": + person = person.drop(columns="person_source_id") + elif source_state == "nan": + person["person_source_id"] = [np.nan, np.nan, 20, 20] + else: + person["person_source_id"] = pd.array([pd.NA, pd.NA, 20, 20], dtype="Int64") + frame = _frame(person) + for reader in ( + module.us_prior_year_income_summary, + us_prior_year_income_signal_gate, + ): + with pytest.raises( + ValueError, + match="assembled support metadata requires.*person_source_id", + ): + reader(frame) + + +def test_clone_availability_compares_repeated_historical_clone_index() -> None: + # A malformed historical table whose tail copy repeats clone index 1: + # grouping by source ID still compares that copy (the old role-occurrence + # pairing left it unpaired and hid the disagreement). + native = _frame( + pd.DataFrame( + { + "self_employment_income_last_year": [10.0, -5.0, 0.0], + "previous_year_income_available": [True, False, True], + } + ) + ) + tailed = _TAIL.with_capital_gains_tail_copies( + clone_us_frame_for_puf_support(native), [1] + ) + person = tailed.table("person").copy() + tail = person["person_support_clone_index"].eq(2) + person.loc[tail, "person_support_clone_index"] = 1 + person.loc[tail, "previous_year_income_available"] = False + duplicated = module._replace_person_table(tailed, person) + + summary = module.us_prior_year_income_summary(duplicated) + assert summary["clone_availability_mismatches"] == 1 + + +def test_gate_refuses_clone_index_past_int64() -> None: + # float(2**63) is the first float past int64; the base let it wrap to + # INT64_MAX and the gate ran normally. + person = _verdict_flip_person() + clone_index = [0, 1] * 20 + [2] + person["person_support_clone_index"] = np.asarray(clone_index, dtype=np.float64) + person.loc[len(person) - 1, "person_support_clone_index"] = float(2**63) + with pytest.raises( + ValueError, + match=r"must contain nonnegative integers \(finite and representable as int64\)", + ): + us_prior_year_income_signal_gate(_frame(person)) + + def test_source_reconciliation_detects_plausible_but_wrong_asec_carry() -> None: derived = with_us_prior_year_income_inputs( _source_frame(), seed=0, time_period=2024 @@ -595,6 +820,27 @@ def test_source_reconciliation_detects_plausible_but_wrong_asec_carry() -> None: } +def test_source_reconciliation_accepts_complete_assembled_provenance() -> None: + derived = with_us_prior_year_income_inputs( + _source_frame(), seed=0, time_period=2024 + ) + expanded = clone_us_frame_for_puf_support(derived) + person = expanded.table("person").copy() + person["person_spine_source_id"] = person["person_source_id"] + # Assembly channels identify physical sources; clone indices own roles. + person["person_support_channel"] = "asec" + assembled = module._replace_person_table(expanded, person) + + # Reconciliation strips provenance for its temporary source-only input. + # The raw spine ID must leave with the role columns, so that input is not + # mistaken for a malformed assembly when it replays the derivation. + gate = us_prior_year_income_source_reconciliation_gate(assembled) + + assert gate.passed, gate.failures + assert gate.details["asec_rows"] == len(derived.table("person")) + pd.testing.assert_frame_equal(assembled.table("person"), person) + + def test_release_contract_promotes_both_persisted_inputs_without_wage_formula() -> None: manifest = load_release_input_coverage_manifest() for column in US_PRIOR_YEAR_INCOME_PERSISTED_OUTPUT_COLUMNS: diff --git a/packages/microcosm-build/tests/test_us_qbi_inputs.py b/packages/microcosm-build/tests/test_us_qbi_inputs.py index 5317ba6e6a..cf805602bb 100644 --- a/packages/microcosm-build/tests/test_us_qbi_inputs.py +++ b/packages/microcosm-build/tests/test_us_qbi_inputs.py @@ -238,6 +238,8 @@ def _stacked_qbi_universe_frame(*, child_age: float = 12.0) -> Frame: ) person["person_support_clone_index"] = 0 person["person_spine_source_id"] = np.arange(len(person), dtype=np.int64) + # Spine assembly writes the assembly-unique source ID beside the raw one. + person["person_source_id"] = person["person_spine_source_id"] person["age"] = 40.0 person["SEMP"] = person["self_employment_income_before_lsr"] child = 10 diff --git a/packages/microcosm-build/tests/test_us_sipp_head_start.py b/packages/microcosm-build/tests/test_us_sipp_head_start.py index 8c3fa65180..5cd077c370 100644 --- a/packages/microcosm-build/tests/test_us_sipp_head_start.py +++ b/packages/microcosm-build/tests/test_us_sipp_head_start.py @@ -11,6 +11,7 @@ import pytest import microcosm.build.us_runtime.sipp_head_start as module +from microcosm.build.us_runtime.puf_support import clone_us_frame_for_puf_support from microcosm.build.us_runtime.sipp_head_start import ( HEAD_START_SIPP_DICTIONARY_URL, SIPP_2023_HEAD_START_DONOR_REVISION, @@ -34,6 +35,18 @@ from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights _OUTPUT = US_SIPP_HEAD_START_OUTPUT_COLUMNS[0] + + +def _load_tail_fixtures(): + path = Path(__file__).with_name("us_tail_clone_fixtures.py") + spec = importlib.util.spec_from_file_location("us_tail_clone_fixtures", path) + fixtures = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(fixtures) + return fixtures + + +_TAIL = _load_tail_fixtures() _policyengine_us_installed = importlib.util.find_spec("policyengine_us") is not None requires_us = pytest.mark.skipif( not _policyengine_us_installed, @@ -425,7 +438,7 @@ def test_imputer_rejects_identical_duplicate_same_role_source_rows( channels=["asec", "asec"], ) - with pytest.raises(ValueError, match="duplicated same-role rows"): + with pytest.raises(ValueError, match="duplicated support copies"): impute_us_sipp_head_start(duplicate, _donor(), seed=0) @@ -465,6 +478,347 @@ def test_assembled_clone_two_uses_lowest_clone_and_fans_to_every_clone( assert mismatch["clone_mismatch_count"] == 2 +def _historical_tail_frame() -> Frame: + """A non-assembled PUF-support frame whose source 10 has a tail copy.""" + + native = _frame( + [10, 20, 30], + ages=[4, 5, 40], + female=[True, False, True], + ) + native = _replace_person( + native, native.table("person").drop(columns=["person_source_id"]) + ) + cloned = clone_us_frame_for_puf_support(native) + tailed = _TAIL.with_capital_gains_tail_copies(cloned, [1]) + person = tailed.table("person").copy() + # Source IDs are the pre-clone person IDs; restate them as the test's + # source labels so assertions read in source-person terms. + person["person_source_id"] = person["person_source_id"].map({1: 10, 2: 20, 3: 30}) + return _replace_person(tailed, person) + + +def test_historical_tail_copy_fans_the_source_decision_to_every_copy( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(module, "QRF", _FakeQRF) + frame = _historical_tail_frame() + person = frame.table("person") + assert "person_spine_source_id" not in person + assert person["person_support_clone_index"].tolist() == [0, 0, 0, 1, 1, 1, 2] + assert person["person_support_channel"].tolist() == [ + "asec", + "asec", + "asec", + "puf_tax_detail", + "puf_tax_detail", + "puf_tax_detail", + "puf_tax_detail", + ] + + # Give the tail copy a predictor that would flip the fake QRF's decision: + # the native copy must still be the canonical predictor row. + divergent = person.copy() + divergent.loc[divergent["person_support_clone_index"].eq(2), "is_female"] = False + predicted = impute_us_sipp_head_start( + _replace_person(frame, divergent), _donor(), seed=3 + ) + receiver = _FakeQRF.instances[-1].receiver + assert receiver is not None + assert len(receiver) == 2 # one canonical row per eligible source person + by_source = pd.DataFrame( + {"source": person["person_source_id"], "value": predicted} + ).groupby("source")["value"] + assert (by_source.nunique() == 1).all() + assert by_source.first().to_dict() == {10: True, 20: False, 30: False} + tail = person["person_support_clone_index"].eq(2).to_numpy() + twin = ( + person["person_support_clone_index"].eq(1) & person["person_source_id"].eq(10) + ).to_numpy() + assert predicted[tail].tolist() == predicted[twin].tolist() == [True] + + materialized = person.copy() + materialized[_OUTPUT] = predicted.to_numpy() + summary = us_sipp_head_start_summary(_replace_person(frame, materialized)) + assert summary["clone_group_count"] == 3 + assert summary["clone_mismatch_count"] == 0 + materialized.loc[tail, _OUTPUT] = False + mismatch = us_sipp_head_start_summary(_replace_person(frame, materialized)) + assert mismatch["clone_mismatch_count"] == 1 + + +def test_historical_puf_only_survivor_predicts_from_the_primary_detail_copy( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(module, "QRF", _FakeQRF) + frame = _historical_tail_frame() + person = frame.table("person").copy() + # Selection kept only source 10's PUF-role copies, tail copy first. + survivor = ~( + person["person_source_id"].eq(10) & person["person_support_clone_index"].eq(0) + ) + person = person.loc[survivor].iloc[::-1].reset_index(drop=True) + person.loc[person["person_support_clone_index"].eq(2), "is_female"] = False + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables["person"] = person + kept_households = set(person["person_household_id"]) + tables["household"] = tables["household"].loc[ + tables["household"]["household_id"].isin(kept_households) + ] + weights = frame.weights_for("household").values[ + frame.table("household")["household_id"].isin(kept_households).to_numpy() + ] + for entity in ("tax_unit", "spm_unit", "family", "marital_unit"): + membership = f"person_{entity}_id" + tables[entity] = tables[entity].loc[ + tables[entity][f"{entity}_id"].isin(set(person[membership])) + ] + survivor_frame = Frame( + tables, + frame.schema, + {"household": Weights(weights, WeightKind.DESIGN)}, + ) + + predicted = impute_us_sipp_head_start(survivor_frame, _donor(), seed=3) + source_ten = person["person_source_id"].eq(10).to_numpy() + # Copy rank, not row order or role, picks the primary PUF-detail copy. + assert predicted[source_ten].tolist() == [True, True] + + +def test_historical_duplicate_clone_index_still_fails_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(module, "QRF", _FakeQRF) + frame = _historical_tail_frame() + person = frame.table("person").copy() + person.loc[ + person["person_support_clone_index"].eq(2), "person_support_clone_index" + ] = 1 + duplicated = _replace_person(frame, person) + + with pytest.raises(ValueError, match=r"duplicated support copies.*\('10', 1\)"): + impute_us_sipp_head_start(duplicated, _donor(), seed=3) + + +@pytest.mark.parametrize("clone_index", [3, 7, 2**62]) +@pytest.mark.parametrize("dtype", [np.int64, np.float64]) +def test_historical_out_of_domain_copy_refused_before_prediction( + monkeypatch: pytest.MonkeyPatch, + clone_index: int, + dtype: type, +) -> None: + monkeypatch.setattr(module, "QRF", _FakeQRF) + frame = _historical_tail_frame() + person = frame.table("person").copy() + column = "person_support_clone_index" + tail = person[column].eq(2) + person[column] = person[column].astype(dtype) + person.loc[tail, column] = clone_index + with pytest.raises(ValueError, match="historical support clone indices.*0, 1, 2"): + impute_us_sipp_head_start(_replace_person(frame, person), _donor(), seed=3) + assert not _FakeQRF.instances + + +@pytest.mark.parametrize("assembled", [True, False], ids=["assembled", "historical"]) +@pytest.mark.parametrize( + "bad_index", + [np.inf, -np.inf, np.nan, 1.5, -1.0, float(2**63)], + ids=[ + "inf", + "negative_inf", + "nan", + "non_integer", + "negative_float", + "float_past_int64", + ], +) +def test_malformed_clone_index_fails_closed_before_canonical_selection( + monkeypatch: pytest.MonkeyPatch, + assembled: bool, + bad_index: float, +) -> None: + # Microcosm #992 gate finding: an assembled [0, inf] once ranked as + # INT64_MAX and passed canonical selection; a historical tail copy set to + # inf was likewise accepted as a distinct PUF-role copy. A float -1.0 + # would rank ahead of the native row and silently replace it. + monkeypatch.setattr(module, "QRF", _FakeQRF) + column = "person_support_clone_index" + if assembled: + frame = _frame([10, 10], ages=[4, 4], channels=["acs", "acs"]) + person = frame.table("person").copy() + person["person_spine_source_id"] = [100, 100] + person[column] = [0.0, bad_index] + else: + frame = _historical_tail_frame() + person = frame.table("person").copy() + tail = person[column].eq(2) + person[column] = person[column].astype(np.float64) + person.loc[tail, column] = bad_index + + with pytest.raises( + ValueError, + match=( + r"clone-role metadata: PUF support metadata column " + r"'person_support_clone_index' must contain nonnegative integers" + ), + ): + impute_us_sipp_head_start(_replace_person(frame, person), _donor(), seed=3) + assert not _FakeQRF.instances + + +def test_assembled_frame_without_clone_indices_fails_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Microcosm #992 gate finding: with the clone-index column dropped, an + # assembled ASEC-channel frame once fell back to role ranks and passed + # canonical selection. Assembled channels name physical sources, so they + # cannot tell a native row from a donor copy; the base raised here. + monkeypatch.setattr(module, "QRF", _FakeQRF) + frame = _frame([10, 11], ages=[4, 4], channels=["asec", "asec"]) + person = frame.table("person").copy() + person["person_spine_source_id"] = [100, 101] + person = person.drop(columns=["person_support_clone_index"], errors="ignore") + + with pytest.raises( + ValueError, + match=( + r"assembled support metadata requires " + r"'person_support_clone_index'" + ), + ): + impute_us_sipp_head_start(_replace_person(frame, person), _donor(), seed=3) + assert not _FakeQRF.instances + + +@pytest.mark.parametrize( + ("source_ids", "clone_indices", "missing"), + [ + pytest.param( + [10, 20, 30, 40], + None, + r"'person_support_channel' and 'person_support_clone_index'", + id="no_channel_no_clone_unique_ids", + ), + pytest.param( + [10, 10, 20, 20], + None, + r"'person_support_channel' and 'person_support_clone_index'", + id="no_channel_no_clone_repeated_ids", + ), + pytest.param( + [10, 10, 20, 20], + [0, 1, 0, 1], + r"'person_support_channel'", + id="clone_index_without_channel", + ), + ], +) +def test_assembled_frame_missing_support_provenance_fails_closed( + monkeypatch: pytest.MonkeyPatch, + source_ids: list[int], + clone_indices: list[int] | None, + missing: str, +) -> None: + # Microcosm #992 gate finding (b): an assembled person table (raw spine + # IDs present) stripped of both provenance columns reached the + # no-metadata early return in _support_group_keys, ranked every row 0 and + # passed canonical selection; the base raised KeyError here. Unique IDs + # hid it best: nothing even looked duplicated. + monkeypatch.setattr(module, "QRF", _FakeQRF) + frame = _frame(source_ids, ages=[4] * len(source_ids)) + person = frame.table("person").copy() + person["person_spine_source_id"] = source_ids + if clone_indices is not None: + person["person_support_clone_index"] = clone_indices + stripped = _replace_person(frame, person) + + pattern = ( + r"assembled support metadata requires " + + missing + + r" alongside 'person_spine_source_id'" + ) + with pytest.raises(ValueError, match=pattern): + module._recipient_predictors(stripped) + with pytest.raises(ValueError, match=pattern): + impute_us_sipp_head_start(stripped, _donor(), seed=3) + assert not _FakeQRF.instances + + +@pytest.mark.parametrize("entry", ["predictors", "impute", "gate"]) +def test_assembled_provenance_checked_before_other_receiver_inputs( + monkeypatch: pytest.MonkeyPatch, entry: str +) -> None: + # Missing predictor/output inputs must not bypass the provenance check. + # The pre-fix predictors reported missing age, imputation read the donor, + # and the gate returned only the missing-output failure. + frame = _frame([10, 20, 30, 40]) + person = frame.table("person").copy() + person["person_spine_source_id"] = person["person_source_id"] + person = person.drop(columns=["age"]) + stripped = _replace_person(frame, person) + message = "assembled support metadata requires" + if entry == "gate": + gate = us_sipp_head_start_signal_gate(stripped) + assert not gate.passed + assert any(message in failure for failure in gate.failures) + assert gate.details["support_channel_invalid"] is True + else: + monkeypatch.setattr(module, "QRF", _FakeQRF) + with pytest.raises(ValueError, match=message): + if entry == "predictors": + module._recipient_predictors(stripped) + else: + impute_us_sipp_head_start(stripped, pd.DataFrame(), seed=3) + assert not _FakeQRF.instances + + +@pytest.mark.parametrize( + "channels", + [None, ["asec"] * 4, ["asec", "puf_tax_detail"] * 2], + ids=["no_channel", "all_asec", "role_labels"], +) +def test_gate_flags_assembled_frame_missing_clone_indices( + channels: list[str] | None, +) -> None: + # The summary groups copies by source ID and flags the incomplete + # provenance instead of raising, so the gate reports every failure. It + # must never read an assembled channel as a historical role. + frame = _frame( + [10, 10, 20, 20], + channels=channels, + output=[False, True, True, True], + ) + person = frame.table("person").copy() + person["person_spine_source_id"] = [10, 10, 20, 20] + stripped = _replace_person(frame, person) + + summary = us_sipp_head_start_summary(stripped) + assert summary["support_channel_invalid"] is True + assert summary["clone_group_count"] == 2 + assert summary["clone_mismatch_count"] == 1 + assert "channel_eligible_weighted_take_up_shares" not in summary + gate = us_sipp_head_start_signal_gate(stripped) + assert not gate.passed + assert f"{_OUTPUT}: support-channel provenance is invalid" in gate.failures + + +def test_gate_flags_clone_index_past_int64() -> None: + # float(2**63) is the first float past int64; the base let it wrap to + # INT64_MAX. The gate now flags it as invalid provenance. + frame = _historical_tail_frame() + person = frame.table("person").copy() + column = "person_support_clone_index" + tail = person[column].eq(2) + person[column] = person[column].astype(np.float64) + person.loc[tail, column] = float(2**63) + person[_OUTPUT] = [True, False, False, True, False, False, True] + + summary = us_sipp_head_start_summary(_replace_person(frame, person)) + assert summary["support_channel_invalid"] is True + gate = us_sipp_head_start_signal_gate(_replace_person(frame, person)) + assert f"{_OUTPUT}: support-channel provenance is invalid" in gate.failures + + def test_wrapper_heals_stale_output_and_is_exactly_idempotent( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/packages/microcosm-build/tests/test_us_spine_assembly.py b/packages/microcosm-build/tests/test_us_spine_assembly.py index 947e91af71..3eeeb8d30a 100644 --- a/packages/microcosm-build/tests/test_us_spine_assembly.py +++ b/packages/microcosm-build/tests/test_us_spine_assembly.py @@ -1,5 +1,7 @@ from __future__ import annotations +import warnings +from collections.abc import Callable from itertools import permutations import numpy as np @@ -13,7 +15,9 @@ ) from microcosm.build.us_runtime.spine_assembly import assemble_spines from microcosm.build.us_runtime.support_provenance import ( + PUF_TAX_DETAIL_CLONE_INDEX, SPINE_ASSEMBLY_MANIFEST_KEY, + spine_provenance_counts, spine_source_id_column, support_channel_column, support_clone_index_column, @@ -265,6 +269,84 @@ def test_assembly_manifest_detects_cross_grain_channel_disagreement() -> None: ) +def _cloned_assembly() -> Frame: + return clone_us_frame_for_puf_support( + assemble_spines( + {"asec": _asec_frame(), "acs": _acs_frame()}, + household_mass_shares={"asec": 0.5, "acs": 0.5}, + ) + ) + + +_ASSEMBLY_PROVENANCE_READERS = pytest.mark.parametrize( + "reader", + [validate_assembly_provenance, spine_provenance_counts], + ids=["validate", "counts"], +) + + +@_ASSEMBLY_PROVENANCE_READERS +@pytest.mark.parametrize( + "bad_index", + [np.inf, -np.inf, np.nan, 1.5, -1.0, 2.0**63], + ids=[ + "inf", + "negative_inf", + "nan", + "non_integer", + "negative_float", + "float_past_int64", + ], +) +def test_assembly_provenance_rejects_malformed_clone_indices_before_casting( + reader: Callable[..., object], + bad_index: float, +) -> None: + # Corrupt a PUF-detail copy rather than a native row, so the frozen + # native-row counts still match: only clone-index validation can refuse. + # Before #992's gate fix, +inf here passed validation and the receipt + # counter then died in an opaque pandas cast. + cloned = _cloned_assembly() + person = cloned.table("person") + column = support_clone_index_column("person") + copy_row = person.index[person[column].eq(PUF_TAX_DETAIL_CLONE_INDEX)][0] + person[column] = person[column].astype(np.float64) + person.loc[copy_row, column] = bad_index + + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + with pytest.raises( + ValueError, + match=( + r"test assembly output: assembly manifest provenance column " + r"'person_support_clone_index' must contain nonnegative integers " + r"\(finite and representable as int64\)" + ), + ): + reader(cloned, boundary="test assembly output") + + +def test_assembly_provenance_readers_accept_integral_float_clone_indices() -> None: + cloned = _cloned_assembly() + baseline = spine_provenance_counts(cloned, boundary="test assembly output") + person_counts = baseline["person"]["by_clone_index"] + assert person_counts == {"0": 3, "1": 3} + + for entity in cloned.entities: + table = cloned.table(entity) + column = support_clone_index_column(entity) + table[column] = table[column].astype(np.float64) + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + assert validate_assembly_provenance( + cloned, + boundary="test assembly output", + ) + assert ( + spine_provenance_counts(cloned, boundary="test assembly output") == baseline + ) + + @pytest.mark.parametrize( ("spines", "shares", "match"), [ 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 e4718596eb..404450f1aa 100644 --- a/packages/microcosm-build/tests/test_us_spm_universe_source.py +++ b/packages/microcosm-build/tests/test_us_spm_universe_source.py @@ -637,7 +637,7 @@ def test_a_missing_clone_index_cannot_collapse_a_cross_copy_collision(): _classify([dict(row, clone=None) for row in rows]) -@pytest.mark.parametrize("bad", [-1.0, 0.5, float("inf")]) +@pytest.mark.parametrize("bad", [-1.0, 0.5, float("inf"), float(2**63), 2**63]) def test_an_invalid_clone_index_is_refused(bad): with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): _classify( @@ -654,6 +654,53 @@ def test_an_invalid_clone_index_is_refused(bad): ) +@pytest.mark.parametrize( + "missing_column", + ["person_support_clone_index", "person_support_channel"], +) +def test_assembled_missing_support_provenance_is_refused(missing_column): + household, person, unit_ids = _tables( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}], + person_extra_columns={ + "person_source_id": [10], + "person_spine_source_id": [10], + "person_support_channel": ["asec"], + }, + ) + person = person.drop(columns=[missing_column]) + + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): + classify_spm_universe(household=household, person=person, unit_ids=unit_ids) + + +def test_large_assembled_clone_indices_keep_distinct_native_partitions(): + statuses = _classify( + [ + { + "hid": 1, + "unit": 1, + "chan": _ASEC_CHANNEL, + "spm_id": 11, + "clone": 2**62, + }, + { + "hid": 2, + "unit": 2, + "chan": _ASEC_CHANNEL, + "spm_id": 11, + "clone": 2**62 + 1, + }, + ], + person_extra_columns={ + "person_source_id": [10, 10], + "person_spine_source_id": [10, 10], + "person_support_channel": ["asec", "asec"], + }, + ) + + assert statuses == {1: INCLUDED, 2: INCLUDED} + + @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. diff --git a/packages/microcosm-build/tests/test_us_ssi_disability_criteria.py b/packages/microcosm-build/tests/test_us_ssi_disability_criteria.py index 349c4275a7..c535ec11d5 100644 --- a/packages/microcosm-build/tests/test_us_ssi_disability_criteria.py +++ b/packages/microcosm-build/tests/test_us_ssi_disability_criteria.py @@ -40,6 +40,18 @@ from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights _OUTPUT = US_SSI_DISABILITY_CRITERIA_OUTPUT_COLUMNS[0] + + +def _load_tail_fixtures(): + path = Path(__file__).with_name("us_tail_clone_fixtures.py") + spec = importlib.util.spec_from_file_location("us_tail_clone_fixtures", path) + fixtures = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(fixtures) + return fixtures + + +_TAIL = _load_tail_fixtures() _policyengine_us_installed = importlib.util.find_spec("policyengine_us") is not None requires_us = pytest.mark.skipif( not _policyengine_us_installed, @@ -630,6 +642,167 @@ def test_stacked_clone_divergence_diagnostic_checks_clone_two() -> None: assert summary["clone_divergence_source_people"] == 1 +def test_historical_tail_copy_predicts_in_the_puf_role_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(module, "QRF", _FakeQRF) + tailed = _TAIL.with_capital_gains_tail_copies( + clone_us_frame_for_puf_support(_frame()), [1, 3] + ) + person = tailed.table("person") + assert "person_spine_source_id" not in person + tail = person["person_support_clone_index"].eq(2) + assert int(tail.sum()) == 2 + assert set(person.loc[tail, "person_support_channel"]) == {"puf_tax_detail"} + # Give source person 3's PUF-role copies, the tail copy included, a + # positive model draw and disability signal. + puf = person["person_support_channel"].astype(str).eq("puf_tax_detail") + source_three = person["person_source_id"].eq(3) + person.loc[puf & source_three, "bank_account_assets"] = 100.0 + person.loc[puf & source_three, "PEDISDRS"] = 1 + + result = impute_us_ssi_disability_criteria(tailed, _donor(), seed=7) + assert [len(receiver) for receiver in _FakeQRF.predict_receivers] == [20, 22] + assert _FakeQRF.predict_start_offsets == [0, 0] + rows = pd.DataFrame( + { + "source": person["person_source_id"].to_numpy(), + "clone": person["person_support_clone_index"].to_numpy(), + "value": result.to_numpy(), + } + ) + reporter = rows[rows["source"] == 1].set_index("clone")["value"] + source_three_values = rows[rows["source"] == 3].set_index("clone")["value"] + # The ASEC reporter anchor stays on the native copy; the tail copy is + # treated exactly like its primary PUF-detail twin. + assert reporter.to_dict() == {0: True, 1: False, 2: False} + assert source_three_values.to_dict() == {0: False, 1: True, 2: True} + + materialized = _replace_person(tailed, **{_OUTPUT: result.to_numpy()}) + summary = us_ssi_disability_criteria_summary(materialized) + # Sources 1 and 3 diverge between their native and PUF-role copies; the + # diagnostic groups every copy, the tail copy included, by source person. + assert summary["clone_divergence_source_people"] == 2 + + +def test_historical_tail_copy_divergence_joins_its_source_group() -> None: + tailed = _TAIL.with_capital_gains_tail_copies( + clone_us_frame_for_puf_support(_frame(3)), [2] + ) + person = tailed.table("person") + values = np.zeros(len(person), dtype=bool) + values[ + np.flatnonzero( + person["person_support_clone_index"].eq(2) + & person["person_source_id"].eq(2) + ) + ] = True + + summary = us_ssi_disability_criteria_summary( + _replace_person(tailed, **{_OUTPUT: values}) + ) + + assert summary["clone_divergence_source_people"] == 1 + + +@pytest.mark.parametrize( + "columns", + [ + pytest.param({"person_support_channel": ["asec"] * 4}, id="all_asec"), + pytest.param( + {"person_support_channel": ["asec", "puf_tax_detail"] * 2}, + id="role_labels", + ), + pytest.param({}, id="no_channel_no_clone"), + pytest.param( + {"person_support_clone_index": [0, 1, 0, 1]}, + id="clone_index_without_channel", + ), + ], +) +def test_assembled_frame_missing_support_provenance_is_flagged( + columns: dict[str, list[object]], +) -> None: + # Microcosm #992 gate finding: with raw spine IDs present but the + # clone-index column gone, the divergence diagnostic paired copies by + # (source, role) occurrence and reported 0 diverging source people where + # the base, grouping by source ID, reported 1; the gate did not notice, + # because that count is not gated. The role reader now refuses an + # assembled table missing either provenance column, so the summary flags + # missing provenance, exactly like invalid role metadata: its channels are + # never read as roles, its copies are never paired by occurrence, and the + # gate fails. A table with neither column was already flagged. + frame = _replace_person( + _frame(4), + **{ + "person_source_id": np.asarray([10, 10, 20, 20]), + "person_spine_source_id": np.asarray([10, 10, 20, 20]), + _OUTPUT: np.asarray([False, True, True, True]), + **{column: np.asarray(values) for column, values in columns.items()}, + }, + ) + + summary = us_ssi_disability_criteria_summary(frame) + assert summary["support_provenance_missing"] is True + assert summary["channels"] == {} + gate = us_ssi_disability_criteria_signal_gate(frame) + assert not gate.passed + assert ( + "SSI disability support rows lack complete clone-role or " + "person_source_id provenance." in gate.failures + ) + + +def test_divergence_compares_repeated_historical_clone_index() -> None: + # A malformed historical table whose tail copy repeats clone index 1: + # grouping by source ID still compares that copy (the old role-occurrence + # pairing left it unpaired and hid the divergence). + tailed = _TAIL.with_capital_gains_tail_copies( + clone_us_frame_for_puf_support(_frame(3)), [2] + ) + person = tailed.table("person") + tail = person["person_support_clone_index"].eq(2).to_numpy() + clone_index = person["person_support_clone_index"].to_numpy().copy() + clone_index[tail] = 1 + values = np.zeros(len(person), dtype=bool) + values[tail] = True + + summary = us_ssi_disability_criteria_summary( + _replace_person( + tailed, + **{"person_support_clone_index": clone_index, _OUTPUT: values}, + ) + ) + + assert summary["support_provenance_missing"] is False + assert summary["clone_divergence_source_people"] == 1 + + +def test_clone_index_past_int64_is_flagged() -> None: + # float(2**63) is the first float past int64; the base let it wrap to + # INT64_MAX and read the row as a PUF-role copy. + tailed = _TAIL.with_capital_gains_tail_copies( + clone_us_frame_for_puf_support(_frame(3)), [2] + ) + person = tailed.table("person") + clone_index = person["person_support_clone_index"].to_numpy(dtype=np.float64) + clone_index = clone_index.copy() + clone_index[clone_index == 2.0] = float(2**63) + + summary = us_ssi_disability_criteria_summary( + _replace_person( + tailed, + **{ + "person_support_clone_index": clone_index, + _OUTPUT: np.zeros(len(person), dtype=bool), + }, + ) + ) + + assert summary["support_provenance_missing"] is True + assert summary["channels"] == {} + + def test_summary_checks_harmonized_ssi_on_native_role() -> None: expanded = clone_us_frame_for_puf_support(_frame()) person = expanded.table("person") @@ -638,9 +811,7 @@ def test_summary_checks_harmonized_ssi_on_native_role() -> None: source_two = person["person_source_id"].eq(2) person.loc[native & source_two, "SSI_VAL"] = np.nan person.loc[native & source_two, "ssi_reported"] = 900.0 - preserved_existing_anchor = ( - native & person["person_source_id"].eq(1) - ).to_numpy() + preserved_existing_anchor = (native & person["person_source_id"].eq(1)).to_numpy() invalid = _replace_person( expanded, **{_OUTPUT: preserved_existing_anchor}, diff --git a/packages/microcosm-build/tests/test_us_voluntary_filing.py b/packages/microcosm-build/tests/test_us_voluntary_filing.py index ab33414871..cdd58a6ab0 100644 --- a/packages/microcosm-build/tests/test_us_voluntary_filing.py +++ b/packages/microcosm-build/tests/test_us_voluntary_filing.py @@ -42,6 +42,18 @@ from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights _OUTPUT = US_VOLUNTARY_FILING_OUTPUT_COLUMNS[0] + + +def _load_tail_fixtures(): + path = Path(__file__).with_name("us_tail_clone_fixtures.py") + spec = importlib.util.spec_from_file_location("us_tail_clone_fixtures", path) + fixtures = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(fixtures) + return fixtures + + +_TAIL = _load_tail_fixtures() _policyengine_us_installed = importlib.util.find_spec("policyengine_us") is not None requires_us = pytest.mark.skipif( not _policyengine_us_installed, @@ -648,10 +660,526 @@ def test_duplicate_same_role_source_rows_fail_closed() -> None: asec_rows[0], "tax_unit_source_id" ] - with pytest.raises(ValueError, match="duplicated same-role rows"): + with pytest.raises(ValueError, match="duplicated support copies"): impute_us_voluntary_filing(expanded, _donor(), seed=17) +class _IncomeThresholdQRF: + """Predict filing iff tax-unit wages reach 10,000; records the receiver.""" + + receivers: list[pd.DataFrame] = [] + + def __init__(self, **_kwargs: object) -> None: + pass + + def fit(self, *_args: object, **_kwargs: object) -> _IncomeThresholdQRF: + return self + + def predict(self, receiver: pd.DataFrame) -> pd.DataFrame: + self.receivers.append(receiver.copy()) + return pd.DataFrame( + {_OUTPUT: receiver["employment_income"].ge(10_000.0).to_numpy()}, + index=receiver.index, + ) + + +def _historical_tail_frame() -> Frame: + """A non-assembled PUF-support frame with tail copies of units 3 and 6.""" + + return _TAIL.with_capital_gains_tail_copies( + clone_us_frame_for_puf_support(_frame(6)), [3, 6] + ) + + +def _divergent_tail_wages(frame: Frame) -> Frame: + """Zero the tail copies' wages so a tail-row prediction would flip.""" + + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + person = tables["person"] + tail = person["person_support_clone_index"].eq(2) + person.loc[tail, "employment_income_before_lsr"] = 0.0 + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + ) + + +def test_historical_tail_copy_fans_the_source_decision_to_every_copy( + monkeypatch: pytest.MonkeyPatch, +) -> None: + frame = _historical_tail_frame() + tax_unit = frame.table("tax_unit") + assert "tax_unit_spine_source_id" not in tax_unit + assert tax_unit["tax_unit_support_clone_index"].value_counts().to_dict() == { + 0: 6, + 1: 6, + 2: 2, + } + assert set( + tax_unit.loc[ + tax_unit["tax_unit_support_clone_index"].eq(2), + "tax_unit_support_channel", + ] + ) == {"puf_tax_detail"} + + _IncomeThresholdQRF.receivers.clear() + monkeypatch.setattr(module, "QRF", _IncomeThresholdQRF) + predicted = impute_us_voluntary_filing( + _divergent_tail_wages(frame), _donor(), seed=17 + ) + (receiver,) = _IncomeThresholdQRF.receivers + assert len(receiver) == 6 # one canonical row per source tax unit + by_source = pd.DataFrame( + {"source": tax_unit["tax_unit_source_id"], "value": predicted} + ).groupby("source")["value"] + assert (by_source.nunique() == 1).all() + # Unit wages are 3,000 x household id (+2,000 per spouse): units 3 to 6 + # file. The zero-wage tail copies inherit the native copy's decision. + assert by_source.first().to_dict() == { + 101: False, + 102: False, + 103: True, + 104: True, + 105: True, + 106: True, + } + tail = tax_unit["tax_unit_support_clone_index"].eq(2).to_numpy() + assert predicted[tail].all() + + materialized = _replace_tax_unit(frame, **{_OUTPUT: predicted.to_numpy()}) + summary = us_voluntary_filing_summary(materialized) + assert summary["clone_source_units"] == 6 + assert summary["clone_mismatch_source_units"] == 0 + + flipped = predicted.to_numpy().copy() + flipped[np.flatnonzero(tail)[0]] = False + mismatch = _replace_tax_unit(frame, **{_OUTPUT: flipped}) + mismatch_summary = us_voluntary_filing_summary(mismatch) + assert mismatch_summary["clone_mismatch_source_units"] == 1 + gate = us_voluntary_filing_signal_gate(mismatch) + assert any("disagree for 1 source unit" in failure for failure in gate.failures) + + +def test_historical_puf_only_survivor_predicts_from_the_primary_detail_copy( + monkeypatch: pytest.MonkeyPatch, +) -> None: + frame = _divergent_tail_wages(_historical_tail_frame()) + tax_unit = frame.table("tax_unit") + person = frame.table("person") + native_unit_six = tax_unit.loc[ + tax_unit["tax_unit_source_id"].eq(106) + & tax_unit["tax_unit_support_clone_index"].eq(0), + "tax_unit_id", + ] + survivors = frame.select( + ~person["person_tax_unit_id"].isin(native_unit_six).to_numpy() + ) + + _IncomeThresholdQRF.receivers.clear() + monkeypatch.setattr(module, "QRF", _IncomeThresholdQRF) + predicted = impute_us_voluntary_filing(survivors, _donor(), seed=17) + surviving_units = survivors.table("tax_unit") + source_six = surviving_units["tax_unit_source_id"].eq(106).to_numpy() + surviving_clones = surviving_units.loc[source_six, "tax_unit_support_clone_index"] + assert surviving_clones.tolist() == [1, 2] + # Copy rank picks the primary PUF-detail copy (wages 18,000), never the + # zero-wage tail copy. + assert predicted[source_six].tolist() == [True, True] + + +def test_historical_duplicate_clone_index_still_fails_closed() -> None: + frame = _historical_tail_frame() + tax_unit = frame.table("tax_unit") + tail = tax_unit["tax_unit_support_clone_index"].eq(2).to_numpy() + clone_index = tax_unit["tax_unit_support_clone_index"].to_numpy().copy() + clone_index[tail] = 1 + duplicated = _replace_tax_unit( + frame, **{"tax_unit_support_clone_index": clone_index} + ) + + with pytest.raises(ValueError, match=r"duplicated support copies.*\('103', 1\)"): + impute_us_voluntary_filing(duplicated, _donor(), seed=17) + + +@pytest.mark.parametrize("clone_index", [3, 7, 2**62]) +@pytest.mark.parametrize("dtype", [np.int64, np.float64], ids=["int", "float"]) +def test_historical_out_of_domain_clone_refused_before_prediction( + clone_index: int, + dtype, + monkeypatch: pytest.MonkeyPatch, +) -> None: + frame = _historical_tail_frame() + tax_unit = frame.table("tax_unit") + clone_indices = tax_unit["tax_unit_support_clone_index"].to_numpy( + dtype=dtype, copy=True + ) + clone_indices[clone_indices == 2] = clone_index + frame = _replace_tax_unit(frame, tax_unit_support_clone_index=clone_indices) + receiver = pd.DataFrame( + { + predictor: np.zeros(len(tax_unit)) + for predictor in SIPP_VOLUNTARY_FILING_MODEL_PREDICTORS + }, + index=tax_unit["tax_unit_id"], + ) + + def unexpected_qrf(**kwargs): + pytest.fail("Invalid historical clone indices must fail before QRF") + + monkeypatch.setattr(module, "QRF", unexpected_qrf) + with pytest.raises(ValueError, match="historical"): + module._source_receiver_rows(frame, receiver) + with pytest.raises(ValueError, match="historical"): + impute_us_voluntary_filing(frame, _donor(), seed=17) + + +def test_assembled_frame_without_clone_indices_fails_closed() -> None: + # Microcosm #992 gate finding: with the clone-index column dropped, an + # assembled ASEC-channel frame once fell back to role ranks. Assembled + # channels name physical sources, so the copy rank needs clone indices. + tax_unit = pd.DataFrame( + { + "tax_unit_id": [100, 200], + "tax_unit_source_id": [10, 20], + "tax_unit_spine_source_id": [1, 2], + "tax_unit_support_channel": ["asec", "asec"], + } + ) + receiver = pd.DataFrame( + { + predictor: np.arange(2, dtype=np.float64) + for predictor in SIPP_VOLUNTARY_FILING_MODEL_PREDICTORS + }, + index=tax_unit["tax_unit_id"], + ) + + class TaxUnitFrame: + def table(self, entity: str) -> pd.DataFrame: + assert entity == "tax_unit" + return tax_unit + + with pytest.raises( + ValueError, + match=r"assembled support metadata requires 'tax_unit_support_clone_index'", + ): + module._source_receiver_rows(TaxUnitFrame(), receiver) + + +def _assembled_gate_frame( + output: list[bool], + *, + channels: list[str] | None, + clone_indices: list[int] | None = None, +) -> Frame: + """Four tax units over source IDs [10, 10, 20, 20] with raw spine IDs.""" + + columns: dict[str, np.ndarray] = { + _OUTPUT: np.asarray(output), + "tax_unit_source_id": np.asarray([10, 10, 20, 20]), + "tax_unit_spine_source_id": np.asarray([10, 10, 20, 20]), + } + if channels is not None: + columns["tax_unit_support_channel"] = np.asarray(channels, dtype=object) + if clone_indices is not None: + columns["tax_unit_support_clone_index"] = np.asarray(clone_indices) + return _replace_tax_unit(_frame(4), **columns) + + +@pytest.mark.parametrize( + "channels", + [["asec"] * 4, ["asec", "puf_tax_detail"] * 2], + ids=["all_asec", "role_labels"], +) +@pytest.mark.parametrize( + "output", + [[False, True, True, True], [False, False, True, True]], + ids=["copies_disagree", "copies_agree"], +) +def test_gate_refuses_assembled_frame_without_clone_indices( + channels: list[str], + output: list[bool], +) -> None: + # Microcosm #992 gate finding (c), exactly as reported: source IDs + # [10, 10, 20, 20] with matching spine IDs, all 'asec' channels and + # outputs [F, T, T, T]. With the clone-index column gone the summary fell + # back to (source, role) occurrence pairing, reported 0 mismatched source + # units and the gate passed; the base grouped by source ID and failed. + # When the copies agree the fallback still under-reported + # clone_source_units (0 instead of 2). The gate now refuses the table. + frame = _assembled_gate_frame(output, channels=channels) + pattern = ( + r"assembled support metadata requires 'tax_unit_support_clone_index' " + r"alongside 'tax_unit_spine_source_id'" + ) + with pytest.raises(ValueError, match=pattern): + us_voluntary_filing_summary(frame) + with pytest.raises(ValueError, match=pattern): + us_voluntary_filing_signal_gate(frame) + + +@pytest.mark.parametrize( + ("channels", "clone_indices", "missing"), + [ + pytest.param( + None, + None, + r"'tax_unit_support_channel' and 'tax_unit_support_clone_index'", + id="no_channel_no_clone", + ), + pytest.param( + None, + [0, 1, 0, 1], + r"'tax_unit_support_channel'", + id="clone_index_without_channel", + ), + ], +) +def test_gate_refuses_assembled_frame_without_support_channel( + channels: list[str] | None, + clone_indices: list[int] | None, + missing: str, +) -> None: + # Before #992 the gate skipped the clone comparison entirely for a table + # with neither provenance column, even with raw spine IDs and copies that + # disagree. An assembled table must carry complete provenance. + frame = _assembled_gate_frame( + [False, True, True, True], + channels=channels, + clone_indices=clone_indices, + ) + with pytest.raises( + ValueError, + match=r"assembled support metadata requires " + missing, + ): + us_voluntary_filing_signal_gate(frame) + + +@pytest.mark.parametrize( + "consumer", [us_voluntary_filing_summary, us_voluntary_filing_signal_gate] +) +@pytest.mark.parametrize( + "missing_columns", + [ + ("tax_unit_support_channel",), + ("tax_unit_support_clone_index",), + ("tax_unit_support_channel", "tax_unit_support_clone_index"), + ], + ids=["no_channel", "no_clone_index", "no_support_metadata"], +) +@pytest.mark.parametrize("empty", [False, True], ids=["missing_output", "empty"]) +def test_gate_validates_assembled_provenance_before_output_or_weights( + consumer, + missing_columns: tuple[str, ...], + empty: bool, +) -> None: + # The earlier fix checked provenance only after reading the output and + # weights. The gate's missing-output return bypassed that check entirely. + # Empty tables cannot form a valid Frame, but direct callers still need + # the same provenance refusal before any output/weight processing. + tax_unit = pd.DataFrame( + { + "tax_unit_id": [100], + "tax_unit_source_id": [10], + "tax_unit_spine_source_id": [1], + "tax_unit_support_channel": ["asec"], + "tax_unit_support_clone_index": [0], + } + ).drop(columns=list(missing_columns)) + if empty: + tax_unit = tax_unit.iloc[:0].assign(**{_OUTPUT: pd.Series(dtype=bool)}) + + class TaxUnitFrame: + def table(self, entity: str) -> pd.DataFrame: + assert entity == "tax_unit" + return tax_unit + + def resolve_weights(self, entity: str): + pytest.fail("Incomplete assembled provenance must precede weights") + + with pytest.raises(ValueError, match="assembled support metadata requires"): + consumer(TaxUnitFrame()) + + +def test_receiver_refuses_assembled_frame_without_support_metadata() -> None: + # Unique source IDs give the one-row-per-source path nothing to reject, + # so the stripped assembled table must be refused by provenance alone. + tax_unit = pd.DataFrame( + { + "tax_unit_id": [100, 200, 300, 400], + "tax_unit_source_id": [10, 20, 30, 40], + "tax_unit_spine_source_id": [1, 2, 3, 4], + } + ) + receiver = pd.DataFrame( + { + predictor: np.arange(4, dtype=np.float64) + for predictor in SIPP_VOLUNTARY_FILING_MODEL_PREDICTORS + }, + index=tax_unit["tax_unit_id"], + ) + + class TaxUnitFrame: + def table(self, entity: str) -> pd.DataFrame: + assert entity == "tax_unit" + return tax_unit + + with pytest.raises( + ValueError, + match=( + r"assembled support metadata requires 'tax_unit_support_channel' " + r"and 'tax_unit_support_clone_index'" + ), + ): + module._source_receiver_rows(TaxUnitFrame(), receiver) + + +def test_gate_compares_every_historical_copy_with_repeated_clone_indices() -> None: + # A historical table whose tail copy repeats clone index 1 is malformed + # (both imputers refuse it). The gate still compares every copy of the + # source unit, so the duplicated copy's disagreement is caught; the old + # (source, role) occurrence pairing left it unpaired and unchecked. + frame = _historical_tail_frame() + tax_unit = frame.table("tax_unit") + clone_index = tax_unit["tax_unit_support_clone_index"].to_numpy().copy() + tail = clone_index == 2 + clone_index[tail] = 1 + output = tax_unit["tax_unit_source_id"].isin([103, 104, 105, 106]).to_numpy() + agreeing = _replace_tax_unit( + frame, + **{"tax_unit_support_clone_index": clone_index, _OUTPUT: output}, + ) + summary = us_voluntary_filing_summary(agreeing) + assert summary["clone_source_units"] == 6 + assert summary["clone_mismatch_source_units"] == 0 + + disagreeing_output = output.copy() + disagreeing_output[np.flatnonzero(tail)[0]] = False + disagreeing = _replace_tax_unit(agreeing, **{_OUTPUT: disagreeing_output}) + assert us_voluntary_filing_summary(disagreeing)["clone_mismatch_source_units"] == 1 + gate = us_voluntary_filing_signal_gate(disagreeing) + assert any("disagree for 1 source unit" in failure for failure in gate.failures) + + +@pytest.mark.parametrize( + "output", + [[False, False, True, True], [False, True, True, True]], + ids=["copies_agree", "copies_disagree"], +) +def test_gate_flags_duplicate_sources_without_support_metadata( + output: list[bool], +) -> None: + frame = _replace_tax_unit( + _frame(4), + tax_unit_source_id=np.asarray([10, 10, 20, 20]), + **{_OUTPUT: np.asarray(output)}, + ) + summary = us_voluntary_filing_summary(frame) + assert summary["clone_metadata_missing"] is True + gate = us_voluntary_filing_signal_gate(frame) + assert not gate.passed + assert any("provenance" in failure for failure in gate.failures) + + +@pytest.mark.parametrize( + "consumer", [us_voluntary_filing_summary, us_voluntary_filing_signal_gate] +) +@pytest.mark.parametrize( + "output", + [ + [False, False, False, True, True], + [False, False, True, True, True], + ], + ids=["copies_agree", "unpaired_copy_disagrees"], +) +def test_channel_only_gate_refuses_duplicated_source_role( + consumer, + output: list[bool], +) -> None: + frame = _replace_tax_unit( + _frame(5), + tax_unit_source_id=np.asarray([10, 10, 10, 20, 20]), + tax_unit_support_channel=np.asarray( + ["asec", "puf_tax_detail", "puf_tax_detail", "asec", "puf_tax_detail"] + ), + **{_OUTPUT: np.asarray(output)}, + ) + with pytest.raises(ValueError, match="duplicated same-role rows"): + consumer(frame) + + +def test_unique_sources_without_metadata_keep_the_unexpanded_summary() -> None: + frame = _replace_tax_unit( + _frame(4), **{_OUTPUT: np.asarray([False, True, True, True])} + ) + source_frame = _replace_tax_unit( + frame, tax_unit_source_id=np.asarray([10, 20, 30, 40]) + ) + assert us_voluntary_filing_summary(source_frame) == us_voluntary_filing_summary( + frame + ) + assert us_voluntary_filing_signal_gate(source_frame).passed + + +def test_channel_only_unique_source_role_pairs_keep_valid_clone_diagnostics() -> None: + frame = _replace_tax_unit( + _frame(4), + tax_unit_source_id=np.asarray([10, 10, 20, 20]), + tax_unit_support_channel=np.asarray(["asec", "puf_tax_detail"] * 2), + **{_OUTPUT: np.asarray([False, False, True, True])}, + ) + summary = us_voluntary_filing_summary(frame) + assert summary["clone_source_units"] == 2 + assert summary["clone_mismatch_source_units"] == 0 + assert summary["clone_metadata_missing"] is False + assert us_voluntary_filing_signal_gate(frame).passed + + +@pytest.mark.parametrize("assembled", [True, False], ids=["assembled", "historical"]) +def test_clone_index_past_int64_fails_closed(assembled: bool) -> None: + # float(2**63) is the first float past int64; the base let it wrap to + # INT64_MAX and both the receiver and the gate ran normally. + malformed = ( + r"PUF support metadata column 'tax_unit_support_clone_index' must " + r"contain nonnegative integers \(finite and representable as int64\)" + ) + if assembled: + frame = _assembled_gate_frame( + [False, False, True, True], + channels=["acs"] * 4, + clone_indices=[0, 1, 0, 1], + ) + frame = _replace_tax_unit( + frame, + tax_unit_support_clone_index=np.asarray([0.0, 1.0, 0.0, float(2**63)]), + ) + else: + frame = _historical_tail_frame() + clone_index = frame.table("tax_unit")["tax_unit_support_clone_index"] + values = clone_index.to_numpy(dtype=np.float64).copy() + values[clone_index.eq(2).to_numpy()] = float(2**63) + frame = _replace_tax_unit( + frame, + **{ + "tax_unit_support_clone_index": values, + _OUTPUT: np.ones(len(values), dtype=bool), + }, + ) + with pytest.raises(ValueError, match=malformed): + us_voluntary_filing_signal_gate(frame) + receiver = pd.DataFrame( + { + predictor: np.zeros(len(frame.table("tax_unit"))) + for predictor in SIPP_VOLUNTARY_FILING_MODEL_PREDICTORS + }, + index=frame.table("tax_unit")["tax_unit_id"], + ) + with pytest.raises(ValueError, match=malformed): + module._source_receiver_rows(frame, receiver) + + def test_assembled_clone_two_uses_explicit_index_and_checks_every_clone() -> None: tax_unit = pd.DataFrame( { @@ -665,9 +1193,7 @@ def test_assembled_clone_two_uses_explicit_index_and_checks_every_clone() -> Non receiver = pd.DataFrame( { predictor: np.arange(5, dtype=np.float64) + offset - for offset, predictor in enumerate( - SIPP_VOLUNTARY_FILING_MODEL_PREDICTORS - ) + for offset, predictor in enumerate(SIPP_VOLUNTARY_FILING_MODEL_PREDICTORS) }, index=tax_unit["tax_unit_id"], ) @@ -677,9 +1203,7 @@ def table(self, entity: str) -> pd.DataFrame: assert entity == "tax_unit" return tax_unit - prediction_rows, fan_keys = module._source_receiver_rows( - TaxUnitFrame(), receiver - ) + prediction_rows, fan_keys = module._source_receiver_rows(TaxUnitFrame(), receiver) assert prediction_rows.index.tolist() == ["10", "20"] assert fan_keys.tolist() == ["10", "10", "10", "20", "20"] # Source 10 prefers clone 0; source 20 has no native survivor and picks diff --git a/packages/microcosm-build/tests/test_us_weeks_unemployed.py b/packages/microcosm-build/tests/test_us_weeks_unemployed.py index 479b114906..ab760d32af 100644 --- a/packages/microcosm-build/tests/test_us_weeks_unemployed.py +++ b/packages/microcosm-build/tests/test_us_weeks_unemployed.py @@ -266,6 +266,10 @@ def _stacked_gate_frame() -> Frame: np.arange(10_000, 10_000 + acs_native_rows, dtype=np.int64), ] ) + # Spine assembly writes the assembly-unique source ID beside the raw one + # (the ACS offset above keeps it unique); a support clone keeps its + # source's ID. + person["person_source_id"] = person["person_spine_source_id"] weeks = np.zeros(len(person), dtype=np.float64) weeks[:18] = 17.0 weeks[asec_native_rows : asec_native_rows + 12] = 17.0 diff --git a/packages/microcosm-build/tests/us_tail_clone_fixtures.py b/packages/microcosm-build/tests/us_tail_clone_fixtures.py new file mode 100644 index 0000000000..f5deb80101 --- /dev/null +++ b/packages/microcosm-build/tests/us_tail_clone_fixtures.py @@ -0,0 +1,100 @@ +"""Build capital-gains own-tail copies with the production clone operator. + +A PUF-support base built without a selection source keeps every support copy: +the native ASEC record (clone index 0), its primary PUF-detail copy (clone +index 1) and, for the households the capital-gains tail transfer selects, an +own-tail copy of the PUF-detail household (clone index 2). The tail copy keeps +its source IDs and the ``puf_tax_detail`` channel, so a historical +(non-assembled) frame carries two PUF-role rows per selected source unit. + +Stage tests use :func:`with_capital_gains_tail_copies` to reproduce that layout +through ``puf_capital_gains_tail._clone_and_transfer`` itself rather than a +hand-written approximation of it. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import numpy as np +import pandas as pd + +import microcosm.build.us_runtime.puf_capital_gains_tail as tail_module +from microcosm.build.us_runtime.puf_capital_gains_tail import ( + PUF_CAPITAL_GAINS_TAIL_PERSON_COLUMNS, + PUF_CAPITAL_GAINS_TAIL_TAX_UNIT_COLUMNS, +) +from microcosm.frame import Frame + +TAIL_CLONE_INDEX = 2 + + +def _with_zero_capital_gains_columns(frame: Frame) -> Frame: + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + for column in PUF_CAPITAL_GAINS_TAIL_PERSON_COLUMNS: + if column not in tables["person"]: + tables["person"][column] = 0.0 + for column in PUF_CAPITAL_GAINS_TAIL_TAX_UNIT_COLUMNS: + if column not in tables["tax_unit"]: + tables["tax_unit"][column] = 0.0 + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + metadata=frame.metadata, + ) + + +def with_capital_gains_tail_copies( + frame: Frame, + source_household_ids: Sequence[int], + *, + weight_share: float = 0.5, +) -> Frame: + """Add one own-tail copy per selected source household. + + ``frame`` must be a PUF-support clone (clone indices 0 and 1). Each selected + source household's primary PUF-detail copy is split: ``weight_share`` of + its household weight moves to a new clone-index-2 copy whose first + tax-unit member receives a nonzero capital-gains vector, exactly as the + production transfer writes it. + """ + + frame = _with_zero_capital_gains_columns(frame) + household = frame.table("household") + clone_index = pd.to_numeric( + household["household_support_clone_index"], errors="raise" + ).to_numpy(dtype=np.int64) + source = household["household_source_id"].to_numpy(dtype=np.int64) + weights = frame.weights_for("household").values + person = frame.table("person") + rows: list[dict[str, object]] = [] + for position, source_household_id in enumerate(source_household_ids): + match = np.flatnonzero((clone_index == 1) & (source == source_household_id)) + if len(match) != 1: + raise ValueError( + "Tail fixture needs exactly one primary PUF-detail copy of " + f"source household {source_household_id}." + ) + row = int(match[0]) + household_id = int(household["household_id"].iloc[row]) + members = person.loc[person["person_household_id"].eq(household_id)] + tax_unit_id = int(members["person_tax_unit_id"].iloc[0]) + assignment: dict[str, object] = { + "recipient_household_id": household_id, + "recipient_tax_unit_id": tax_unit_id, + "assigned_weight": float(weights[row]) * float(weight_share), + "donor_source_id": 900_000 + position, + tail_module._TAIL_SYNTHETIC_COLUMN: False, + "filing_status_code": 1, + tail_module._TAIL_AGI_BAND_INDEX_COLUMN: 0, + } + for column in PUF_CAPITAL_GAINS_TAIL_PERSON_COLUMNS: + assignment[column] = 1_000_000.0 + position + for column in PUF_CAPITAL_GAINS_TAIL_TAX_UNIT_COLUMNS: + assignment[column] = 0.0 + rows.append(assignment) + transferred, _receipt = tail_module._clone_and_transfer(frame, pd.DataFrame(rows)) + return transferred