Tail assignments deliver the donor's joint vector verbatim (fixes the recipient-value overwrite)

changelog.d/cg-tail-concentration-register.fixed.md

@@ -0,0 +1 @@
+The capital-gains tail transfer delivers every joint-vector column verbatim from the selected donor: the recipient-candidate merge no longer overwrites donor values held at tax-unit grain, fidelity is asserted at construction, and a donor-keyed test binds the chain.

packages/populace-build/src/populace/build/us_runtime/puf_capital_gains_tail.py

@@ -432,6 +432,23 @@ def transfer_puf_capital_gains_tail(
         assigned_weights=assigned_weights,
         candidates=candidates,
     )
+    # Fidelity is asserted by construction (populace#570 review): every
+    # joint-vector column in every assignment must equal the SELECTED
+    # donor's value, keyed by donor_source_id — reconciliation downstream
+    # derives expectations from assignments, so a leak here would
+    # self-confirm if it were not caught at the source.
+    donor_by_id = tail.set_index("tax_unit_id")
+    for column in _JOINT_VECTOR_COLUMNS:
+        expected_vector = donor_by_id.loc[
+            assignments["donor_source_id"], column
+        ].to_numpy(dtype=np.float64)
+        assigned_vector = assignments[column].to_numpy(dtype=np.float64)
+        if not np.array_equal(expected_vector, assigned_vector):
+            raise ValueError(
+                "PUF tail assignments leaked recipient values into "
+                f"{column}; the selected donor's joint vector must arrive "
+                "verbatim."
+            )
     before_distribution = _frame_combined_distribution(frame)
     transferred, clone_receipt = _clone_and_transfer(
         frame,
@@ -805,6 +822,14 @@ def _assign_tail_donors(
         row["donor_source_id"] = int(donor_row["tax_unit_id"])
         del row["tax_unit_id"]
         row.update(candidate.to_dict())
+        # The candidate carries the recipient's EXISTING tax-unit values for
+        # joint-vector columns held at tax-unit grain, so the merge above
+        # would silently replace the donor's transferred vector with the
+        # recipient's old value (populace#570 review, Critical: 99.7% of
+        # donor unrecaptured-1250 mass was lost this way). The selected
+        # donor's joint vector always wins.
+        for column in _JOINT_VECTOR_COLUMNS:
+            row[column] = donor_row[column]
         row["assigned_weight"] = float(assigned_weights[donor_position])
         rows.append(row)
     result = pd.DataFrame(rows)

packages/populace-build/tests/test_us_puf_capital_gains_tail.py

@@ -586,3 +586,54 @@ def test_tail_stratum_passes_existing_weighted_top_100_gate() -> None:
         == tail_count
     )
     assert gate.details["top_share"]["short_term_plus_long_term_capital_gains"] < 0.75
+
+
+def test_joint_vectors_arrive_verbatim_from_selected_donors() -> None:
+    """populace#570 review, Critical: the candidate merge previously
+    replaced donor joint-vector values held at tax-unit grain (notably
+    unrecaptured_section_1250_gain) with the RECIPIENT's existing values —
+    99.7% of intended donor mass lost, and reconciliation self-confirmed
+    because it derives expectations from assignments. Every joint-vector
+    column must arrive verbatim from the SELECTED donor, per-column, into
+    assignments, the manifest, and the materialized frame."""
+    from populace.build.us_runtime.puf_capital_gains_tail import (
+        transfer_puf_capital_gains_tail,
+    )
+
+    frame = _expanded_recipient_frame()
+    donor = _donor()
+    transferred, manifest = transfer_puf_capital_gains_tail(frame, donor, seed=7)
+
+    donor_by_id = donor.set_index("tax_unit_id")
+    joint_columns = (
+        "short_term_capital_gains",
+        "long_term_capital_gains_before_response",
+        "long_term_capital_gains_on_collectibles",
+        "non_sch_d_capital_gains",
+        "unrecaptured_section_1250_gain",
+    )
+    records = manifest["records"]
+    assert records
+    for record in records:
+        source = donor_by_id.loc[record["donor_source_id"]]
+        for column in joint_columns:
+            assert float(record["joint_vector"][column]) == float(source[column]), (
+                f"{column} did not arrive verbatim for donor "
+                f"{record['donor_source_id']}"
+            )
+    # And the materialized frame carries the same values on the tail clones.
+    tax_unit = transferred.table("tax_unit")
+    by_tail = {record["tail_tax_unit_id"]: record for record in records}
+    clone_mask = tax_unit["tax_unit_id"].isin(list(by_tail))
+    assert int(clone_mask.sum()) == len(records)
+    for _, clone in tax_unit.loc[clone_mask].iterrows():
+        record = by_tail[int(clone["tax_unit_id"])]
+        for column in (
+            "long_term_capital_gains_on_collectibles",
+            "non_sch_d_capital_gains",
+            "unrecaptured_section_1250_gain",
+        ):
+            if column in tax_unit.columns:
+                assert float(clone[column]) == float(
+                    record["joint_vector"][column]
+                )