Publish PE-native target diagnostics sidecar

src/microplex_us/pipelines/__init__.py

@@ -120,6 +120,7 @@ def _exports(module: str, names: tuple[str, ...]) -> dict[str, str]:
         "microplex_us.pipelines.pe_native_scores",
         (
             "PolicyEngineUSEnhancedCPSNativeScores",
+            "build_us_pe_native_target_diagnostics_payload",
             "compare_us_pe_native_target_deltas",
             "compute_batch_us_pe_native_scores",
             "compute_policyengine_us_enhanced_cps_native_scores",

src/microplex_us/pipelines/artifacts.py

@@ -98,6 +98,7 @@ class USMicroplexArtifactPaths:
     policyengine_harness: Path | None = None
     policyengine_native_scores: Path | None = None
     policyengine_native_audit: Path | None = None
+    policyengine_native_target_diagnostics: Path | None = None
     child_tax_unit_agi_drift: Path | None = None
     capital_gains_lots: Path | None = None
     source_weight_diagnostics: Path | None = None
@@ -1286,6 +1287,7 @@ def save_us_microplex_artifacts(
         policyengine_harness=policyengine_harness_path,
         policyengine_native_scores=policyengine_native_scores_path,
         policyengine_native_audit=None,
+        policyengine_native_target_diagnostics=None,
         child_tax_unit_agi_drift=child_tax_unit_agi_drift_path,
         capital_gains_lots=capital_gains_lots_path,
         source_weight_diagnostics=source_weight_diagnostics_path,

src/microplex_us/pipelines/backfill_pe_native_audit.py

@@ -13,6 +13,7 @@
     discover_us_candidate_artifact_dirs,
 )
 from microplex_us.pipelines.pe_native_scores import (
+    build_us_pe_native_target_diagnostics_payload,
     compute_batch_us_pe_native_support_audits,
     compute_batch_us_pe_native_target_deltas,
 )
@@ -245,10 +246,40 @@ def _write_native_audit_payload_to_bundle(
     artifacts["policyengine_native_audit"] = str(
         native_audit_path.relative_to(bundle_dir)
     )
+    extra_outputs = [native_audit_path.name]
+    target_delta_payload = payload.get("targetDelta")
+    if isinstance(target_delta_payload, dict):
+        target_diagnostics = build_us_pe_native_target_diagnostics_payload(
+            period=int(payload.get("period") or 2024),
+            from_label="policyengine-us-data",
+            to_label="microplex-us",
+            policyengine_targets_db_path=dict(manifest.get("config", {})).get(
+                "policyengine_targets_db"
+            ),
+            target_delta_payload=target_delta_payload,
+            artifact_id=str(payload.get("artifactId") or bundle_dir.name),
+            run_id=str(payload.get("artifactId") or bundle_dir.name),
+        )
+        target_diagnostics_path = resolve_us_stage_artifact_contract_path(
+            bundle_dir,
+            "09_validation_benchmarking",
+            "policyengine_native_target_diagnostics",
+        )
+        target_diagnostics_path.write_text(
+            json.dumps(target_diagnostics, indent=2, sort_keys=True)
+        )
+        artifacts["policyengine_native_target_diagnostics"] = str(
+            target_diagnostics_path.relative_to(bundle_dir)
+        )
+        extra_outputs.append(target_diagnostics_path.name)
     manifest["artifacts"] = artifacts
     manifest["policyengine_native_audit"] = dict(payload.get("verdictHints", {}))
 
-    _refresh_checkpoint_data_flow_snapshot(bundle_dir, manifest)
+    _refresh_checkpoint_data_flow_snapshot(
+        bundle_dir,
+        manifest,
+        extra_outputs=tuple(extra_outputs),
+    )
     assert_valid_benchmark_artifact_manifest(
         manifest,
         artifact_dir=bundle_dir,

src/microplex_us/pipelines/experiments.py

@@ -292,6 +292,11 @@ def to_dict(self) -> dict[str, Any]:
                     if self.artifact_paths.policyengine_native_audit is not None
                     else None
                 ),
+                "policyengine_native_target_diagnostics": (
+                    str(self.artifact_paths.policyengine_native_target_diagnostics)
+                    if self.artifact_paths.policyengine_native_target_diagnostics is not None
+                    else None
+                ),
                 "capital_gains_lots": (
                     str(self.artifact_paths.capital_gains_lots)
                     if self.artifact_paths.capital_gains_lots is not None
@@ -380,6 +385,12 @@ def from_dict(cls, payload: dict[str, Any]) -> USMicroplexExperimentResult:
                     if artifact_paths.get("policyengine_native_audit") is not None
                     else None
                 ),
+                policyengine_native_target_diagnostics=(
+                    Path(artifact_paths["policyengine_native_target_diagnostics"])
+                    if artifact_paths.get("policyengine_native_target_diagnostics")
+                    is not None
+                    else None
+                ),
                 capital_gains_lots=(
                     Path(artifact_paths["capital_gains_lots"])
                     if artifact_paths.get("capital_gains_lots") is not None
@@ -819,6 +830,10 @@ def _refresh_experiment_artifact_paths(
             artifact_root,
             artifacts.get("policyengine_native_audit"),
         ),
+        policyengine_native_target_diagnostics=_resolve_optional_result_artifact_path(
+            artifact_root,
+            artifacts.get("policyengine_native_target_diagnostics"),
+        ),
         run_registry=Path(run_registry_path),
         run_index_db=run_index_path,
     )