Ordering train: per-stratum tail support receipts (#652) + late-producer input DAG (#653)

changelog.d/652-capital-gains-tail-thin-strata.fixed.md

@@ -0,0 +1 @@
+Declare capital-gains tail support per filing status: require one eligible unique single-tax-unit PUF-detail recipient household per selected q99.5 donor, skip a thin status as a whole with a named and counted `insufficient_support` receipt, and reject altered receipts at terminal gates. Preserve byte-exact attachment for adequate statuses and unchanged full-scale output. Replace the fixed post-clone source-then-transfer sequence with an import-validated 38-producer/71-edge dependency DAG whose readiness fence derives the byte-stable order, separates missing inputs from invalid finite numerics, rejects named cycles and unfilled inputs, transfers PUF SSTB income before adult care and tuition before education, and makes the ACS earnings-universe producer, all sixteen source producers, and their once-only finalizer explicit. Content-hash every declared input alternative, reconciled readiness row, output, callback receipt, execution row, and entry/output frame; anchor the signed chain in immutable frame metadata and carry its independent authority through version-9 stacked authority, outer stacked materializer v10, version-6 stacked pool checkpoints, and schema-7 pool manifests and consumers while preserving the retiring legacy schema-4/materializer-3 bytes and rejecting stacked-envelope downgrade attempts. Bind the primary donor bytes, routed QRF bank and sidecar, exact optional allocation/passthrough reads, worker controls, once-resolved tail spec/SOI assets and gates, all nineteen transfer predictor/codec/model configurations and bank identities, all sixteen source runtime configurations—including packaged `SourceStageSpec` identities for the fifteen manifest-backed callbacks—finalizer doctrine, and the ACS universe runtime rule/config through exact kind-specific schema-v1/v2/v3 resource evidence in late-registry schema v14/receipt schema v3. Bind every virtual-resource resolution mode into the outer checkpoint identity, prove current checkpoint discovery succeeds and stale resource semantics fail, require adult-care transfer's tax-unit role before dispatch, keep Schedule-D ownership in the whole-pool derive rather than bounded transfer side effects, and reject shallow, forged, stale, identityless, runtime-drifted, or cross-producer absence resources before their callbacks. Scope ACS-only `TYPEHUGQ` readiness and structural output coverage to ACS-origin rows, keep the 1,688 ASEC-origin structural nulls untouched, reject any ACS-side null, and retire whole-pool `RELSHIPP`, `TEN`, and `H_TENURE` transfer fallbacks in favor of canonical dual-origin predictors. Preserve all canonical late-stage boolean outputs as nullable booleans at primary-PUF and source-merge materialization seams, fail closed on numeric or object family drift across every registry-declared late write, and retain boolean dtype through per-stratum tail cloning. Serialize pandas nullable booleans through conditional frame-checkpoint schema v3 as canonical NumPy-bool values plus an explicit lossless null mask only when absences exist, restore the logical nullable family on load, reject malformed or downgraded encodings, and retain byte-identical schema-v2 generic, UK, and legacy artifacts when no nullable boolean is present. Declare the exact final owner for every origin and clone role of qualified tuition, traditional IRA contributions, and self-employed pension contributions in a content-addressed 18-row receipt bound into late-registry schema v15 and the tail manifest; prove education is byte-exact consume-only, mirror the two source-owned ASEC retirement values from clone 1 to clone 2 by assembly-unique source ID, and reject any unregistered recipient-owned or tail-owned dual-write intersection without weakening the terminal preservation guard.

packages/microcosm-build/src/microcosm/build/frame_checkpoint.py

@@ -40,7 +40,11 @@
     "write_frame_checkpoint",
 ]
 
-FRAME_CHECKPOINT_SCHEMA_VERSION = 2
+FRAME_CHECKPOINT_SCHEMA_VERSION = 3
+_LEGACY_FRAME_CHECKPOINT_SCHEMA_VERSION = 2
+_SUPPORTED_FRAME_CHECKPOINT_SCHEMA_VERSIONS = frozenset(
+    {_LEGACY_FRAME_CHECKPOINT_SCHEMA_VERSION, FRAME_CHECKPOINT_SCHEMA_VERSION}
+)
 
 _ARTIFACT_KIND = "populace_frame_checkpoint"
 _ROOT = "_populace_frame_checkpoint"
@@ -50,6 +54,7 @@
 _ENCODING_DATETIME = "datetime64"
 _ENCODING_TIMEDELTA = "timedelta64"
 _ENCODING_OBJECT = "object_scalars_v1"
+_ENCODING_NULLABLE_BOOLEAN = "nullable_boolean_v1"
 
 _TAG_NONE = 0
 _TAG_PD_NA = 1
@@ -339,10 +344,26 @@ def _checkpoint_metadata(
             }
         )
 
+    strata_spec = _series_spec(frame.strata, label="strata")
+    uses_nullable_boolean = any(
+        column.get("encoding") == _ENCODING_NULLABLE_BOOLEAN
+        for table in table_specs
+        for column in table["columns"]
+    ) or any(
+        table["index"].get("encoding") == _ENCODING_NULLABLE_BOOLEAN
+        for table in table_specs
+    )
+    uses_nullable_boolean = uses_nullable_boolean or (
+        strata_spec.get("encoding") == _ENCODING_NULLABLE_BOOLEAN
+    )
     schema = frame.schema
     return {
         "artifact_kind": _ARTIFACT_KIND,
-        "schema_version": FRAME_CHECKPOINT_SCHEMA_VERSION,
+        "schema_version": (
+            FRAME_CHECKPOINT_SCHEMA_VERSION
+            if uses_nullable_boolean
+            else _LEGACY_FRAME_CHECKPOINT_SCHEMA_VERSION
+        ),
         "schema": {
             "person_entity": schema.person_entity,
             "group_entities": list(schema.group_entities),
@@ -368,7 +389,7 @@ def _checkpoint_metadata(
             if "household" in frame.weighted_entities
             else None
         ),
-        "strata": _series_spec(frame.strata, label="strata"),
+        "strata": strata_spec,
         "mass_log": [_mass_change_payload(record) for record in frame.mass_log],
         "external_metadata": external_metadata,
     }
@@ -380,7 +401,7 @@ def _frame_tables(frame: Frame) -> tuple[tuple[str, pd.DataFrame], ...]:
     return (*entities, *links)
 
 
-def _series_spec(series: pd.Series, *, label: str) -> dict[str, str]:
+def _series_spec(series: pd.Series, *, label: str) -> dict[str, object]:
     dtype = series.dtype
     if isinstance(dtype, pd.CategoricalDtype):
         raise TypeError(
@@ -392,6 +413,12 @@ def _series_spec(series: pd.Series, *, label: str) -> dict[str, str]:
             f"Frame checkpoint does not support timezone-aware dtype for {label!r}; "
             "convert it to timezone-naive datetime64 first."
         )
+    if isinstance(dtype, pd.BooleanDtype):
+        return {
+            "dtype": "boolean",
+            "encoding": _ENCODING_NULLABLE_BOOLEAN,
+            "has_null_mask": bool(series.isna().any()),
+        }
     if isinstance(dtype, pd.api.extensions.ExtensionDtype) and not isinstance(
         dtype, pd.StringDtype
     ):
@@ -523,6 +550,17 @@ def _write_series(group: Any, series: pd.Series, spec: Mapping[str, object]) ->
         _write_numpy_dataset(group, "offsets", offsets)
         _write_bytes_dataset(group, "payload", payload)
         return
+    if encoding == _ENCODING_NULLABLE_BOOLEAN:
+        values = series.to_numpy(
+            dtype=np.bool_,
+            na_value=False,
+            copy=False,
+        )
+        _write_numpy_dataset(group, "values", values)
+        if spec.get("has_null_mask") is True:
+            null_mask = series.isna().to_numpy(dtype=np.uint8, copy=False)
+            _write_numpy_dataset(group, "null_mask", null_mask)
+        return
     raise RuntimeError(f"Unknown checkpoint series encoding {encoding!r}.")
 
 
@@ -552,6 +590,57 @@ def _read_series(
         payload = _read_bytes_dataset(group, "payload", path)
         values = _decode_object_values(offsets, payload, path=path, label=label)
         series = pd.Series(values, dtype=restore_dtype, copy=False)
+    elif encoding == _ENCODING_NULLABLE_BOOLEAN:
+        if dtype != "boolean":
+            raise ValueError(
+                f"Frame checkpoint {path} {label!r} nullable boolean encoding "
+                f"has declared dtype {dtype!r}, not 'boolean'."
+            )
+        has_null_mask = spec.get("has_null_mask")
+        if type(has_null_mask) is not bool:
+            raise ValueError(
+                f"Frame checkpoint {path} {label!r} nullable boolean "
+                "has_null_mask must be a boolean."
+            )
+        values = _read_numpy_dataset(group, "values", path)
+        if values.ndim != 1 or values.dtype != np.dtype(np.bool_):
+            raise ValueError(
+                f"Frame checkpoint {path} {label!r} nullable boolean values "
+                "must be a one-dimensional bool array."
+            )
+        if has_null_mask:
+            if "null_mask" not in group:
+                raise ValueError(
+                    f"Frame checkpoint {path} {label!r} is missing its null mask."
+                )
+            null_mask = _read_numpy_dataset(group, "null_mask", path)
+            if (
+                null_mask.ndim != 1
+                or null_mask.dtype != np.dtype(np.uint8)
+                or len(null_mask) != len(values)
+                or ((null_mask != 0) & (null_mask != 1)).any()
+                or not null_mask.any()
+            ):
+                raise ValueError(
+                    f"Frame checkpoint {path} {label!r} null mask must be a "
+                    "one-dimensional uint8 0/1 array aligned to its values."
+                )
+            mask = null_mask.astype(np.bool_, copy=False)
+            if values[mask].any():
+                raise ValueError(
+                    f"Frame checkpoint {path} {label!r} null mask covers "
+                    "noncanonical true storage bits."
+                )
+        else:
+            if "null_mask" in group:
+                raise ValueError(
+                    f"Frame checkpoint {path} {label!r} has an unexpected null mask."
+                )
+            mask = np.zeros(len(values), dtype=np.bool_)
+        series = pd.Series(
+            pd.arrays.BooleanArray(values, mask, copy=False),
+            copy=False,
+        )
     else:
         raise ValueError(
             f"Frame checkpoint {path} has unknown encoding {encoding!r} for {label!r}."
@@ -574,7 +663,7 @@ def _read_series(
 
 def _declared_dtype(
     spec: Mapping[str, Any], *, path: Path, label: str
-) -> str | pd.StringDtype:
+) -> str | pd.BooleanDtype | pd.StringDtype:
     """The concrete dtype a spec restores to, environment-independently.
 
     String dtypes resolve through the recorded storage and NA marker — the
@@ -586,6 +675,8 @@ def _declared_dtype(
     verified in produced, and it is the build's canonical string policy.
     """
     dtype = _require_string(spec, "dtype", label=label)
+    if dtype == "boolean":
+        return pd.BooleanDtype()
     if dtype not in ("str", "string"):
         return dtype
     storage = spec.get("string_storage", "python")
@@ -605,7 +696,12 @@ def _declared_dtype(
     )
 
 
-def _dtype_matches(actual: Any, expected: str | pd.StringDtype) -> bool:
+def _dtype_matches(
+    actual: Any,
+    expected: str | pd.BooleanDtype | pd.StringDtype,
+) -> bool:
+    if isinstance(expected, pd.BooleanDtype):
+        return actual == expected
     if isinstance(expected, pd.StringDtype):
         return actual == expected and getattr(actual, "storage", None) == (
             expected.storage
@@ -737,14 +833,69 @@ def _read_metadata(root: Any, path: Path) -> dict[str, Any]:
             f"{metadata.get('artifact_kind')!r}."
         )
     version = metadata.get("schema_version")
-    if version != FRAME_CHECKPOINT_SCHEMA_VERSION:
+    if version not in _SUPPORTED_FRAME_CHECKPOINT_SCHEMA_VERSIONS:
         raise ValueError(
             f"Frame checkpoint {path} schema version is {version!r}; expected "
-            f"{FRAME_CHECKPOINT_SCHEMA_VERSION}."
+            f"one of {sorted(_SUPPORTED_FRAME_CHECKPOINT_SCHEMA_VERSIONS)}."
+        )
+    nullable_spec_count = 0
+    for label, spec in _checkpoint_series_specs(metadata):
+        dtype = spec.get("dtype")
+        encoding = spec.get("encoding")
+        declares_nullable = dtype == "boolean"
+        uses_nullable_encoding = encoding == _ENCODING_NULLABLE_BOOLEAN
+        if version == _LEGACY_FRAME_CHECKPOINT_SCHEMA_VERSION and (
+            declares_nullable or uses_nullable_encoding
+        ):
+            raise ValueError(
+                f"Frame checkpoint {path} schema version 2 cannot carry nullable "
+                f"boolean spec {label!r}."
+            )
+        if version == FRAME_CHECKPOINT_SCHEMA_VERSION and (
+            declares_nullable != uses_nullable_encoding
+        ):
+            raise ValueError(
+                f"Frame checkpoint {path} schema version 3 nullable boolean spec "
+                f"{label!r} must pair declared dtype 'boolean' with encoding "
+                f"{_ENCODING_NULLABLE_BOOLEAN!r}."
+            )
+        if uses_nullable_encoding:
+            nullable_spec_count += 1
+            if type(spec.get("has_null_mask")) is not bool:
+                raise ValueError(
+                    f"Frame checkpoint {path} {label!r} nullable boolean "
+                    "has_null_mask must be a boolean."
+                )
+    if version == FRAME_CHECKPOINT_SCHEMA_VERSION and nullable_spec_count == 0:
+        raise ValueError(
+            f"Frame checkpoint {path} schema version 3 requires at least one "
+            "nullable boolean spec."
         )
     return metadata
 
 
+def _checkpoint_series_specs(
+    metadata: Mapping[str, Any],
+) -> tuple[tuple[str, dict[str, Any]], ...]:
+    """Return every serialized series spec for version-policy validation."""
+
+    specs: list[tuple[str, dict[str, Any]]] = []
+    for table_index, raw_table in enumerate(_require_list(metadata, "tables")):
+        table = _require_dict(raw_table, f"tables[{table_index}]")
+        name = table.get("name", f"tables[{table_index}]")
+        index = _require_dict(table.get("index"), f"tables[{table_index}].index")
+        if index.get("kind") == "values":
+            specs.append((f"{name}.index", index))
+        for column_index, raw_column in enumerate(_require_list(table, "columns")):
+            column = _require_dict(
+                raw_column,
+                f"tables[{table_index}].columns[{column_index}]",
+            )
+            specs.append((f"{name}.{column.get('name', column_index)}", column))
+    specs.append(("strata", _require_dict(metadata.get("strata"), "strata")))
+    return tuple(specs)
+
+
 def _schema_from_metadata(metadata: Mapping[str, Any]) -> EntitySchema:
     raw_schema = _require_dict(metadata.get("schema"), "schema")
     person_entity = _require_string(raw_schema, "person_entity", label="schema")

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

@@ -216,6 +216,7 @@
     US_AOTC_ELIGIBILITY_OUTPUT_COLUMNS,
     US_EDUCATION_INPUTS_NONCONSTANT_PERSON_COLUMNS,
     US_EDUCATION_INPUTS_OUTPUT_COLUMNS,
+    US_EDUCATION_INPUTS_OWNED_OUTPUT_COLUMNS,
     US_EDUCATION_INPUTS_REQUIRED_SOURCE_COLUMNS,
     US_EDUCATION_INPUTS_STAGE_NAME,
     derive_us_education_inputs_from_manifest,
@@ -540,13 +541,17 @@
     PUF_CAPITAL_GAINS_TAIL_QUANTILE,
     PUF_CAPITAL_GAINS_TAIL_STAGE_NAME,
     PUF_CAPITAL_GAINS_TAIL_SUPPORT_CHANNEL,
+    PUF_CAPITAL_GAINS_TAIL_SUPPORT_CONTRACT_VERSION,
     PUF_CAPITAL_GAINS_TAIL_TAX_UNIT_COLUMNS,
     PUF_CAPITAL_GAINS_TAIL_TRANSFER_WEIGHT_COLUMN,
     assert_puf_capital_gains_tail_survives_selection,
     puf_capital_gains_tail_concentration_gate,
+    puf_capital_gains_tail_support_contract_identity,
+    puf_capital_gains_tail_terminal_support_receipt,
     select_puf_capital_gains_tail_donors,
     transfer_puf_capital_gains_tail,
     validate_puf_capital_gains_tail_manifest,
+    validate_puf_capital_gains_tail_terminal_support_receipt,
     write_puf_capital_gains_tail_manifest,
 )
 from microcosm.build.us_runtime.puf_donor_io import load_puf_tax_unit_donor
@@ -1448,6 +1453,7 @@
     "US_AOTC_ELIGIBILITY_OUTPUT_COLUMNS",
     "US_EDUCATION_INPUTS_NONCONSTANT_PERSON_COLUMNS",
     "US_EDUCATION_INPUTS_OUTPUT_COLUMNS",
+    "US_EDUCATION_INPUTS_OWNED_OUTPUT_COLUMNS",
     "ASEC_EDUCATION_ASSISTANCE_ARCHIVES",
     "ASEC_EDUCATION_ASSISTANCE_INCOME_YEARS",
     "fetch_asec_education_assistance_source",
@@ -1811,6 +1817,7 @@
     "PUF_CAPITAL_GAINS_TAIL_POSITIVE_MASS_FIVE_X_TARGET",
     "PUF_CAPITAL_GAINS_TAIL_QUANTILE",
     "PUF_CAPITAL_GAINS_TAIL_STAGE_NAME",
+    "PUF_CAPITAL_GAINS_TAIL_SUPPORT_CONTRACT_VERSION",
     "PUF_CAPITAL_GAINS_TAIL_SUPPORT_CHANNEL",
     "PUF_CAPITAL_GAINS_TAIL_TAX_UNIT_COLUMNS",
     "PUF_CAPITAL_GAINS_TAIL_TRANSFER_WEIGHT_COLUMN",
@@ -1865,6 +1872,8 @@
     "load_asec_h5_tables",
     "out_of_sample_reform_specs",
     "puf_capital_gains_tail_concentration_gate",
+    "puf_capital_gains_tail_support_contract_identity",
+    "puf_capital_gains_tail_terminal_support_receipt",
     "puf_tax_unit_donor_from_arrays",
     "pool_asec_sources",
     "prepare_us_puf_tax_detail_chain_inputs",
@@ -1950,6 +1959,7 @@
     "puf_processed_capital_gains_stage",
     "puf_raw_e01000_stage",
     "validate_puf_capital_gains_tail_manifest",
+    "validate_puf_capital_gains_tail_terminal_support_receipt",
     "validation_only_family_ids",
     "translate_congressional_district_facts_to_current_vintage",
     "with_household_congressional_districts",

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

@@ -324,7 +324,7 @@ def resolve_acs_pums_earnings_universe(
                     person,
                     column=column,
                     raw_source_column=source_column,
-                    scope=scope,
+                    scope=acs_scope,
                 )
                 if raw_source_present
                 else None