Add PUF-native challenger donor conditioning

Author: MaxGhenis

docs/imputation-conditioning-contract.md

@@ -57,6 +57,12 @@ The current donor-conditioning modes are:
   - use a PE-style structural predictor backbone declared in variable semantics
   - optionally admit a narrow supplemental shared set from the *actual*
     compatible overlap
+- `pe_plus_puf_native_challenger`
+  - keep the same PE structural predictor backbone
+  - for the explicitly marked problematic PUF tax-leaf blocks only, append a
+    narrow source-native raw-overlap set declared in semantics
+  - treat this as a non-default challenger lane, not as part of the PE-aligned
+    contract
 
 For the current PUF IRS tax-leaf family, PE alignment means the structural-only
 path. The local `policyengine-us-data`
@@ -85,6 +91,8 @@ Experimental:
 
 - whether `all_shared`, `top_correlated`, or `pe_prespecified` wins for a given
   block family
+- whether `pe_plus_puf_native_challenger` is worth keeping after a real
+  checkpoint comparison
 - whether a particular variable should admit a
   `supplemental_shared_condition_vars` set
 - which compatible shared predictors should be let back into a PE-structured

docs/methodology-ledger.md

@@ -1813,3 +1813,50 @@ canonical description is:
     conditioning
   - treat any future widening as an explicit challenger experiment using
     source-native PUF predictors, not as a PE-alignment patch
+
+## 2026-04-14 PUF native challenger diagnostic smoke
+
+- run:
+  - `artifacts/live_pe_us_data_rebuild_checkpoint_20260414_pe_plus_puf_native_challenger_diag_smoke/puf-native-challenger-diag-smoke-v1`
+- question:
+  - if we add an explicit non-default challenger lane that keeps the PE
+    structural backbone but appends a narrow source-native PUF overlap, do
+    those vars actually enter the four problematic tax-leaf blocks on a live
+    artifact?
+- setup:
+  - `donor_imputer_condition_selection = pe_plus_puf_native_challenger`
+  - keep the PE structural predictors for the PUF IRS tax-leaf family
+  - append only explicit source-native challengers:
+    - dividend / taxable-interest blocks:
+      `self_employment_income`, `rental_income`,
+      `social_security_retirement`
+    - taxable-pension block:
+      `social_security_retirement`, `social_security_disability`,
+      `unemployment_compensation`
+    - partnership block:
+      `self_employment_income`, `rental_income`, `alimony_income`
+- read:
+  - the challenger vars now enter the live artifact for all four targeted
+    blocks
+  - selected sets were:
+    - dividend split:
+      PE structural backbone + `self_employment_income`, `rental_income`,
+      `social_security_retirement`
+    - `taxable_interest_income`:
+      PE structural backbone + `self_employment_income`, `rental_income`,
+      `social_security_retirement`
+    - `taxable_pension_income`:
+      PE structural backbone + `social_security_retirement`,
+      `social_security_disability`, `unemployment_compensation`
+    - `partnership_s_corp_income`:
+      PE structural backbone + `self_employment_income`, `rental_income`
+      while `alimony_income` failed with `incompatible_condition_support`
+- interpretation:
+  - this clears the immediate blocker from the earlier failed supplement patch:
+    we now have a real opt-in challenger lane whose native PUF predictors are
+    visible in live `donor_conditioning_diagnostics`
+  - the next real question is no longer "can the vars get in?" but "does this
+    challenger help or hurt the PE-oracle losses once we run a full checkpoint"
+- next step:
+  - run one matched broader checkpoint with this challenger mode and compare it
+    against the structural-only PE-aligned default

docs/source-semantics.md

@@ -83,6 +83,11 @@ two distinct selection modes:
     backbone
   - optionally admit a narrow `supplemental_shared_condition_vars` set from the
     actual shared overlap, instead of reopening the full common-predictor pool
+- `pe_plus_puf_native_challenger` selection:
+  - keep the same PE structural backbone
+  - for the explicitly marked problematic PUF tax-leaf blocks only, append a
+    narrow set of source-native raw-overlap predictors declared in semantics
+  - treat that lane as an opt-in challenger, not as a PE-alignment update
 
 For the problematic PUF tax-leaf family, the PE-aligned default is still the
 structural backbone only. The local `policyengine-us-data`
@@ -124,8 +129,11 @@ executed donor block, including:
 - selected condition vars
 - shared vars that were available but dropped
 - requested supplemental shared vars
+- requested challenger shared vars
 - raw-stage supplemental rejection reasons
+- raw-stage challenger rejection reasons
 - prepared-stage supplemental rejection reasons
+- prepared-stage challenger rejection reasons
 - whether the block used a prepared condition surface
 
 Use `python -m microplex_us.pipelines.summarize_donor_conditioning <artifact>`

src/microplex_us/pipelines/pe_us_data_rebuild_checkpoint.py

@@ -4,6 +4,8 @@
 
 import argparse
 import json
+import logging
+import sys
 from dataclasses import dataclass, replace
 from datetime import UTC, datetime
 from pathlib import Path
@@ -67,6 +69,23 @@
 
 DEFAULT_CHECKPOINT_IMPUTATION_ABLATION_EVAL_FRACTION = 0.25
 MIN_CHECKPOINT_IMPUTATION_ABLATION_HOUSEHOLDS = 8
+LOGGER = logging.getLogger(__name__)
+
+
+def _root_logger_has_handlers() -> bool:
+    return bool(logging.getLogger().handlers)
+
+
+def _emit_checkpoint_progress(message: str, /, **context: object) -> None:
+    details = ", ".join(
+        f"{key}={value}"
+        for key, value in context.items()
+        if value is not None and value != ""
+    )
+    line = f"{message} [{details}]" if details else message
+    LOGGER.info(line)
+    if not LOGGER.handlers and not _root_logger_has_handlers():
+        print(line, file=sys.stderr, flush=True)
 
 
 def _resolve_checkpoint_calibration_target_variables(
@@ -1865,6 +1884,14 @@ def run_policyengine_us_data_rebuild_checkpoint(
         "rebuild_profile_expected": True,
         **dict(run_registry_metadata or {}),
     }
+    _emit_checkpoint_progress(
+        "PE-US-data rebuild checkpoint: starting build",
+        output_root=Path(output_root).expanduser(),
+        version_id=version_id or "auto",
+        target_profile=resolved_config.policyengine_target_profile,
+        donor_condition_selection=resolved_config.donor_imputer_condition_selection,
+        providers=",".join(provider_names),
+    )
 
     artifacts = build_and_save_versioned_us_microplex_from_source_providers(
         providers=list(resolved_providers),
@@ -1889,6 +1916,19 @@ def run_policyengine_us_data_rebuild_checkpoint(
         run_registry_metadata=resolved_registry_metadata,
         enable_child_tax_unit_agi_drift=True,
     )
+    _emit_checkpoint_progress(
+        "PE-US-data rebuild checkpoint: build complete",
+        artifact_dir=artifacts.artifact_paths.output_dir,
+        frontier_metric=frontier_metric,
+    )
+    _emit_checkpoint_progress(
+        "PE-US-data rebuild checkpoint: attaching PE evidence",
+        artifact_dir=artifacts.artifact_paths.output_dir,
+        compute_harness=not defer_policyengine_harness,
+        compute_native_scores=not defer_policyengine_native_score,
+        compute_native_audit=not defer_native_audit,
+        compute_imputation_ablation=not defer_imputation_ablation,
+    )
     evidence = attach_policyengine_us_data_rebuild_checkpoint_evidence(
         artifacts.artifact_paths.output_dir,
         build_result=artifacts.build_result,
@@ -1912,11 +1952,21 @@ def run_policyengine_us_data_rebuild_checkpoint(
         run_index_path=run_index_path,
         run_registry_metadata=resolved_registry_metadata,
     )
+    _emit_checkpoint_progress(
+        "PE-US-data rebuild checkpoint: evidence complete",
+        parity_path=evidence.parity_path,
+        native_audit_path=evidence.native_audit_path,
+        imputation_ablation_path=evidence.imputation_ablation_path,
+    )
     refreshed_artifacts = _load_checkpoint_versioned_artifacts(
         build_result=artifacts.build_result,
         artifact_root=artifacts.artifact_paths.output_dir,
         frontier_metric=frontier_metric,
     )
+    _emit_checkpoint_progress(
+        "PE-US-data rebuild checkpoint: checkpoint ready",
+        artifact_dir=refreshed_artifacts.artifact_paths.output_dir,
+    )
     return PEUSDataRebuildCheckpointResult(
         build_config=resolved_config,
         provider_names=provider_names,
@@ -1948,6 +1998,7 @@ def main(argv: list[str] | None = None) -> None:
     parser.add_argument("--calibration-target-profile")
     parser.add_argument("--n-synthetic", type=int, default=100_000)
     parser.add_argument("--random-seed", type=int, default=42)
+    parser.add_argument("--donor-imputer-condition-selection")
     parser.add_argument("--cps-source-year", type=int, default=2023)
     parser.add_argument("--puf-target-year", type=int)
     parser.add_argument("--puf-cps-reference-year", type=int)
@@ -1983,6 +2034,15 @@ def main(argv: list[str] | None = None) -> None:
     parser.add_argument("--require-policyengine-native-score", action="store_true")
     args = parser.parse_args(argv)
 
+    config_overrides = {
+        "n_synthetic": int(args.n_synthetic),
+        "random_seed": int(args.random_seed),
+    }
+    if args.donor_imputer_condition_selection is not None:
+        config_overrides["donor_imputer_condition_selection"] = (
+            args.donor_imputer_condition_selection
+        )
+
     result = run_policyengine_us_data_rebuild_checkpoint(
         output_root=args.output_root,
         policyengine_baseline_dataset=args.baseline_dataset,
@@ -1996,10 +2056,7 @@ def main(argv: list[str] | None = None) -> None:
         calibration_target_variables=tuple(args.calibration_target_variable),
         calibration_target_domains=tuple(args.calibration_target_domain),
         calibration_target_geo_levels=tuple(args.calibration_target_geo_level),
-        config_overrides={
-            "n_synthetic": int(args.n_synthetic),
-            "random_seed": int(args.random_seed),
-        },
+        config_overrides=config_overrides,
         cps_source_year=args.cps_source_year,
         cps_cache_dir=args.cps_cache_dir,
         cps_download=not args.no_cps_download,

src/microplex_us/pipelines/summarize_donor_conditioning.py

@@ -5,8 +5,9 @@
 import argparse
 import json
 from collections import Counter
+from collections.abc import Iterable
 from pathlib import Path
-from typing import Any, Iterable
+from typing import Any
 
 
 def _resolve_artifact_dir(path: str | Path) -> Path:
@@ -50,6 +51,8 @@ def summarize_donor_conditioning(
     dropped_counter: Counter[str] = Counter()
     raw_supplemental_reason_counter: Counter[str] = Counter()
     supplemental_reason_counter: Counter[str] = Counter()
+    raw_challenger_reason_counter: Counter[str] = Counter()
+    challenger_reason_counter: Counter[str] = Counter()
     block_summaries: list[dict[str, Any]] = []
     for entry in diagnostics:
         selected = list(entry.get("selected_condition_vars", []))
@@ -60,6 +63,12 @@ def summarize_donor_conditioning(
         supplemental_status = list(
             entry.get("supplemental_shared_condition_var_status", [])
         )
+        raw_challenger_status = list(
+            entry.get("raw_challenger_shared_condition_var_status", [])
+        )
+        challenger_status = list(
+            entry.get("challenger_shared_condition_var_status", [])
+        )
         selected_counter.update(selected)
         dropped_counter.update(dropped)
         raw_supplemental_reason_counter.update(
@@ -72,6 +81,16 @@ def summarize_donor_conditioning(
             for status in supplemental_status
             if status.get("reason") is not None
         )
+        raw_challenger_reason_counter.update(
+            status.get("reason")
+            for status in raw_challenger_status
+            if status.get("reason") is not None
+        )
+        challenger_reason_counter.update(
+            status.get("reason")
+            for status in challenger_status
+            if status.get("reason") is not None
+        )
         block_summaries.append(
             {
                 "donor_source": entry.get("donor_source"),
@@ -92,8 +111,13 @@ def summarize_donor_conditioning(
                 "requested_supplemental_shared_condition_vars": list(
                     entry.get("requested_supplemental_shared_condition_vars", [])
                 ),
+                "requested_challenger_shared_condition_vars": list(
+                    entry.get("requested_challenger_shared_condition_vars", [])
+                ),
                 "raw_supplemental_shared_condition_var_status": raw_supplemental_status,
+                "raw_challenger_shared_condition_var_status": raw_challenger_status,
                 "supplemental_shared_condition_var_status": supplemental_status,
+                "challenger_shared_condition_var_status": challenger_status,
                 "selected_condition_vars": selected,
                 "dropped_shared_vars": dropped,
             }
@@ -108,9 +132,15 @@ def summarize_donor_conditioning(
         "raw_supplemental_shared_condition_reason_frequency": dict(
             sorted(raw_supplemental_reason_counter.items())
         ),
+        "raw_challenger_shared_condition_reason_frequency": dict(
+            sorted(raw_challenger_reason_counter.items())
+        ),
         "supplemental_shared_condition_reason_frequency": dict(
             sorted(supplemental_reason_counter.items())
         ),
+        "challenger_shared_condition_reason_frequency": dict(
+            sorted(challenger_reason_counter.items())
+        ),
         "blocks": block_summaries,
     }