diff --git a/changelog.d/keogh-puf-refit.fixed.md b/changelog.d/keogh-puf-refit.fixed.md
new file mode 100644
index 00000000..09bf89c5
--- /dev/null
+++ b/changelog.d/keogh-puf-refit.fixed.md
@@ -0,0 +1 @@
+Preserve the base builder's completed retirement-distribution surface after frozen-support selection, preventing a selected rare Keogh donor from being refit and broadcast across PUF-support rows.
diff --git a/packages/populace-build/src/populace/build/us_runtime/retirement_distributions.py b/packages/populace-build/src/populace/build/us_runtime/retirement_distributions.py
index ee99ccc7..15ddaa70 100644
--- a/packages/populace-build/src/populace/build/us_runtime/retirement_distributions.py
+++ b/packages/populace-build/src/populace/build/us_runtime/retirement_distributions.py
@@ -595,16 +595,27 @@ def with_us_retirement_distribution_inputs(
     *,
     seed: int,
     time_period: int,
+    force_puf_imputation: bool = False,
 ) -> Frame:
-    """Materialize measured retirement-distribution leaves on a US frame."""
+    """Materialize measured retirement-distribution leaves on a US frame.
+
+    ``force_puf_imputation`` belongs only at the base builder's post-clone
+    boundary.  Every later support-frame call is consume-only, including when
+    a frozen selection is missing or has flattened a rare leaf.  Refitting
+    there would make support selection redefine the donor universe and can
+    broadcast a rare leaf such as ``keogh_distributions`` across the retained
+    PUF rows.  The downstream signal gate, rather than a refit, owns support
+    surface completeness and signal.
+    """
 
     if frame.schema != US_SCHEMA:
         raise ValueError("US retirement distributions require the US schema.")
     person = frame.table("person")
     has_support_channels = _PERSON_SUPPORT_CHANNEL_COLUMN in person.columns
-    if (
-        _retirement_distribution_surface_carries_signal(frame)
-        and not has_support_channels
+    if has_support_channels and not force_puf_imputation:
+        return frame
+    if not has_support_channels and _retirement_distribution_surface_carries_signal(
+        frame
     ):
         return frame
 
diff --git a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py
index e979d274..76be49f9 100644
--- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py
+++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py
@@ -119,9 +119,9 @@ def test__given_target_frame_checkpoint__then_builder_round_trips_frame(
         ssi_take_up_assignment_sha256="ssi-flags-sha",
         selection_identities_sha256=None,
     )
-    # 9 = the #374 SIPP+SCF blend changes the pre-materialization frame;
-    # SCF-only checkpoints (8 = post-#539 ORG rewrite) must not be reused.
-    assert identity["materializer_version"] == 9
+    # 10 = #557 preserves the staged retirement surface through release
+    # materialization; pre-#557 QRF-refitted checkpoints (9) must not serve.
+    assert identity["materializer_version"] == 10
     # The SSI prior-weight basis is identity-bearing (populace#543 instance
     # 2): unflagged runs carry the key as None.
     assert identity["ssi_take_up_prior_weight_basis_sha256"] is None
@@ -179,10 +179,10 @@ def test__given_stale_materializer_version_checkpoint__then_builder_rejects_it(
 ) -> None:
     """A checkpoint stored under a superseded materializer version must not load.
 
-    #539 rewrote staged org-wage inputs without bumping the materializer
-    version, so version-7 checkpoints carry pre-fix ORG rows (populace#543).
-    The version constant participates in the identity comparison; this pins
-    the rejection path for the stored-7 vs current shape specifically.
+    #557 changed the staged retirement-surface semantics: version-9
+    checkpoints can carry release-refitted leaves instead of the preserved
+    support-built surface. The version constant participates in the identity
+    comparison; this pins the stored-9 versus current-10 rejection directly.
     """
     builder = _load_builder_module()
     monkeypatch.setattr(builder, "US_SCHEMA", small_frame.schema)
@@ -216,10 +216,10 @@ def test__given_stale_materializer_version_checkpoint__then_builder_rejects_it(
         ssi_take_up_assignment_sha256="ssi-flags-sha",
         selection_identities_sha256=None,
     )
-    # 8 = current-main SCF-only-era checkpoints (the #510 review's stale
-    # hazard); 7 = pre-#539. Both must miss against expected version 9.
-    stale_identity = {**dict(identity), "materializer_version": 8}
-    older_identity = {**dict(identity), "materializer_version": 7}
+    # 9 = the pre-#557 release-refit world; 8 = the still-older pre-#374 blend
+    # world. Both must miss against expected version 10.
+    stale_identity = {**dict(identity), "materializer_version": 9}
+    older_identity = {**dict(identity), "materializer_version": 8}
     path = tmp_path / "target_frame_checkpoint.h5"
     builder._write_target_frame_checkpoint(
         path,
@@ -2972,7 +2972,10 @@ def test_release_calibration_diagnostics_writes_nan_final_loss_as_null(
     assert diagnostics["build"]["default_dataset"]["final_loss"] is None
 
 
-@pytest.mark.parametrize("terminal_mode", ["merge", "integrity", "crash", "telemetry"])
+@pytest.mark.parametrize(
+    "terminal_mode",
+    ["merge", "integrity", "retirement", "crash", "telemetry"],
+)
 def test_main_writes_diagnostics_before_post_calibration_gate_failure(
     monkeypatch, tmp_path, terminal_mode
 ) -> None:
@@ -2985,6 +2988,9 @@ def test_main_writes_diagnostics_before_post_calibration_gate_failure(
     Bernoulli-law violation; that failure reaches the written diagnostics
     and the single terminal batch while the delivery gate passes and every
     later terminal group still runs.
+    ``retirement``: an incomplete consume-only retirement surface reports its
+    missing leaves through the same written diagnostics and terminal batch,
+    while later terminal groups still run.
     ``crash``: the degraded-mode guards themselves are exercised — the
     health-input evaluation raises, the incumbent path is missing, and
     ``_release_gate_failures`` raises; each records a line instead of
@@ -3027,6 +3033,9 @@ def test_main_writes_diagnostics_before_post_calibration_gate_failure(
         "source_stage_events": [],
         "terminal_gate_events": [],
     }
+    retirement_missing_failure = (
+        "person columns missing: ['taxable_403b_distributions', 'keogh_distributions']."
+    )
 
     class FakeFrame:
         def n(self, entity):
@@ -3102,7 +3111,7 @@ def complete(self):
             "_staging_telemetry",
             lambda *args, **kwargs: live_telemetry,
         )
-    if terminal_mode in {"integrity", "telemetry"}:
+    if terminal_mode in {"integrity", "retirement", "telemetry"}:
         monkeypatch.setattr(
             builder,
             "PolicyEngineUSEngine",
@@ -3663,14 +3672,29 @@ def fake_energy_subsidy_signal_gate(frame):
             details={"checked": True},
         ),
     )
+
+    def fake_retirement_distributions_signal_gate(frame):
+        missing = terminal_mode == "retirement"
+        return builder.GateResult(
+            name="retirement_distributions_signal",
+            passed=not missing,
+            failures=((retirement_missing_failure,) if missing else ()),
+            details=(
+                {
+                    "missing": [
+                        "taxable_403b_distributions",
+                        "keogh_distributions",
+                    ]
+                }
+                if missing
+                else {"checked": True}
+            ),
+        )
+
     monkeypatch.setattr(
         builder,
         "us_retirement_distributions_signal_gate",
-        lambda frame: builder.GateResult(
-            name="retirement_distributions_signal",
-            passed=True,
-            details={"checked": True},
-        ),
+        fake_retirement_distributions_signal_gate,
     )
     monkeypatch.setattr(
         builder,
@@ -4277,14 +4301,29 @@ def fake_materialize_target_frame(frame, specs, **kwargs):
         captured["materialize_kwargs"] = kwargs
         return frame, registry, {"dropped_target_names": []}
 
-    monkeypatch.setattr(
-        builder,
-        "_degenerate_input_signal_gate",
-        lambda frame, engine: builder.GateResult(
+    def fake_degenerate_input_signal_gate(frame, engine):
+        # In retirement mode this gate ALSO fails: the production masking
+        # route (PR #557 round 2 finding 1) was the generic degenerate raise
+        # superseding the specific missing-leaf diagnosis. The degraded-mode
+        # append must carry BOTH lines to the single terminal batch while the
+        # run continues through the solve (the #547/#548 evidence contract).
+        if terminal_mode == "retirement":
+            return builder.GateResult(
+                name="degenerate_input_signal",
+                passed=False,
+                failures=("keogh_distributions flattened [degenerate-sentinel]",),
+                details={"checked": True},
+            )
+        return builder.GateResult(
             name="degenerate_input_signal",
             passed=True,
             details={"checked": True},
-        ),
+        )
+
+    monkeypatch.setattr(
+        builder,
+        "_degenerate_input_signal_gate",
+        fake_degenerate_input_signal_gate,
     )
     monkeypatch.setattr(
         builder,
@@ -4362,11 +4401,11 @@ def fake_final_medicaid_diagnostics(
     def fake_ssi_delivery_gate(diagnostics, *, targets):
         captured["ssi_delivery_gate_called"] = True
         captured["ssi_delivery_gate_targets"] = dict(targets)
-        # The integrity case passes delivery to isolate FINAL-INTEGRITY
-        # collection. Other modes retain the populace#547 delivery cofailure
-        # and its written retry basis.
+        # The integrity and retirement cases pass delivery to isolate their
+        # own early failure. Other modes retain the populace#547 delivery
+        # cofailure and its written retry basis.
         captured.setdefault("ssi_event_order", []).append("delivery_gate")
-        passes = terminal_mode == "integrity"
+        passes = terminal_mode in {"integrity", "retirement"}
         return builder.GateResult(
             name="ssi_take_up_delivery",
             passed=passes,
@@ -4404,6 +4443,22 @@ def recording_ssi_write(diagnostics, path):
     def fake_release_gate_failures(*args, **kwargs):
         if terminal_mode == "crash":
             raise RuntimeError("release-gate evaluation exploded [crash-sentinel]")
+        if terminal_mode == "retirement":
+            # The degraded pre-solve contract (PR #557 round 3): the failing
+            # degenerate gate is NOT raised early — the gate object itself
+            # must arrive here, failures intact, and ride the single
+            # terminal batch. Emitting from the argument (the real
+            # function's contract) proves the un-raised object reached us.
+            degenerate_gate = args[8]
+            assert degenerate_gate is not None
+            assert not degenerate_gate.passed
+            return [
+                *(
+                    f"Degenerate input signal failed: {failure}"
+                    for failure in degenerate_gate.failures
+                ),
+                "ctc failed",
+            ]
         return ["ctc failed"]
 
     monkeypatch.setattr(
@@ -4431,7 +4486,20 @@ def fake_release_gate_failures(*args, **kwargs):
         # coverage/parity evaluation errors on the fake frame may append
         # further lines after them.
         message = str(exc)
-        if terminal_mode == "integrity":
+        if terminal_mode == "retirement":
+            assert message.startswith(
+                "Release gates failed: Retirement-distribution signal failed: "
+                f"{retirement_missing_failure}"
+            )
+            # The co-failing degenerate gate must batch AFTER the specific
+            # retirement diagnosis, never supersede it with an early raise
+            # (PR #557 round 2 finding 1).
+            assert (
+                "Degenerate input signal failed: keogh_distributions "
+                "flattened [degenerate-sentinel]" in message
+            )
+            assert "SSI take-up delivery failed:" not in message
+        elif terminal_mode == "integrity":
             assert message.startswith(
                 "Release gates failed: SSI take-up final measurement failed: "
                 "Bernoulli-law violation [final-integrity-sentinel]"
@@ -4465,7 +4533,13 @@ def fake_release_gate_failures(*args, **kwargs):
     written_diagnostics = json.loads(
         (release_dir / "calibration_diagnostics.json").read_text()
     )
-    if terminal_mode == "integrity":
+    if terminal_mode == "retirement":
+        assert (
+            "Retirement-distribution signal failed: "
+            f"{retirement_missing_failure}"
+            in written_diagnostics["build"]["release_gates"]["failures"]
+        )
+    elif terminal_mode == "integrity":
         assert (
             "SSI take-up final measurement failed: "
             "Bernoulli-law violation [final-integrity-sentinel]"
@@ -4480,22 +4554,26 @@ def fake_release_gate_failures(*args, **kwargs):
     assert not list(release_dir.glob("*manifest*"))
     if terminal_mode == "telemetry":
         assert captured["telemetry_crashed"] is True
-    if terminal_mode in {"integrity", "telemetry"}:
+    if terminal_mode in {"integrity", "retirement", "telemetry"}:
         assert captured["terminal_gate_events"] == [
             "input_coverage",
             "input_mass_parity",
             "qrf_tail_concentration",
         ]
     if terminal_mode != "crash":
-        # The retry line carries the written artifact's sha256 — the
-        # required --ssi-take-up-prior-weight-basis-sha256 pin, handed out
-        # by the failure itself (sol round 2, new minor).
-        import hashlib
-
-        written_sha = hashlib.sha256(
-            (release_dir / "us_ssi_take_up.json").read_bytes()
-        ).hexdigest()
-        if terminal_mode == "integrity":
+        if terminal_mode == "retirement":
+            expected_gate_failures = [
+                f"Retirement-distribution signal failed: {retirement_missing_failure}",
+                "Other health insurance signal failed on the export frame: "
+                "premiums signal flattened [cofailure-sentinel]",
+                # The co-failing pre-solve degenerate gate is never raised
+                # early and never duplicated: its line arrives once, through
+                # _release_gate_failures (PR #557 round 3).
+                "Degenerate input signal failed: keogh_distributions "
+                "flattened [degenerate-sentinel]",
+                "ctc failed",
+            ]
+        elif terminal_mode == "integrity":
             expected_gate_failures = [
                 "SSI take-up final measurement failed: "
                 "Bernoulli-law violation [final-integrity-sentinel]",
@@ -4508,6 +4586,14 @@ def fake_release_gate_failures(*args, **kwargs):
                 "ctc failed",
             ]
         else:
+            # The retry line carries the written artifact's sha256 — the
+            # required --ssi-take-up-prior-weight-basis-sha256 pin, handed out
+            # by the failure itself (sol round 2, new minor).
+            import hashlib
+
+            written_sha = hashlib.sha256(
+                (release_dir / "us_ssi_take_up.json").read_bytes()
+            ).hexdigest()
             expected_gate_failures = [
                 "SSI take-up delivery failed: 18_64 delivered over envelope "
                 "[cofailure-sentinel]",
@@ -4688,18 +4774,27 @@ def fake_release_gate_failures(*args, **kwargs):
     assert captured["ssi_delivery_gate_targets"] == fake_band_targets
     # The frozen-assignment digest invalidates the materialization cache on
     # any retry whose flags differ (populace#507/#508 split-brain fix).
-    assert (
-        captured["materialize_kwargs"]["target_materialization_cache_context"][
-            "ssi_take_up_assignment_sha256"
-        ]
-        == "ssi-digest-sentinel"
-    )
-    assert (
-        captured["materialize_kwargs"]["target_materialization_cache_context"][
-            "selection_identities_sha256"
-        ]
-        is None
+    cache_context = captured["materialize_kwargs"][
+        "target_materialization_cache_context"
+    ]
+    assert cache_context["ssi_take_up_assignment_sha256"] == "ssi-digest-sentinel"
+    assert cache_context["selection_identities_sha256"] is None
+    expected_materializer_identity = builder._target_frame_checkpoint_identity(
+        base_dataset_sha256=cache_context["base_dataset_sha256"],
+        policyengine_us_version=cache_context["policyengine_us_version"],
+        seed=cache_context["seed"],
+        target_period=cache_context["target_period"],
+        target_registry_version=cache_context["target_registry_version"],
+        weeks_unemployed_source_sha256=cache_context["weeks_unemployed_source_sha256"],
+        congressional_district_vintage_crosswalk_sha256=cache_context[
+            "congressional_district_vintage_crosswalk_sha256"
+        ],
+        ssi_take_up_assignment_sha256=cache_context["ssi_take_up_assignment_sha256"],
+        selection_identities_sha256=cache_context["selection_identities_sha256"],
     )
+    assert cache_context[
+        "target_frame_materializer_identity_sha256"
+    ] == builder._target_frame_checkpoint_digest(expected_materializer_identity)
     # Evidence-first ordering (sol round 2, findings 3/10, reconciled with
     # the #548 batched terminal gates): the final measurement hits disk
     # BEFORE the final integrity gate runs. Delivery still evaluates after
@@ -4711,7 +4806,7 @@ def fake_release_gate_failures(*args, **kwargs):
         "integrity_gate",  # persisted-flag recheck on the export frame
         "delivery_gate",  # enforced-band delivery, after the artifact exists
     ]
-    if terminal_mode != "integrity":
+    if terminal_mode not in {"integrity", "retirement"}:
         # A delivery miss rewrites the final measurement as the retry basis.
         expected_ssi_event_order.append("write:us_ssi_take_up.json")
     assert captured["ssi_event_order"] == expected_ssi_event_order
@@ -5825,7 +5920,18 @@ def fake_assert_no_formula_owned_columns(frame_arg):
         "seed": 0,
         "target_period": builder.PERIOD,
         "target_registry_version": "test-target-registry",
+        # Required declaration (PR #557): the reform-vector projection
+        # fail-closes without it — see the dedicated rejection test.
+        "target_frame_materializer_identity_sha256": "test-materializer-digest",
     }
+    with pytest.raises(ValueError, match="target_frame_materializer_identity_sha256"):
+        builder._reform_vector_cache_context(
+            {
+                k: v
+                for k, v in cache_context.items()
+                if k != "target_frame_materializer_identity_sha256"
+            }
+        )
     target_frame, registry, dropped = builder._materialize_target_frame(
         frame,
         (target,),
@@ -5944,6 +6050,67 @@ def test_target_materialization_cache_rejects_value_hash_mismatch(tmp_path) -> N
         )
 
 
+def test_target_materialization_cache_rejects_pre_557_identities(tmp_path) -> None:
+    """Schema-2 and pre-preservation materializer vectors cannot serve."""
+
+    builder = _load_builder_module()
+    assert builder.TARGET_MATERIALIZATION_CACHE_SCHEMA_VERSION == 3
+    reform_spec = SimpleNamespace(
+        measure="jct_mock_tax_expenditure",
+        neutralized_variable="mock_credit",
+    )
+    current_context = {
+        "target_frame_materializer_identity_sha256": "version-10-preserved-surface",
+    }
+    current_identity = builder._target_materialization_cache_identity(
+        context=current_context,
+        reform_spec=reform_spec,
+        n_households=2,
+    )
+
+    for stale_schema in (2, 1):
+        stale_identity = {
+            **current_identity,
+            "schema_version": stale_schema,
+        }
+        builder._write_reform_income_tax_cache(
+            tmp_path,
+            stale_identity,
+            np.asarray([1.0, 2.0]),
+        )
+        assert (
+            builder._read_reform_income_tax_cache(
+                tmp_path,
+                current_identity,
+                n_households=2,
+            )
+            is None
+        )
+
+    pre_557_identity = builder._target_materialization_cache_identity(
+        context={
+            "target_frame_materializer_identity_sha256": (
+                "version-9-release-refitted-surface"
+            ),
+        },
+        reform_spec=reform_spec,
+        n_households=2,
+    )
+    builder._write_reform_income_tax_cache(
+        tmp_path,
+        pre_557_identity,
+        np.asarray([3.0, 4.0]),
+    )
+    assert (
+        builder._read_reform_income_tax_cache(
+            tmp_path,
+            current_identity,
+            n_households=2,
+        )
+        is None
+    )
+
+
 def test_soi_eitc_child_targets_materialize_distinct_child_slices(
     monkeypatch,
 ) -> None:
@@ -7745,6 +7912,7 @@ def _base_cache_context(builder):
         "target_period": builder.PERIOD,
         "target_registry_version": "registry-A",
         "congressional_district_vintage_crosswalk_sha256": None,
+        "target_frame_materializer_identity_sha256": "materializer-sha-A",
     }
 
 
@@ -7893,6 +8061,7 @@ def test__given_changed_reform_vector__then_stale_checkpoint_is_not_reused(
     [
         ("base_dataset_sha256", "base-sha-B"),
         ("weeks_unemployed_source_sha256", "weeks-source-sha-B"),
+        ("target_frame_materializer_identity_sha256", "materializer-sha-B"),
     ],
 )
 def test__given_changed_frame_identity__then_stale_checkpoint_is_not_reused(
@@ -7926,8 +8095,8 @@ def test__given_changed_frame_identity__then_stale_checkpoint_is_not_reused(
     )
     assert calls_a == ["credit_a"]
 
-    # A different base H5 or measured LKWEEKS source must not share
-    # per-household vectors even at the same record count.
+    # A different base H5, measured LKWEEKS source, or complete target-frame
+    # materializer identity must not share vectors at the same record count.
     context_b = _base_cache_context(builder)
     context_b[identity_key] = new_value
     calls_b: list[str] = []
@@ -7955,9 +8124,8 @@ def test__given_only_build_commit_changed__then_reform_cache_is_reused(
     monkeypatch,
     tmp_path,
 ) -> None:
-    # #217 acceptance criterion 1: a rerun that changes only build state that does
-    # not affect per-household reform estimates (build commit, seed, registry
-    # version) must reuse the cached reform vectors rather than recompute them.
+    # #217 acceptance criterion 1: a rerun that changes only the build commit
+    # must reuse the cached reform vectors rather than recompute them.
     builder = _load_builder_module()
     frame = _multi_reform_frame(builder)
     reforms = (("jct_reform_a", "credit_a"),)
@@ -7982,13 +8150,10 @@ def test__given_only_build_commit_changed__then_reform_cache_is_reused(
     )
     assert calls_a == ["credit_a"]
 
-    # Only build_commit / seed / target_registry_version change (a code-only or
-    # calibration-only rerun). Everything that determines the reform estimate is
-    # identical, so the reform must load from cache.
+    # Only build_commit changes. The full materializer identity remains equal,
+    # so the reform must load from cache.
     context_b = _base_cache_context(builder)
     context_b["build_commit"] = "commit-B"
-    context_b["seed"] = 12345
-    context_b["target_registry_version"] = "registry-B"
     calls_b: list[str] = []
     _install_multi_reform_fakes(
         builder,
@@ -9029,14 +9194,14 @@ def test_final_medicaid_green_path_evaluates_normally() -> None:
     assert failures == []
 
 
-def test_reform_vector_cache_context_tracks_assignment_and_selection() -> None:
-    """The #217 reform-vector cache whitelist carries both support digests.
+def test_reform_vector_cache_context_tracks_support_and_materializer() -> None:
+    """The reform-vector whitelist carries support and materializer digests.
 
     Whether a JCT reform income-tax estimate can move with
     takes_up_ssi_if_eligible is an engine-graph question the build must not
     answer by assumption, while two selected supports can share positional
-    SSI flag bytes. The assignment and selection digests therefore invalidate
-    reform vectors independently."""
+    SSI flag bytes. The assignment, selection, and complete target-frame
+    materializer digests therefore invalidate reform vectors independently."""
 
     builder = _load_builder_module()
     base = {
@@ -9048,10 +9213,15 @@ def test_reform_vector_cache_context_tracks_assignment_and_selection() -> None:
         "build_commit": "irrelevant-to-reform-vectors",
         "ssi_take_up_assignment_sha256": "digest-a",
         "selection_identities_sha256": None,
+        "target_frame_materializer_identity_sha256": "materializer-a",
     }
     changed_assignment = {**base, "ssi_take_up_assignment_sha256": "digest-b"}
     selected = {**base, "selection_identities_sha256": "cd" * 32}
     selected_other = {**base, "selection_identities_sha256": "ef" * 32}
+    changed_materializer = {
+        **base,
+        "target_frame_materializer_identity_sha256": "materializer-b",
+    }
     projected = builder._reform_vector_cache_context(base)
     assert builder._reform_vector_cache_context(
         base
@@ -9060,8 +9230,14 @@ def test_reform_vector_cache_context_tracks_assignment_and_selection() -> None:
     assert builder._reform_vector_cache_context(
         selected
     ) != builder._reform_vector_cache_context(selected_other)
+    assert projected != builder._reform_vector_cache_context(changed_materializer)
     assert "selection_identities_sha256" in builder.REFORM_VECTOR_CACHE_CONTEXT_KEYS
     assert projected["selection_identities_sha256"] is None
+    assert (
+        "target_frame_materializer_identity_sha256"
+        in builder.REFORM_VECTOR_CACHE_CONTEXT_KEYS
+    )
+    assert projected["target_frame_materializer_identity_sha256"] == "materializer-a"
     assert "build_commit" not in projected
 
 
diff --git a/packages/populace-build/tests/test_us_puf_support_base_builder.py b/packages/populace-build/tests/test_us_puf_support_base_builder.py
index b026a29d..27cd6c59 100644
--- a/packages/populace-build/tests/test_us_puf_support_base_builder.py
+++ b/packages/populace-build/tests/test_us_puf_support_base_builder.py
@@ -1043,7 +1043,7 @@ def test_main_runs_cps_only_inputs_before_clone_and_after_puf_then_fails_gate(
     salt_refund_gate_frames: list[object] = []
     adult_care_gate_frames: list[object] = []
     energy_subsidy_gate_frames: list[object] = []
-    retirement_distribution_calls: list[tuple[object, int, int]] = []
+    retirement_distribution_calls: list[tuple[object, int, int, bool]] = []
     retirement_distribution_gate_frames: list[object] = []
     prior_year_income_calls: list[tuple[object, int, int]] = []
     prior_year_income_gate_frames: list[object] = []
@@ -1255,8 +1255,16 @@ def fake_with_weeks_unemployed(
         lambda frame, *, seed, time_period: frame,
     )
 
-    def fake_retirement_distributions(frame, *, seed, time_period):
-        retirement_distribution_calls.append((frame, seed, time_period))
+    def fake_retirement_distributions(
+        frame,
+        *,
+        seed,
+        time_period,
+        force_puf_imputation=False,
+    ):
+        retirement_distribution_calls.append(
+            (frame, seed, time_period, force_puf_imputation)
+        )
         if frame == "disability-benefits-direct":
             return "retirement-distributions-direct"
         return "retirement-distributions-puf"
@@ -1652,10 +1660,12 @@ def fake_retirement_distributions_signal_gate(frame):
         if failing_gate in {"energy_subsidy", "retirement_distributions"}
         else []
     )
-    expected_retirement_distribution_calls = [("disability-benefits-direct", 7, 2024)]
+    expected_retirement_distribution_calls = [
+        ("disability-benefits-direct", 7, 2024, False)
+    ]
     if failing_gate == "retirement_distributions":
         expected_retirement_distribution_calls.append(
-            ("disability-benefits-puf", 7, 2024)
+            ("disability-benefits-puf", 7, 2024, True)
         )
     assert retirement_distribution_calls == expected_retirement_distribution_calls
     assert retirement_distribution_gate_frames == (
@@ -1665,6 +1675,22 @@ def fake_retirement_distributions_signal_gate(frame):
     )
 
 
+def test_resume_retirement_stage_forces_puf_imputation() -> None:
+    """The resume-side ownership boundary is pinned (PR #557 round 2, low).
+
+    The live post-clone boundary is behaviorally asserted above; this pins
+    the named-stage resume branch so deleting its force flag fails a test
+    (source-pin precedent: the main-summary gate tests below).
+    """
+    builder = _load_support_builder_module()
+    source = Path(builder.__file__).read_text(encoding="utf-8")
+    marker = 'elif stage == "retirement_distributions_post_clone":'
+    assert marker in source
+    window = source.split(marker, 1)[1].split("elif ", 1)[0]
+    assert "with_us_retirement_distribution_inputs(" in window
+    assert "force_puf_imputation=True" in window
+
+
 def test_main_summary_records_retirement_distribution_gate() -> None:
     builder = _load_support_builder_module()
     source = Path(builder.__file__).read_text(encoding="utf-8")
diff --git a/packages/populace-build/tests/test_us_retirement_distributions.py b/packages/populace-build/tests/test_us_retirement_distributions.py
index 00aac069..7f80cf16 100644
--- a/packages/populace-build/tests/test_us_retirement_distributions.py
+++ b/packages/populace-build/tests/test_us_retirement_distributions.py
@@ -316,7 +316,12 @@ def fit(
             return FakeFitted()
 
     monkeypatch.setattr(module, "QRF", FakeQRF)
-    result = with_us_retirement_distribution_inputs(expanded, seed=7, time_period=2024)
+    result = with_us_retirement_distribution_inputs(
+        expanded,
+        seed=7,
+        time_period=2024,
+        force_puf_imputation=True,
+    )
 
     assert calls["init"] == {"n_estimators": 100, "seed": 7}
     assert len(calls["training"]) == len(direct.table("person"))
@@ -342,6 +347,126 @@ def fit(
     assert gate.passed, gate.failures
 
 
+def test_completed_puf_surface_survives_narrowed_support_without_refit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    direct = with_us_retirement_distribution_inputs(_frame(), seed=0, time_period=2024)
+    expanded = clone_us_frame_for_puf_support(direct)
+
+    class ZeroFitted:
+        def predict(self, test: pd.DataFrame, **kwargs) -> pd.DataFrame:
+            return pd.DataFrame(
+                0.0,
+                index=test.index,
+                columns=list(_PUF_QRF_OUTPUTS),
+            )
+
+    class ZeroQRF:
+        def __init__(self, **kwargs: object) -> None:
+            pass
+
+        def fit(self, *args: object, **kwargs: object) -> ZeroFitted:
+            return ZeroFitted()
+
+    monkeypatch.setattr(module, "QRF", ZeroQRF)
+    completed = with_us_retirement_distribution_inputs(
+        expanded,
+        seed=7,
+        time_period=2024,
+        force_puf_imputation=True,
+    )
+
+    # Model frozen-support recovery by removing one all-zero PUF household.
+    # Frame.select deliberately preserves the surviving person index, so the
+    # selected frame also covers the non-RangeIndex path that changed the
+    # historical 5,000-row donor sample.
+    person = completed.table("person")
+    puf_zero = person["person_support_channel"].eq("puf_tax_detail") & ~person[
+        list(_PUF_QRF_OUTPUTS)
+    ].any(axis=1)
+    drop_index = person.index[puf_zero][0]
+    selected = completed.select(person.index != drop_index)
+    before = us_retirement_distributions_signal_gate(selected)
+    assert before.passed, before.failures
+    before_share = before.details["nonzero_shares"]["keogh_distributions"]
+    assert 0.0000001 <= before_share <= 0.005
+    before_values = selected.table("person")[list(_OUTPUTS)].copy()
+    before_keogh_carriers = int((before_values["keogh_distributions"] > 0).sum())
+
+    class UnexpectedQRF:
+        def __init__(self, **kwargs: object) -> None:
+            raise AssertionError("a completed retirement surface must not be refit")
+
+    monkeypatch.setattr(module, "QRF", UnexpectedQRF)
+    result = with_us_retirement_distribution_inputs(
+        selected,
+        seed=7,
+        time_period=2024,
+    )
+
+    assert result is selected
+    after = us_retirement_distributions_signal_gate(result)
+    assert after.passed, after.failures
+    assert after.details["nonzero_shares"]["keogh_distributions"] == before_share
+    pd.testing.assert_frame_equal(
+        result.table("person")[list(_OUTPUTS)],
+        before_values,
+    )
+    assert (
+        result.table("person")["keogh_distributions"].gt(0).sum()
+        == before_keogh_carriers
+    )
+
+    # If support selection removes every rare carrier from one leaf, preserve
+    # the completed surface and fail closed at the gate instead of refitting.
+    # Removing the measured carrier rows keeps source reconciliation valid and
+    # isolates the degeneration/prevalence checks this regression owns.
+    keogh_carriers = result.table("person")["keogh_distributions"].gt(0)
+    assert keogh_carriers.any()
+    selected_away = result.select(~keogh_carriers)
+    degenerate = with_us_retirement_distribution_inputs(
+        selected_away,
+        seed=7,
+        time_period=2024,
+    )
+    assert degenerate is selected_away
+    degenerate_gate = us_retirement_distributions_signal_gate(degenerate)
+    assert not degenerate_gate.passed
+    assert degenerate_gate.failures == (
+        "keogh_distributions: degenerate with 1 distinct value(s).",
+        "keogh_distributions: weighted nonzero share 0.00000000 outside "
+        "[0.00000010, 0.00500000].",
+    )
+    assert degenerate_gate.details["source_mismatches"]["keogh_distributions"] == 0
+
+
+@pytest.mark.parametrize("missing", _OUTPUTS)
+def test_incomplete_puf_surface_is_consume_only_and_fails_at_the_signal_gate(
+    monkeypatch: pytest.MonkeyPatch,
+    missing: str,
+) -> None:
+    direct = with_us_retirement_distribution_inputs(_frame(), seed=0, time_period=2024)
+    incomplete = clone_us_frame_for_puf_support(direct)
+    incomplete.table("person").drop(columns=[missing], inplace=True)
+
+    class UnexpectedQRF:
+        def __init__(self, **kwargs: object) -> None:
+            raise AssertionError("an incomplete support surface must not be refit")
+
+    monkeypatch.setattr(module, "QRF", UnexpectedQRF)
+    result = with_us_retirement_distribution_inputs(
+        incomplete,
+        seed=7,
+        time_period=2024,
+    )
+
+    assert result is incomplete
+    gate = us_retirement_distributions_signal_gate(result)
+    assert not gate.passed
+    assert gate.failures == (f"person columns missing: {[missing]}.",)
+    assert gate.details == {"missing": [missing]}
+
+
 def test_gate_rejects_a_default_or_source_divergent_leaf() -> None:
     result = with_us_retirement_distribution_inputs(_frame(), seed=0, time_period=2024)
     result.table("person")["keogh_distributions"] = 0.0
diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py
index 42ff1eda..20c83e50 100644
--- a/tools/build_us_fiscal_refresh_release.py
+++ b/tools/build_us_fiscal_refresh_release.py
@@ -286,14 +286,15 @@
 # the inputs that actually determine per-household reform estimates and no longer
 # includes build_commit / seed / target_registry_version. Old (v1) coarse-key
 # entries live under different filenames, so a mixed cache dir never collides.
-TARGET_MATERIALIZATION_CACHE_SCHEMA_VERSION = 2
+# Bumped 2 -> 3 for #557: absolute reform vectors now bind to the target-frame
+# materializer identity. Pre-#557 vectors can reflect release-refitted retirement
+# leaves and must not mix with a preserved-surface baseline.
+TARGET_MATERIALIZATION_CACHE_SCHEMA_VERSION = 3
 # The subset of the materialization cache context that determines per-household JCT
-# reform income-tax vectors (#217). Anything outside this set — build commit, RNG
-# seed, target registry version, calibration settings — cannot change a reform's
-# per-household estimate, so it is deliberately excluded from the reform-vector
-# cache key. That lets a restart at a newer build commit (or after a registry-only
-# change) reuse already-materialized reforms instead of recomputing all of them,
-# while a change to any of these keys still invalidates the entry (no stale reuse).
+# reform income-tax vectors (#217). The raw build commit and calibration settings
+# stay excluded. The materializer-identity digest transitively binds staged-frame
+# semantics, seed, registry, and support selection, so any such change invalidates
+# the vectors even when the on-disk base hash is stable.
 REFORM_VECTOR_CACHE_CONTEXT_KEYS: tuple[str, ...] = (
     "base_dataset_sha256",
     "weeks_unemployed_source_sha256",
@@ -311,6 +312,10 @@
     # describe. Same-length supports can share positional SSI flag bytes, so
     # this digest remains independent of the assignment digest.
     "selection_identities_sha256",
+    # Absolute reform-tax vectors are later subtracted from the freshly
+    # materialized baseline. Bind them to the complete target-frame identity
+    # so pre-#557 release-refitted surfaces cannot mix with preserved surfaces.
+    "target_frame_materializer_identity_sha256",
 )
 TARGET_FRAME_CHECKPOINT_SCHEMA_VERSION = 1
 # 2: the medicaid_take_up stage (populace #331) changed base_frame's
@@ -340,7 +345,10 @@
 # reused (populace#543, post-merge audit).
 # 9: #374 SIPP+SCF financial-asset blend changes the pre-materialization
 # frame; warm SCF-only checkpoints must not calibrate the blended frame.
-TARGET_FRAME_CHECKPOINT_MATERIALIZER_VERSION = 9
+# 10: #557 preserves the staged retirement-distribution surface through
+# release materialization; pre-#557 checkpoints can carry QRF-refitted leaves
+# and must not serve the preserved-surface baseline.
+TARGET_FRAME_CHECKPOINT_MATERIALIZER_VERSION = 10
 DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE = 5_000
 DEFAULT_L0_REFIT_LAMBDA_SHARE = 0.8
 DEFAULT_US_FISCAL_CALIBRATION_EPOCHS = 1_500
@@ -1152,7 +1160,7 @@ def _parse_args() -> argparse.Namespace:
             "stores expensive per-household target materialization artifacts "
             "such as JCT reform income-tax vectors and is content-addressed by "
             "base H5, policyengine-us version, period, geography crosswalk, and "
-            "reform (see #217)."
+            "the target-frame materializer identity and reform (see #217/#557)."
         ),
     )
     parser.add_argument(
@@ -1433,6 +1441,18 @@ def _reform_vector_cache_context(context: Mapping[str, object]) -> dict[str, obj
     that omits the crosswalk sha is distinct from one that sets it); absent keys
     are simply not part of the identity.
     """
+    # The materializer identity is the anti-mixing key (PR #557): a producer
+    # that cannot state which target-frame semantics built its vectors must
+    # not share the cache. Presence is required; an explicit None is a valid
+    # declaration for non-release producers (the scorers) and is
+    # identity-distinct from every release digest.
+    if "target_frame_materializer_identity_sha256" not in context:
+        raise ValueError(
+            "Reform-vector cache context must declare "
+            "target_frame_materializer_identity_sha256 (explicit None for "
+            "non-release producers); silently omitting it would let vectors "
+            "from different target-frame semantics mix (PR #557 review)."
+        )
     return {
         key: context[key] for key in REFORM_VECTOR_CACHE_CONTEXT_KEYS if key in context
     }
@@ -1452,10 +1472,9 @@ def _target_materialization_cache_identity(
         "n_households": int(n_households),
         "reform_measure": str(reform_spec.measure),
         "neutralized_variable": str(reform_spec.neutralized_variable),
-        # #217: reform-vector identity uses only the inputs that determine the
-        # per-household estimate. Build commit / seed / registry version are
-        # intentionally excluded so calibration-only or commit-only reruns reuse
-        # the cache; base H5 / PE-US version / period / geography still invalidate.
+        # #217/#557: build commit remains intentionally excluded, while the
+        # target-frame materializer digest binds seed, registry, staged-frame
+        # semantics, and support selection to the absolute reform vector.
         "context": dict(sorted(_reform_vector_cache_context(context).items())),
     }
 
@@ -8027,22 +8046,29 @@ def main() -> None:
         time_period=PERIOD,
     )
     retirement_distributions_gate = us_retirement_distributions_signal_gate(base_frame)
+    early_terminal_gate_failures: list[str] = []
     if not retirement_distributions_gate.passed:
-        if telemetry is not None:
-            telemetry.stage(
-                "retirement_distribution_inputs_gate",
-                status="failed",
-                message="Retirement-distribution signal gate failed.",
-                failures=list(retirement_distributions_gate.failures),
-                force_upload=True,
-            )
-        raise RuntimeError(
-            "Release gates failed: "
-            + "; ".join(
-                "Retirement-distribution signal failed: " + failure
-                for failure in retirement_distributions_gate.failures
-            )
+        # A selected support is consume-only at this boundary. Preserve the
+        # gate's batched missing/degenerate leaf evidence and carry it to the
+        # #548 terminal accumulator instead of refitting or raising early.
+        early_terminal_gate_failures.extend(
+            "Retirement-distribution signal failed: " + failure
+            for failure in retirement_distributions_gate.failures
         )
+        try:
+            if telemetry is not None:
+                telemetry.stage(
+                    "retirement_distribution_inputs_gate",
+                    status="failed",
+                    message="Retirement-distribution signal gate failed.",
+                    failures=list(retirement_distributions_gate.failures),
+                    force_upload=True,
+                )
+        except Exception as error:
+            early_terminal_gate_failures.append(
+                "Retirement-distribution gate failure telemetry crashed; "
+                f"recorded instead of masking the failure: {error}"
+            )
     if telemetry is not None:
         telemetry.stage(
             "eligibility_inputs",
@@ -8721,40 +8747,89 @@ def main() -> None:
         and not input_mass_reference_gate.passed
         and not args.allow_input_mass_drift
     ):
-        if telemetry is not None:
-            telemetry.stage(
-                "input_mass_reference_gate",
-                status="failed",
-                message="Base-frame input mass parity gate failed.",
-                failures=list(input_mass_reference_gate.failures),
-                force_upload=True,
-            )
-        raise RuntimeError(
-            "Release gates failed: "
-            + "; ".join(
-                f"Input mass parity failed: {failure}"
-                for failure in input_mass_reference_gate.failures
+        # Degraded pre-solve (PR #557 rounds 2-3): when the retirement
+        # boundary already failed, this raise would supersede the specific
+        # missing-leaf diagnosis. The gate object already rides
+        # _release_gate_failures into the single terminal batch, so the
+        # degraded branch simply does NOT raise — no duplicate append — and
+        # its telemetry is guarded so a reporting crash cannot mask the
+        # pending diagnosis (the #547 secure-before-report pattern). A green
+        # run keeps today's fail-fast raise.
+        if early_terminal_gate_failures:
+            try:
+                if telemetry is not None:
+                    telemetry.stage(
+                        "input_mass_reference_gate",
+                        status="failed",
+                        message="Base-frame input mass parity gate failed.",
+                        failures=list(input_mass_reference_gate.failures),
+                        force_upload=True,
+                    )
+            except Exception as error:
+                early_terminal_gate_failures.append(
+                    "Input-mass gate failure telemetry crashed in degraded "
+                    f"mode; recorded instead of masking the diagnosis: {error}"
+                )
+        else:
+            if telemetry is not None:
+                telemetry.stage(
+                    "input_mass_reference_gate",
+                    status="failed",
+                    message="Base-frame input mass parity gate failed.",
+                    failures=list(input_mass_reference_gate.failures),
+                    force_upload=True,
+                )
+            raise RuntimeError(
+                "Release gates failed: "
+                + "; ".join(
+                    f"Input mass parity failed: {failure}"
+                    for failure in input_mass_reference_gate.failures
+                )
             )
-        )
     degenerate_input_gate = _degenerate_input_signal_gate(
         base_frame, PolicyEngineUSEngine()
     )
     if not degenerate_input_gate.passed:
-        if telemetry is not None:
-            telemetry.stage(
-                "degenerate_input_gate",
-                status="failed",
-                message="Degenerate input signal gate failed.",
-                failures=list(degenerate_input_gate.failures),
-                force_upload=True,
-            )
-        raise RuntimeError(
-            "Release gates failed: "
-            + "; ".join(
-                f"Degenerate input signal failed: {failure}"
-                for failure in degenerate_input_gate.failures
+        # Same degraded contract as the input-mass gate above: the gate
+        # object rides _release_gate_failures to the batch; no raise, no
+        # duplicate append, guarded telemetry.
+        if early_terminal_gate_failures:
+            try:
+                if telemetry is not None:
+                    telemetry.stage(
+                        "degenerate_input_gate",
+                        status="failed",
+                        message="Degenerate input signal gate failed.",
+                        failures=list(degenerate_input_gate.failures),
+                        force_upload=True,
+                    )
+            except Exception as error:
+                early_terminal_gate_failures.append(
+                    "Degenerate-input gate failure telemetry crashed in "
+                    f"degraded mode; recorded instead of masking the "
+                    f"diagnosis: {error}"
+                )
+        else:
+            if telemetry is not None:
+                telemetry.stage(
+                    "degenerate_input_gate",
+                    status="failed",
+                    message="Degenerate input signal gate failed.",
+                    failures=list(degenerate_input_gate.failures),
+                    force_upload=True,
+                )
+            raise RuntimeError(
+                "Release gates failed: "
+                + "; ".join(
+                    f"Degenerate input signal failed: {failure}"
+                    for failure in degenerate_input_gate.failures
+                )
             )
-        )
+    # No combined pre-solve raise: a degraded run continues through the
+    # solve so calibration_diagnostics.json and the single terminal batch
+    # exist (the #547/#548 evidence contract — compute is cheaper than a
+    # destroyed failure record). The two gates above keep fail-fast raises
+    # on otherwise-green runs only.
     if telemetry is not None:
         telemetry.stage(
             "ecps_parity_gate",
@@ -8762,21 +8837,44 @@ def main() -> None:
         )
     ecps_parity_gate = _ecps_parity_gate(base_frame)
     if not ecps_parity_gate.passed and not args.allow_ecps_parity_gaps:
-        if telemetry is not None:
-            telemetry.stage(
-                "ecps_parity_gate",
-                status="failed",
-                message="eCPS parity gate failed.",
-                failures=list(ecps_parity_gate.failures),
-                force_upload=True,
-            )
-        raise RuntimeError(
-            "Release gates failed: "
-            + "; ".join(
-                f"eCPS parity failed: {failure}"
-                for failure in ecps_parity_gate.failures
+        # Same degraded contract as the input-mass/degenerate gates above
+        # (PR #557 round 3 finding 1): the pinned parity reference requires
+        # the retirement leaves, so a missing-leaf frame fails HERE too and
+        # an unconditional raise would supersede the retirement diagnosis
+        # before the solve. The gate object rides _release_gate_failures
+        # (as enforced_ecps_parity_gate) into the terminal batch; degraded
+        # runs continue, green runs keep the fail-fast raise.
+        if early_terminal_gate_failures:
+            try:
+                if telemetry is not None:
+                    telemetry.stage(
+                        "ecps_parity_gate",
+                        status="failed",
+                        message="eCPS parity gate failed.",
+                        failures=list(ecps_parity_gate.failures),
+                        force_upload=True,
+                    )
+            except Exception as error:
+                early_terminal_gate_failures.append(
+                    "eCPS parity gate failure telemetry crashed in degraded "
+                    f"mode; recorded instead of masking the diagnosis: {error}"
+                )
+        else:
+            if telemetry is not None:
+                telemetry.stage(
+                    "ecps_parity_gate",
+                    status="failed",
+                    message="eCPS parity gate failed.",
+                    failures=list(ecps_parity_gate.failures),
+                    force_upload=True,
+                )
+            raise RuntimeError(
+                "Release gates failed: "
+                + "; ".join(
+                    f"eCPS parity failed: {failure}"
+                    for failure in ecps_parity_gate.failures
+                )
             )
-        )
     if telemetry is not None:
         telemetry.stage("target_compilation", message="Materializing target frame.")
     target_compilation_started = time.perf_counter()
@@ -8816,6 +8914,9 @@ def main() -> None:
             else ssi_take_up_prior_basis.source_sha256
         ),
     )
+    target_frame_materializer_identity_sha256 = _target_frame_checkpoint_digest(
+        target_frame_checkpoint_identity
+    )
     target_frame, registry, compilation = _load_or_materialize_target_frame(
         base_frame,
         target_specs,
@@ -8846,6 +8947,12 @@ def main() -> None:
             "selection_identities_sha256": (
                 None if selection_source is None else selection_source.identities_sha256
             ),
+            # Reform caches store absolute income-tax vectors, which are
+            # subtracted from a freshly materialized baseline. Bind both sides
+            # to one complete materializer identity (populace#557).
+            "target_frame_materializer_identity_sha256": (
+                target_frame_materializer_identity_sha256
+            ),
         },
         gate_congressional_district_targets=args.gate_congressional_district_targets,
     )
@@ -9044,7 +9151,6 @@ def main() -> None:
     # diagnostics and skipped every other gate group (populace#547). A law
     # violation additionally quarantines SSI-dependent evaluations below.
     ssi_law_degraded = not final_ssi_take_up_gate.passed
-    early_terminal_gate_failures: list[str] = []
     if not final_ssi_take_up_gate.passed:
         # Failures enter the list BEFORE any reporting: the telemetry stage
         # performs local writes and must not be able to mask the gate
diff --git a/tools/build_us_puf_support_base.py b/tools/build_us_puf_support_base.py
index ba2c6b78..1ac65361 100644
--- a/tools/build_us_puf_support_base.py
+++ b/tools/build_us_puf_support_base.py
@@ -1238,6 +1238,10 @@ def _run_all(
         imputed,
         seed=args.seed,
         time_period=args.target_year,
+        # This is the one ownership boundary where the copied ASEC leaves must
+        # be replaced on the newly created PUF support. Downstream consumers
+        # preserve this completed draw even after selecting a narrower support.
+        force_puf_imputation=True,
     )
     retirement_distributions_gate = us_retirement_distributions_signal_gate(imputed)
     if not retirement_distributions_gate.passed:
@@ -2193,6 +2197,9 @@ def _post_qrf_frame_stage(
             frame,
             seed=args.seed,
             time_period=args.target_year,
+            # Resuming this named base-builder stage is equivalent to crossing
+            # the live post-clone ownership boundary above.
+            force_puf_imputation=True,
         )
         signals["retirement_distributions_signal"] = _checked_gate_payload(
             us_retirement_distributions_signal_gate(frame),
diff --git a/tools/score_us_fiscal_targets.py b/tools/score_us_fiscal_targets.py
index 55e63ec1..0a8651da 100644
--- a/tools/score_us_fiscal_targets.py
+++ b/tools/score_us_fiscal_targets.py
@@ -489,6 +489,10 @@ def score_frame(
             "seed": 0,
             "target_period": release.PERIOD,
             "target_registry_version": target_registry.version,
+            # Scorer vectors declare no release materializer identity: the
+            # explicit None is identity-distinct from every release digest,
+            # so scorer and release vectors can never mix (PR #557).
+            "target_frame_materializer_identity_sha256": None,
             "congressional_district_vintage_crosswalk_sha256": (
                 congressional_district_vintage_crosswalk_metadata or {}
             ).get("sha256"),
diff --git a/tools/score_us_state_files.py b/tools/score_us_state_files.py
index abba5315..42d874e4 100644
--- a/tools/score_us_state_files.py
+++ b/tools/score_us_state_files.py
@@ -385,6 +385,10 @@ def score_state_files(
             "target_period": release.PERIOD,
             "target_registry_version": target_registry.version,
             "state_file_collection": collection_sha256,
+            # Scorer vectors declare no release materializer identity:
+            # the explicit None is identity-distinct from every release
+            # digest, so scorer and release vectors can never mix (PR #557).
+            "target_frame_materializer_identity_sha256": None,
         }
         if target_materialization_cache_dir is not None
         else None