diff --git a/changelog.d/1800.fixed.md b/changelog.d/1800.fixed.md
new file mode 100644
index 0000000000..e3b323c4c5
--- /dev/null
+++ b/changelog.d/1800.fixed.md
@@ -0,0 +1 @@
+Structural reform switches now follow each calculation period, including future Scottish Child Payment changes and active-year reported benefit inputs. Bare-year switch keys in reform dictionaries and Scenario `parameter_changes` apply only to the named year. Original formulas and genuine supplied inputs apply outside each reform interval.
diff --git a/policyengine_uk/reforms/conservatives/household_based_hitc.py b/policyengine_uk/reforms/conservatives/household_based_hitc.py
index 05295e581d..826799841e 100644
--- a/policyengine_uk/reforms/conservatives/household_based_hitc.py
+++ b/policyengine_uk/reforms/conservatives/household_based_hitc.py
@@ -37,3 +37,25 @@ def create_household_based_hitc_reform(parameters, period):
         return make_cb_hitc_household_based
     else:
         return None
+
+
+def create_period_aware_household_hitc_reform(parameters, period):
+    from policyengine_uk.reforms.reforms import (
+        parameter_ever_true,
+        apply_period_aware_reform,
+    )
+
+    path = "gov.contrib.conservatives.cb_hitc_household"
+    if not parameter_ever_true(parameters, path):
+        return None
+
+    def active(root, year):
+        return root(year).gov.contrib.conservatives.cb_hitc_household
+
+    class reform(Reform):
+        def apply(self):
+            apply_period_aware_reform(
+                self, make_cb_hitc_household_based, ("CB_HITC",), active
+            )
+
+    return reform
diff --git a/policyengine_uk/reforms/cps/marriage_tax_reforms.py b/policyengine_uk/reforms/cps/marriage_tax_reforms.py
index 16ddaa7960..4645e1e522 100644
--- a/policyengine_uk/reforms/cps/marriage_tax_reforms.py
+++ b/policyengine_uk/reforms/cps/marriage_tax_reforms.py
@@ -7,6 +7,7 @@
 def create_expanded_ma_reform(
     max_child_age: Optional[int] = None,
     child_education_levels: Optional[List[str]] = None,
+    _parameter_driven: bool = False,
 ) -> Reform:
     class meets_expanded_ma_conditions(Variable):
         label = "Qualifies for an expanded Marriage Allowance"
@@ -19,9 +20,15 @@ def formula(person, period):
             # couple) under the age limit, or meeting the education condition.
             benunit = person.benunit
             is_couple_child = ~person("is_claimant_or_partner", period)
-            if max_child_age is not None:
+            child_age_limit = max_child_age
+            if _parameter_driven:
+                cap = person.simulation.tax_benefit_system.parameters(
+                    period
+                ).gov.contrib.cps.marriage_tax_reforms.expanded_ma.max_child_age
+                child_age_limit = cap if cap > 0 else None
+            if child_age_limit is not None:
                 child_meets_age_condition = is_couple_child & (
-                    person("age", period) < max_child_age
+                    person("age", period) < child_age_limit
                 )
                 return benunit.any(child_meets_age_condition)
             if child_education_levels is not None:
@@ -105,6 +112,7 @@ def apply(self):
 def create_marriage_neutral_income_tax_reform(
     max_child_age: Optional[int] = None,
     child_education_levels: Optional[List[str]] = None,
+    _parameter_driven: bool = False,
 ) -> Reform:
     class meets_ma_neutral_tax_conditions(Variable):
         label = "Qualifies for an expanded Marriage Allowance"
@@ -117,9 +125,15 @@ def formula(person, period):
             # couple) under the age limit, or meeting the education condition.
             benunit = person.benunit
             is_couple_child = ~person("is_claimant_or_partner", period)
-            if max_child_age is not None:
+            child_age_limit = max_child_age
+            if _parameter_driven:
+                cap = person.simulation.tax_benefit_system.parameters(
+                    period
+                ).gov.contrib.cps.marriage_tax_reforms.marriage_neutral_it.max_child_age
+                child_age_limit = cap if cap > 0 else None
+            if child_age_limit is not None:
                 child_meets_age_condition = is_couple_child & (
-                    person("age", period) < max_child_age
+                    person("age", period) < child_age_limit
                 )
                 return benunit.any(child_meets_age_condition)
             if child_education_levels is not None:
@@ -259,3 +273,71 @@ def create_marriage_tax_reform(parameters, period):
 # applies to every married couple regardless of child presence.
 expanded_ma_reform = create_expanded_ma_reform()
 marriage_neutral_it_reform = create_marriage_neutral_income_tax_reform()
+
+
+def create_period_aware_marriage_tax_reform(parameters, period):
+    """Install only families with a real activation in their finite history.
+
+    This constructor dispatch differs from the public period-only factory:
+    an available route still chooses its policy separately in each year.
+    """
+    from policyengine_uk.reforms.reforms import (
+        finite_parameter_activation,
+        apply_period_aware_reform,
+    )
+
+    prefix = "gov.contrib.cps.marriage_tax_reforms."
+    paths = (
+        "expanded_ma.remove_income_condition",
+        "expanded_ma.ma_rate",
+        "expanded_ma.max_child_age",
+        "marriage_neutral_it.neutralise_income_tax",
+        "marriage_neutral_it.max_child_age",
+    )
+    current = [parameters.get_child(prefix + path) for path in paths]
+    baseline = [parameters.baseline.get_child(prefix + path) for path in paths]
+    expanded = finite_parameter_activation(
+        current + baseline, lambda values: bool(values[0]) or values[1] != values[6]
+    )
+    neutral = finite_parameter_activation(
+        current + baseline, lambda values: bool(values[3])
+    )
+    if not expanded and not neutral:
+        return None
+
+    def expanded_active(root, year):
+        p = root(year).gov.contrib.cps.marriage_tax_reforms.expanded_ma
+        return (
+            p.remove_income_condition
+            or p.ma_rate
+            != root.baseline(
+                year
+            ).gov.contrib.cps.marriage_tax_reforms.expanded_ma.ma_rate
+        )
+
+    def neutral_active(root, year):
+        return root(
+            year
+        ).gov.contrib.cps.marriage_tax_reforms.marriage_neutral_it.neutralise_income_tax
+
+    class reform(Reform):
+        def apply(self):
+            if expanded:
+                apply_period_aware_reform(
+                    self,
+                    create_expanded_ma_reform(_parameter_driven=True),
+                    (
+                        "meets_marriage_allowance_income_conditions",
+                        "marriage_allowance",
+                    ),
+                    expanded_active,
+                )
+            if neutral:
+                apply_period_aware_reform(
+                    self,
+                    create_marriage_neutral_income_tax_reform(_parameter_driven=True),
+                    ("net_income",),
+                    neutral_active,
+                )
+
+    return reform
diff --git a/policyengine_uk/reforms/policyengine/disable_simulated_benefits.py b/policyengine_uk/reforms/policyengine/disable_simulated_benefits.py
index 6c4ef3aeec..de2956b4c9 100644
--- a/policyengine_uk/reforms/policyengine/disable_simulated_benefits.py
+++ b/policyengine_uk/reforms/policyengine/disable_simulated_benefits.py
@@ -84,12 +84,26 @@
 YEARS_IN_FUTURE = 10
 
 
-def disable_simulated_benefits(parameters, period):
-    if parameters(period).gov.contrib.policyengine.disable_simulated_benefits:
+def disable_simulated_benefits(parameters, period, _period_aware=False):
+    if _period_aware:
+        from policyengine_uk.reforms.reforms import parameter_ever_true
+
+        available = parameter_ever_true(
+            parameters, "gov.contrib.policyengine.disable_simulated_benefits"
+        )
+    else:
+        available = parameters(
+            period
+        ).gov.contrib.policyengine.disable_simulated_benefits
+    if available:
 
         class DisableSimulatedBenefits(Reform):
             """Use the dataset year's reported benefit amounts in every year.
 
+            Parameter-driven construction stores only active calculation years
+            and temporarily removes its owned inputs for an inactive query.
+            Explicit application keeps this reform's existing ten-year mode.
+
             Each listed benefit is set, in the dataset year and the nine years
             after it, to its amount reported in the dataset year. The amount
             is not uprated, while most of the *_reported variables are.
@@ -144,6 +158,15 @@ def apply(self):
                 system = simulation.tax_benefit_system
                 time_period = int(str(simulation.dataset.time_period)[:4])
                 years = range(time_period, time_period + YEARS_IN_FUTURE)
+                if _period_aware:
+                    years = tuple(
+                        year
+                        for year in years
+                        if system.parameters(
+                            year
+                        ).gov.contrib.policyengine.disable_simulated_benefits
+                    )
+                prepared = {}
 
                 def reported(variable, members=None):
                     """A benefit's reported amount in the dataset year, for
@@ -179,8 +202,7 @@ def reported(variable, members=None):
                         entity = system.variables[variable].entity.key
                         count = simulation.populations[entity].count
                         values[variable] = np.zeros(count, dtype=bool)
-                    for target, value in values.items():
-                        simulation.set_input(target, year, value)
+                    prepared[year] = values
 
                 # Where a household lives is read in each year, as the model
                 # reads it (country is a YEAR variable). The reported amounts
@@ -215,15 +237,11 @@ def reported(variable, members=None):
                     paid_as_csp = households.project(in_scotland[year]) & (
                         year >= CARER_SUPPORT_PAYMENT_FROM
                     )
-                    simulation.set_input(
-                        "carers_allowance",
-                        year,
-                        np.where(paid_as_csp, 0, reported_care),
+                    prepared[year]["carers_allowance"] = np.where(
+                        paid_as_csp, 0, reported_care
                     )
-                    simulation.set_input(
-                        "carer_support_payment",
-                        year,
-                        np.where(paid_as_csp, reported_care, 0),
+                    prepared[year]["carer_support_payment"] = np.where(
+                        paid_as_csp, reported_care, 0
                     )
 
                 # The survey reports one winter heating payment: FRS benefit
@@ -250,15 +268,22 @@ def reported(variable, members=None):
                 for year in years:
                     scheme = system.parameters(year).gov.social_security_scotland
                     paid_as_pawhp = in_scotland[year] & scheme.pawhp.active
-                    simulation.set_input(
-                        "winter_fuel_allowance",
-                        year,
-                        np.where(paid_as_pawhp, 0, reported_winter_heating),
+                    prepared[year]["winter_fuel_allowance"] = np.where(
+                        paid_as_pawhp, 0, reported_winter_heating
                     )
-                    simulation.set_input(
-                        "pawhp",
-                        year,
-                        np.where(paid_as_pawhp, reported_winter_heating, 0),
+                    prepared[year]["pawhp"] = np.where(
+                        paid_as_pawhp, reported_winter_heating, 0
+                    )
+
+                # Read all source reports and current-year mappings before
+                # storing anything that could become a carried input.
+                if _period_aware:
+                    simulation._install_reported_benefit_inputs(
+                        prepared, time_period, YEARS_IN_FUTURE
                     )
+                else:
+                    for year, values in prepared.items():
+                        for target, value in values.items():
+                            simulation.set_input(target, year, value)
 
         return DisableSimulatedBenefits
diff --git a/policyengine_uk/reforms/reforms.py b/policyengine_uk/reforms/reforms.py
index be00af9a36..9af5e9a5eb 100644
--- a/policyengine_uk/reforms/reforms.py
+++ b/policyengine_uk/reforms/reforms.py
@@ -1,10 +1,14 @@
-from .cps import create_marriage_tax_reform
-from .conservatives import create_household_based_hitc_reform
+from .cps.marriage_tax_reforms import create_period_aware_marriage_tax_reform
+from .conservatives.household_based_hitc import (
+    create_period_aware_household_hitc_reform,
+)
 from .policyengine import (
     disable_simulated_benefits,
 )
 from .policyengine.adjust_budgets import adjust_budgets
-from .scotland import create_scottish_child_payment_reform
+from .scotland.scottish_child_payment_reform import (
+    create_period_aware_scottish_child_payment_reform,
+)
 from policyengine_core.model_api import *
 from policyengine_core import periods
 
@@ -12,11 +16,11 @@
 def create_structural_reforms_from_parameters(parameters, period):
     period = periods.period(period)
     reforms = [
-        create_marriage_tax_reform(parameters, period),
-        create_household_based_hitc_reform(parameters, period),
-        disable_simulated_benefits(parameters, period),
+        create_period_aware_marriage_tax_reform(parameters, period),
+        create_period_aware_household_hitc_reform(parameters, period),
+        disable_simulated_benefits(parameters, period, _period_aware=True),
         adjust_budgets(parameters, period),
-        create_scottish_child_payment_reform(parameters, period),
+        create_period_aware_scottish_child_payment_reform(parameters, period),
     ]
     reforms = tuple(filter(lambda x: x is not None, reforms))
 
@@ -26,3 +30,86 @@ def apply(self):
                 reform.apply(self)
 
     return combined_reform
+
+
+def finite_parameter_activation(leaves, predicate):
+    """Inspect actual effective points, including raw early-year sentinels.
+
+    Sampling starts only where every participating leaf has history. A missing
+    or null value there is unsupported, rather than an implicit off switch.
+    """
+    histories = [leaf.values_list for leaf in leaves]
+    if any(not history for history in histories):
+        raise ValueError("Structural reform parameters require effective history")
+    first_covered = max(
+        min(value.instant_str for value in history) for history in histories
+    )
+    points = sorted({value.instant_str for history in histories for value in history})
+    available = False
+    for point in points:
+        if point < first_covered:
+            continue
+        # Parameter._get_at_instant accepts the stored string directly. The
+        # public callable would try to turn year 0000 into a datetime.date.
+        values = [leaf._get_at_instant(point) for leaf in leaves]
+        if any(value is None for value in values):
+            raise ValueError(f"Structural reform parameter has no value at {point}")
+        # Validate later points even after an earlier interval is active.
+        # An explicit null anywhere in covered history remains unsupported.
+        active_at_point = bool(predicate(values))
+        available = active_at_point or available
+    return available
+
+
+def parameter_ever_true(parameters, path):
+    return finite_parameter_activation(
+        [parameters.get_child(path)], lambda values: bool(values[0])
+    )
+
+
+def apply_period_aware_reform(system, reform, names, active):
+    """Route each replacement through the original dated Variable API.
+
+    The contributed and original definitions must share storage and masking
+    metadata. Formula execution, including adds/subtracts, stays in Core.
+    """
+    originals = {name: system.variables[name].clone() for name in names}
+    trial = system.clone()
+    reform.apply(trial)
+    metadata = (
+        "value_type",
+        "definition_period",
+        "default_value",
+        "defined_for",
+        "end",
+        "set_input",
+        "calculate_output",
+        "uprating",
+    )
+    for name, original in originals.items():
+        replacement = trial.variables[name]
+        if original.entity.key != replacement.entity.key or any(
+            getattr(original, key, None) != getattr(replacement, key, None)
+            for key in metadata
+        ):
+            raise ValueError(f"Structural reform changes routing metadata for {name}")
+    reform.apply(system)
+    for name, original in originals.items():
+        replacement = system.variables[name].clone()
+        routed = replacement.clone()
+
+        def route(original, replacement):
+            def formula(population, period, parameters):
+                simulation = population.simulation
+                root = simulation.tax_benefit_system.parameters
+                selected = replacement if active(root, period) else original
+                return simulation._run_formula(selected, population, period)
+
+            return formula
+
+        starts = set(original.formulas) | set(replacement.formulas)
+        if not starts:
+            raise ValueError(f"Structural reform has no formula route for {name}")
+        routed.formulas.clear()
+        routed.formulas[min(starts)] = route(original, replacement)
+        system.variables[name] = routed
diff --git a/policyengine_uk/reforms/scotland/scottish_child_payment_reform.py b/policyengine_uk/reforms/scotland/scottish_child_payment_reform.py
index 44910e5850..49977b4cdd 100644
--- a/policyengine_uk/reforms/scotland/scottish_child_payment_reform.py
+++ b/policyengine_uk/reforms/scotland/scottish_child_payment_reform.py
@@ -87,3 +87,28 @@ def create_scottish_child_payment_reform(parameters, period, bypass: bool = Fals
 
 
 scottish_child_payment_reform = create_scottish_child_payment_baby_bonus_reform()
+
+
+def create_period_aware_scottish_child_payment_reform(parameters, period):
+    from policyengine_uk.reforms.reforms import (
+        parameter_ever_true,
+        apply_period_aware_reform,
+    )
+
+    path = "gov.contrib.scotland.scottish_child_payment.in_effect"
+    if not parameter_ever_true(parameters, path):
+        return None
+
+    def active(root, year):
+        return root(year).gov.contrib.scotland.scottish_child_payment.in_effect
+
+    class reform(Reform):
+        def apply(self):
+            apply_period_aware_reform(
+                self,
+                create_scottish_child_payment_baby_bonus_reform(),
+                ("scottish_child_payment",),
+                active,
+            )
+
+    return reform
diff --git a/policyengine_uk/simulation.py b/policyengine_uk/simulation.py
index 7e87c950d0..856b648d1a 100644
--- a/policyengine_uk/simulation.py
+++ b/policyengine_uk/simulation.py
@@ -599,8 +599,519 @@ def _user_input_keys(self) -> None:
         except KeyError:
             raise AttributeError("_user_input_keys") from None
 
+    def _invalidate_all_caches(self) -> None:
+        """Rebind continuously owned reports across Core's own input replay.
+
+        A caller's equal write is already a new input token and is pruned
+        before this boundary. Only this successful synchronous Core replay
+        may replace the tokens of entries that remain continuously owned.
+        """
+        self._prune_reported_benefit_inputs()
+        ledger = getattr(self, "_reported_benefit_inputs", {})
+        if not ledger:
+            return super()._invalidate_all_caches()
+        from policyengine_core.simulations.simulation import _value_signature
+
+        snapshots = []
+        for key, entry in ledger.items():
+            holder = self.get_holder(entry["variable"])
+            token = self._reported_input_token(holder, entry["branch"], entry["period"])
+            if token is not entry["token"]:
+                self._reported_overlay_refusal(
+                    "Reported input changed before Core replay"
+                )
+            tiers = self._reported_genuine_tiers(
+                holder, entry["branch"], entry["period"]
+            )
+            # Core retains memory first, otherwise disk, rather than treating
+            # a computed memory shadow as the genuine disk input.
+            location = (
+                "memory" if "memory" in tiers else "disk" if "disk" in tiers else None
+            )
+            signature = (
+                _value_signature(tiers[location]) if location is not None else None
+            )
+            if location is not None and signature is None:
+                self._reported_overlay_refusal(
+                    "Unsupported owned input before Core replay"
+                )
+            snapshots.append(
+                (
+                    key,
+                    entry,
+                    token,
+                    entry["visible"],
+                    entry["genuine"],
+                    entry["synthetic"],
+                    location,
+                    signature,
+                )
+            )
+
+        try:
+            # The pinned Core boundary calls storage replay directly, with
+            # no user formulas or input-change events. Unexpected events are
+            # ambiguity, not evidence that a model report became genuine.
+            result = self._reported_owned_input_events(
+                [], super()._invalidate_all_caches
+            )
+            if getattr(self, "_reported_benefit_inputs", None) is not ledger:
+                self._reported_overlay_refusal(
+                    "Reported ledger replaced during Core replay"
+                )
+            rebound = []
+            for (
+                key,
+                entry,
+                old_token,
+                visible,
+                genuine,
+                synthetic,
+                location,
+                signature,
+            ) in snapshots:
+                if (
+                    ledger.get(key) is not entry
+                    or entry["token"] is not old_token
+                    or entry["visible"] is not visible
+                    or entry["genuine"] is not genuine
+                    or entry["synthetic"] is not synthetic
+                ):
+                    self._reported_overlay_refusal(
+                        "Reported ownership changed during Core replay"
+                    )
+                holder = self.get_holder(entry["variable"])
+                branch, period = entry["branch"], entry["period"]
+                token = self._reported_input_token(holder, branch, period)
+                actual = self._reported_genuine_tiers(holder, branch, period)
+                if location is None:
+                    if token is not None or actual or key in self._user_input_keys:
+                        self._reported_overlay_refusal(
+                            "Absent reported input appeared during Core replay"
+                        )
+                elif (
+                    token is old_token
+                    or type(token) is not tuple
+                    or len(token) != 2
+                    or type(token[0]) is not type(period)
+                    or type(token[1]) is not frozenset
+                    or token[0] != period
+                    or token[1] != frozenset((location,))
+                    or set(actual) != {location}
+                    or _value_signature(actual[location]) != signature
+                ):
+                    self._reported_overlay_refusal(
+                        "Owned input did not match exact Core replay"
+                    )
+                rebound.append((entry, token))
+            # Do not leave a partially rebound ledger if a later slot fails.
+            for entry, token in rebound:
+                entry["token"] = token
+            return result
+        except BaseException as error:
+            try:
+                self._reported_overlay_refusal(
+                    "Core reported-input replay failed or was ambiguous"
+                )
+            except RuntimeError as refusal:
+                raise refusal from error
+
+    def _reported_overlay_refusal(self, reason):
+        """Latch ambiguity; it is never proof that a model input is genuine."""
+        self._reported_benefit_observer_error = reason
+        raise RuntimeError(
+            f"Cannot safely synchronize reported benefit inputs: {reason}"
+        )
+
+    def _check_reported_overlay_observer(self):
+        reason = getattr(self, "_reported_benefit_observer_error", None)
+        if reason is not None:
+            raise RuntimeError(
+                f"Cannot safely synchronize reported benefit inputs: {reason}"
+            )
+
+    def _reported_observer_state(self, observer):
+        frame = observer["frame"]
+        events = observer["events"]
+        cursor = observer["cursor"]
+        from policyengine_core.simulations.simulation import _calculation_frames
+
+        if (
+            getattr(frame, "simulation", None) is not self
+            or getattr(frame, "changes", None) is not events
+            or type(events) is not list
+            or not any(current is frame for current in _calculation_frames.get())
+        ):
+            self._reported_overlay_refusal(
+                "Core calculation event-list identity changed"
+            )
+        if type(cursor) is not int or not 0 <= cursor <= len(events):
+            self._reported_overlay_refusal("Core calculation event cursor is invalid")
+        return events, cursor
+
+    def _drain_reported_benefit_events(self):
+        self._check_reported_overlay_observer()
+        ledger = getattr(self, "_reported_benefit_inputs", {})
+        for observer in tuple(getattr(self, "_reported_benefit_observers", ())):
+            events, cursor = self._reported_observer_state(observer)
+            for change in events[cursor:]:
+                if type(change) is not tuple or len(change) != 6:
+                    self._reported_overlay_refusal("Unknown Core input-event schema")
+                operation, simulation, branch, variable, period, signature = change
+                if (
+                    type(operation) is not str
+                    or operation not in ("set", "delete")
+                    or not isinstance(simulation, CoreSimulation)
+                    or type(branch) is not str
+                    or type(variable) is not str
+                    or (period is not None and not isinstance(period, Period))
+                ):
+                    self._reported_overlay_refusal("Unknown Core input-event identity")
+                known_signature = (
+                    signature is None
+                    if operation == "delete"
+                    else (
+                        type(signature) is tuple
+                        and len(signature) == 4
+                        and type(signature[0]) is str
+                        and type(signature[1]) is tuple
+                        and all(
+                            type(size) is int and size >= 0 for size in signature[1]
+                        )
+                        and (
+                            signature[2] is None
+                            or isinstance(signature[2], type)
+                            and issubclass(signature[2], CoreEnum)
+                        )
+                        and type(signature[3]) is bytes
+                    )
+                )
+                if not known_signature:
+                    self._reported_overlay_refusal(
+                        "Unknown Core input-event value signature"
+                    )
+                if simulation is self and operation == "delete":
+                    for key, entry in list(ledger.items()):
+                        if (
+                            entry["variable"] == variable
+                            and entry["branch"] == branch
+                            and (period is None or period.contains(entry["period"]))
+                        ):
+                            # This includes a real no-op deletion while the
+                            # displaced physical input is genuinely absent.
+                            del ledger[key]
+            observer["cursor"] = len(events)
+
+    def _reported_owned_input_events(self, expected, operation):
+        """Consume only the exact suffix made by one synchronous own block.
+
+        The holder/storage operations below run no user formulas. An extra
+        event or an attempt-list replacement is ambiguity, not permission to
+        adopt a synthetic value as a genuine supplied input.
+        """
+        self._drain_reported_benefit_events()
+        marks = [
+            (observer, *self._reported_observer_state(observer))
+            for observer in tuple(getattr(self, "_reported_benefit_observers", ()))
+        ]
+        try:
+            result = operation()
+        except BaseException as error:
+            try:
+                self._reported_overlay_refusal(
+                    "Internal reported-input storage operation failed"
+                )
+            except RuntimeError as refusal:
+                raise refusal from error
+        for observer, events, cursor in marks:
+            actual, current = self._reported_observer_state(observer)
+            if actual is not events or current != cursor:
+                self._reported_overlay_refusal(
+                    "Internal reported-input event cursor changed"
+                )
+            suffix = events[cursor:]
+            if len(suffix) != len(expected):
+                self._reported_overlay_refusal(
+                    "Unexpected event during reported-input storage"
+                )
+            for change, wanted in zip(suffix, expected):
+                if (
+                    type(change) is not tuple
+                    or len(change) != 6
+                    or change[1] is not self
+                    or change != wanted
+                ):
+                    self._reported_overlay_refusal(
+                        "Unverified reported-input storage event"
+                    )
+            observer["cursor"] = len(events)
+        return result
+
+    def _calculate(self, variable_name, period=None):
+        """Observe actual attempt-local events only while reported inputs exist."""
+        self._check_reported_overlay_observer()
+        if not getattr(self, "_reported_benefit_inputs", None):
+            return self._calculate_with_bsp_routing(variable_name, period)
+        from policyengine_core.simulations.simulation import _calculation_frames
+
+        frames = _calculation_frames.get()
+        frame = next((frame for frame in frames if frame.simulation is self), None)
+        if frame is None:
+            self._reported_overlay_refusal(
+                "Reported calculation has no actual Core frame"
+            )
+        observers = getattr(self, "_reported_benefit_observers", None)
+        if observers is None:
+            observers = self._reported_benefit_observers = []
+        observer = next((item for item in observers if item["frame"] is frame), None)
+        if observer is None:
+            # A new outer calculation starts with a new empty attempt list.
+            # Never guess a cursor into history that predates observation.
+            if frame.changes:
+                self._reported_overlay_refusal(
+                    "Core input history predates reported observation"
+                )
+            if observers:
+                self._reported_overlay_refusal(
+                    "Reported observer escaped its Core attempt"
+                )
+            observer = {
+                "frame": frame,
+                "events": frame.changes,
+                "cursor": 0,
+                "users": 0,
+            }
+            observers.append(observer)
+        self._reported_observer_state(observer)
+        observer["users"] += 1
+        original_error = None
+        try:
+            return self._calculate_with_bsp_routing(variable_name, period)
+        except BaseException as error:
+            original_error = error
+            raise
+        finally:
+            try:
+                # This runs BEFORE Core removes or restarts the actual frame.
+                self._drain_reported_benefit_events()
+            except RuntimeError as refusal:
+                if original_error is not None:
+                    raise refusal from original_error
+                raise
+            finally:
+                observer["users"] -= 1
+                if observer["users"] == 0:
+                    observers.remove(observer)
+
+    @staticmethod
+    def _reported_input_token(holder, branch, period):
+        return getattr(holder, "_user_input_storage", {}).get((branch, str(period)))
+
+    def _reported_genuine_tiers(self, holder, branch, period):
+        """Copy only genuine physical inputs, never a derived memory shadow."""
+        key = (holder.variable.name, branch, period)
+        if key not in getattr(self, "_user_input_keys", ()):
+            return {}
+        result = {}
+        for location, storage in (
+            ("memory", holder._memory_storage),
+            ("disk", holder._disk_storage),
+        ):
+            if storage is not None and holder._stores_user_input(
+                period, branch, location
+            ):
+                value = storage.get(period, branch)
+                if value is not None:
+                    result[location] = value.copy()
+        return result
+
+    def _prune_reported_benefit_inputs(self):
+        self._drain_reported_benefit_events()
+        ledger = getattr(self, "_reported_benefit_inputs", {})
+        for key, entry in list(ledger.items()):
+            holder = self.get_holder(entry["variable"])
+            actual = self._reported_input_token(
+                holder, entry["branch"], entry["period"]
+            )
+            if actual is not entry["token"]:
+                # Every actual input write replaces the immutable record,
+                # even for equal values or the same ndarray object.
+                del ledger[key]
+            elif actual is not None and not holder._stores_user_input(
+                entry["period"], entry["branch"]
+            ):
+                del ledger[key]
+
+    def _restore_reported_physical_input(self, holder, branch, period, tiers):
+        """Use Core's storage/input bookkeeping with fresh write history.
+
+        This is an exact physical-slot restore, not a baseline calculation or
+        a carry-over implementation. Both genuine tiers can hold inputs.
+        """
+        from policyengine_core.data_storage.store_history import next_sequence_number
+        from policyengine_core.simulations.simulation import _note_input_change
+
+        for location, value in tiers.items():
+            storage = (
+                holder._memory_storage if location == "memory" else holder._disk_storage
+            )
+            if storage is None:
+                raise ValueError(
+                    f"Missing {location} storage while restoring {holder.variable.name}"
+                )
+            sequence = next_sequence_number()
+            restored = value.copy()
+            storage.put(
+                restored, period, branch, derived=False, sequence_number=sequence
+            )
+            holder._record_store(period, sequence)
+            holder._evict_fast_cache(period, branch)
+            holder._record_input_storage(period, branch, location, is_input=True)
+            self._user_input_keys.add((holder.variable.name, branch, period))
+            _note_input_change(
+                self, "set", holder.variable.name, period, branch, restored
+            )
+
+    def _replace_reported_benefit_input(self, entry, tiers):
+        from policyengine_core.simulations.simulation import _value_signature
+
+        self._prune_reported_benefit_inputs()
+        key = (entry["variable"], entry["branch"], entry["period"])
+        if getattr(self, "_reported_benefit_inputs", {}).get(key) is not entry:
+            return False
+        holder = self.get_holder(entry["variable"])
+        period, branch = entry["period"], entry["branch"]
+        # Holder.delete_arrays uses period containment in memory. Preserve
+        # every other genuine contained input in its original physical tier.
+        companions = []
+        for name, physical_branch, other in tuple(
+            getattr(self, "_user_input_keys", ())
+        ):
+            if (
+                name == entry["variable"]
+                and physical_branch == branch
+                and other != period
+                and period.contains(other)
+            ):
+                genuine = self._reported_genuine_tiers(holder, branch, other)
+                if genuine:
+                    companions.append((other, genuine))
+        expected = [("delete", self, branch, entry["variable"], period, None)]
+        for restored_period, restored_tiers in companions + [(period, tiers)]:
+            for value in restored_tiers.values():
+                signature = _value_signature(value)
+                if signature is None:
+                    self._reported_overlay_refusal(
+                        "Unsupported reported-input storage value"
+                    )
+                expected.append(
+                    ("set", self, branch, entry["variable"], restored_period, signature)
+                )
+
+        def replace():
+            holder.delete_arrays(period, branch)
+            for other, genuine in companions:
+                self._restore_reported_physical_input(holder, branch, other, genuine)
+            self._restore_reported_physical_input(holder, branch, period, tiers)
+
+        self._reported_owned_input_events(expected, replace)
+        entry["token"] = self._reported_input_token(holder, branch, period)
+        return True
+
+    def _sync_reported_benefit_inputs(self, enabled):
+        self._prune_reported_benefit_inputs()
+        changed = False
+        for entry in tuple(getattr(self, "_reported_benefit_inputs", {}).values()):
+            if entry["visible"] == enabled:
+                continue
+            if self._replace_reported_benefit_input(
+                entry, entry["synthetic"] if enabled else entry["genuine"]
+            ):
+                entry["visible"] = enabled
+                changed = True
+        self._reported_benefit_inputs_enabled = enabled
+        if changed:
+            # Preserve genuine inputs; invalidate only this simulation's
+            # computed values and let Core settle any stale calculation frame.
+            self.drop_computed_arrays()
+
+    @contextmanager
+    def _reported_benefit_context(self, period):
+        self._check_reported_overlay_observer()
+        if (
+            not getattr(self, "_reported_benefit_inputs", None)
+            or period.unit == "eternity"
+        ):
+            yield
+            return
+        year = period.this_year
+        start, count = self._reported_benefit_window
+        enabled = start <= year.start.year < start + count and bool(
+            self.tax_benefit_system.parameters(
+                year
+            ).gov.contrib.policyengine.disable_simulated_benefits
+        )
+        previous = getattr(self, "_reported_benefit_inputs_enabled", True)
+        stack = self._reported_benefit_requests
+        stack.append((year, enabled))
+        try:
+            self._sync_reported_benefit_inputs(enabled)
+            yield
+        finally:
+            stack.pop()
+            # A nested request returns to its caller's state. The outer call
+            # leaves the stored active inputs in place for replay/export.
+            # Pruning inside sync makes a genuine write during suspension win.
+            self._sync_reported_benefit_inputs(previous)
+
+    def _install_reported_benefit_inputs(self, prepared, start, count):
+        self._prune_reported_benefit_inputs()
+        ledger = getattr(self, "_reported_benefit_inputs", {}).copy()
+        self._reported_benefit_inputs = ledger
+        self._reported_benefit_window = (start, count)
+        self._reported_benefit_requests = []
+        self._reported_benefit_inputs_enabled = True
+        for year, values in prepared.items():
+            for variable, value in values.items():
+                holder = self.get_holder(variable)
+                period = holder._storage_period(period_(year))
+                branch = self.branch_name
+                key = (variable, branch, period)
+                prior = ledger.get(key)
+                genuine = (
+                    prior["genuine"]
+                    if prior is not None
+                    else self._reported_genuine_tiers(holder, branch, period)
+                )
+                self.set_input(variable, year, value)
+                token = self._reported_input_token(holder, branch, period)
+                if token is None or not holder._stores_user_input(period, branch):
+                    continue
+                ledger[key] = {
+                    "variable": variable,
+                    "period": period,
+                    "branch": branch,
+                    "genuine": genuine,
+                    "synthetic": self._reported_genuine_tiers(holder, branch, period),
+                    "token": token,
+                    "visible": True,
+                }
+        self.drop_computed_arrays()
+
     def delete_arrays(self, variable: str, period: Period = None) -> None:
         super().delete_arrays(variable, period)
+        # Explicit deletion retires ownership even when the inactive view
+        # currently holds genuine absence at that physical slot.
+        ledger = getattr(self, "_reported_benefit_inputs", {})
+        deleted = period_(period) if period is not None else None
+        visible = set(self._get_visible_branch_names())
+        for key, entry in list(ledger.items()):
+            if (
+                entry["variable"] == variable
+                and entry["branch"] in visible
+                and (deleted is None or deleted.contains(entry["period"]))
+            ):
+                del ledger[key]
         # From policyengine-core 3.32.27 (policyengine-core#561) the deletion
         # forgets the input records of the values it removes, but not records
         # that a population rebuild left with no stored value
@@ -620,6 +1131,10 @@ def delete_arrays(self, variable: str, period: Period = None) -> None:
                     del records[recorded_period]
 
     def build_from_populations(self, populations) -> None:
+        self._reported_benefit_inputs = {}
+        self._reported_benefit_requests = []
+        self._reported_benefit_observers = []
+        self._reported_benefit_observer_error = None
         # Take a copy of what another simulation made, so that neither changes
         # the other's values or records.
         populations, adopted = copy_other_populations(self, populations)
@@ -654,16 +1169,54 @@ def set_input(self, variable_name: str, period: Period, value) -> None:
             getattr(self, "input_variables", None) is not None and calculation is None
         )
         stored = set()
+        failures = []
+
+        def remember_failure(error):
+            failures.append(
+                (
+                    error,
+                    error.__traceback__,
+                    error.__cause__,
+                    error.__context__,
+                    error.__suppress_context__,
+                )
+            )
+
         try:
             recording = (
                 self._record_stored_inputs() if external else nullcontext(stored)
             )
             with recording as stored:
-                super().set_input(variable_name, period, value)
-        finally:
+                try:
+                    super().set_input(variable_name, period, value)
+                except BaseException as error:
+                    remember_failure(error)
+        except BaseException as error:
+            remember_failure(error)
+        try:
             forget_replaced_model_values(self, variable_name, records)
+        except BaseException as error:
+            remember_failure(error)
+        try:
             if external and stored:
                 self._external_input_change_pending = True
+        except BaseException as error:
+            remember_failure(error)
+        try:
+            self._prune_reported_benefit_inputs()
+        except BaseException as error:
+            remember_failure(error)
+        if failures:
+            error, traceback, cause, context, suppression = failures[0]
+            BaseException.__setattr__(
+                error, "_policyengine_input_cleanup_failures", tuple(failures[1:])
+            )
+            try:
+                raise BaseException.with_traceback(error, traceback)
+            finally:
+                BaseException.__setattr__(error, "__cause__", cause)
+                BaseException.__setattr__(error, "__context__", context)
+                BaseException.__setattr__(error, "__suppress_context__", suppression)
         # Inputs do not carry back, so give the simulation its identifiers
         # for the year the input starts in if it holds them only for later
         # years. See ``policyengine_uk.data.input_dataframe.hold_identifiers_in``.
@@ -672,7 +1225,26 @@ def set_input(self, variable_name: str, period: Period, value) -> None:
             self._capture_cgt_supplied_inputs(variable_name, replace_period=period)
 
     def clone(self, *args, **kwargs) -> "Simulation":
+        self._prune_reported_benefit_inputs()
         clone = super().clone(*args, **kwargs)
+        clone._reported_benefit_inputs = {}
+        clone._reported_benefit_requests = []
+        clone._reported_benefit_observers = []
+        clone._reported_benefit_observer_error = None
+        for key, entry in getattr(self, "_reported_benefit_inputs", {}).items():
+            copied = {
+                **entry,
+                "genuine": {
+                    tier: value.copy() for tier, value in entry["genuine"].items()
+                },
+                "synthetic": {
+                    tier: value.copy() for tier, value in entry["synthetic"].items()
+                },
+            }
+            copied["token"] = clone._reported_input_token(
+                clone.get_holder(entry["variable"]), entry["branch"], entry["period"]
+            )
+            clone._reported_benefit_inputs[key] = copied
         clone._construction_reform_state = copy.deepcopy(
             getattr(self, "_construction_reform_state", None)
         )
@@ -689,6 +1261,10 @@ def clone(self, *args, **kwargs) -> "Simulation":
         clone._savings_income_projection = getattr(
             self, "_savings_income_projection", None
         )
+        # Expose a stable physical model-input state even when this clone was
+        # created during an inactive caller. A direct Holder deletion before
+        # its first calculate then removes a real tuple and cannot be missed.
+        clone._sync_reported_benefit_inputs(True)
         return clone
 
     def _isa_latest_input(self, name, period, *, earlier=False):
@@ -1572,6 +2148,10 @@ def _start_new_population(self, built_from_dataset: bool) -> None:
         """Record the new population's source, and drop what the simulation
         cached for the previous one, so rebuilding in place (e.g. a clone)
         never reads arrays sized for the old population."""
+        self._reported_benefit_inputs = {}
+        self._reported_benefit_requests = []
+        self._reported_benefit_observers = []
+        self._reported_benefit_observer_error = None
         self.built_from_dataset = built_from_dataset
         if getattr(self, "_fast_cache", None) is not None:
             self._fast_cache = {}
@@ -1708,7 +2288,7 @@ def calculate_all(self, year: int) -> None:
             savings_income_projection=getattr(self, "_savings_income_projection", None),
         )
 
-    def _calculate(self, variable_name: str, period: Period = None):
+    def _calculate_with_bsp_routing(self, variable_name: str, period: Period = None):
         # Core requires an indexed input to have no formula. Keep the legacy
         # CPI declaration, but carry split dataset BSP nominally after the
         # materialized horizon, including through Core's monthly conversion.
@@ -1759,6 +2339,7 @@ def calculate(
         map_to: str = None,
         decode_enums: bool = False,
     ):
+        self._check_reported_overlay_observer()
         if self._cgt_calculation_depth == 0 and self._external_input_change_pending:
             self._external_input_change_pending = False
             drop_calculated_values(self)
@@ -1775,57 +2356,60 @@ def calculate(
 
         period = period_(period)
 
-        projection = self.tax_benefit_system.capital_gains_projection
-        if projection is not None and variable_name in CGT_INPUTS:
-            holder = self.get_holder(variable_name)
-            annual_period = period.this_year
-            if holder.get_array(annual_period, self.branch_name) is None:
-                earlier_inputs = [
-                    p
-                    for p in holder.get_input_periods(self.branch_name)
-                    if p.unit == "year"
-                    and p.start < annual_period.start
-                    and holder.get_array(p, self.branch_name) is not None
-                ]
-                if earlier_inputs:
-                    source_year = max(p.start.year for p in earlier_inputs)
-                    projection.validate_span(source_year, annual_period.start.year)
-
-        if variable_name in SUPPLIED_INPUT_VARIABLES:
-            # Rebuilding the populations leaves the input records of the
-            # replaced holders behind (policyengine-core#605). Forget a record
-            # whose stored value is gone before the engine can refill the
-            # period with a carried-over value that the record would pass off
-            # as supplied.
-            drop_missing_supplied_inputs(self, variable_name)
-
-        outer_calculation = self._cgt_calculation_depth == 0
-        if outer_calculation:
-            self._cgt_operation_inputs = set()
-        record = None
-        neutralized = False
-        if variable_name in CGT_INPUTS:
-            neutralized = self._cgt_is_neutralized(variable_name, period)
-            record = self._cgt_calculation_record(variable_name, period, neutralized)
-        self._cgt_calculation_depth += 1
-        try:
-            result = super().calculate(
-                variable_name, period, map_to=map_to, decode_enums=decode_enums
-            )
-            if record is not None:
-                if neutralized:
-                    self._cgt_defaulted_periods.add((variable_name, str(period)))
-                else:
-                    self._cgt_input_provenance.setdefault(variable_name, {})[
-                        str(period)
-                    ] = record
-                self._cgt_operation_inputs.add((variable_name, str(period)))
-        finally:
-            self._cgt_calculation_depth -= 1
-        if outer_calculation and self._cgt_operation_inputs:
-            for input_period in sorted({p for _, p in self._cgt_operation_inputs}):
-                self._warn_unused_cgt_projection(period_(input_period))
-        return result
+        with self._reported_benefit_context(period):
+            projection = self.tax_benefit_system.capital_gains_projection
+            if projection is not None and variable_name in CGT_INPUTS:
+                holder = self.get_holder(variable_name)
+                annual_period = period.this_year
+                if holder.get_array(annual_period, self.branch_name) is None:
+                    earlier_inputs = [
+                        p
+                        for p in holder.get_input_periods(self.branch_name)
+                        if p.unit == "year"
+                        and p.start < annual_period.start
+                        and holder.get_array(p, self.branch_name) is not None
+                    ]
+                    if earlier_inputs:
+                        source_year = max(p.start.year for p in earlier_inputs)
+                        projection.validate_span(source_year, annual_period.start.year)
+
+            if variable_name in SUPPLIED_INPUT_VARIABLES:
+                # Rebuilding the populations leaves the input records of the
+                # replaced holders behind (policyengine-core#605). Forget a record
+                # whose stored value is gone before the engine can refill the
+                # period with a carried-over value that the record would pass off
+                # as supplied.
+                drop_missing_supplied_inputs(self, variable_name)
+
+            outer_calculation = self._cgt_calculation_depth == 0
+            if outer_calculation:
+                self._cgt_operation_inputs = set()
+            record = None
+            neutralized = False
+            if variable_name in CGT_INPUTS:
+                neutralized = self._cgt_is_neutralized(variable_name, period)
+                record = self._cgt_calculation_record(
+                    variable_name, period, neutralized
+                )
+            self._cgt_calculation_depth += 1
+            try:
+                result = super().calculate(
+                    variable_name, period, map_to=map_to, decode_enums=decode_enums
+                )
+                if record is not None:
+                    if neutralized:
+                        self._cgt_defaulted_periods.add((variable_name, str(period)))
+                    else:
+                        self._cgt_input_provenance.setdefault(variable_name, {})[
+                            str(period)
+                        ] = record
+                    self._cgt_operation_inputs.add((variable_name, str(period)))
+            finally:
+                self._cgt_calculation_depth -= 1
+            if outer_calculation and self._cgt_operation_inputs:
+                for input_period in sorted({p for _, p in self._cgt_operation_inputs}):
+                    self._warn_unused_cgt_projection(period_(input_period))
+            return result
 
     def apply_dynamics(self, year: int):
         """Apply dynamics to the simulation for a specific year.
diff --git a/policyengine_uk/tests/test_child_benefit_charge_rounding_properties.py b/policyengine_uk/tests/test_child_benefit_charge_rounding_properties.py
index 7e448efab2..3bbfcd7b99 100644
--- a/policyengine_uk/tests/test_child_benefit_charge_rounding_properties.py
+++ b/policyengine_uk/tests/test_child_benefit_charge_rounding_properties.py
@@ -109,11 +109,11 @@ def test_a_float32_annual_amount_is_taken_to_the_penny_first():
     assert round_down_to_pounds(np.float32(1406.5999755859375)) == 1406
 
 
-# Structural reforms are chosen when the simulation is built, so the switch has
-# to be set before data load.
+# Enable the household-based reform for the 2026 period these tests calculate.
+# Apply the parameter change before loading the test household.
 HOUSEHOLD_BASED = Scenario(
     applied_before_data_load=True,
-    parameter_changes={"gov.contrib.conservatives.cb_hitc_household": {"2025": True}},
+    parameter_changes={"gov.contrib.conservatives.cb_hitc_household": {"2026": True}},
 )
 
 
diff --git a/policyengine_uk/tests/test_structural_reform_periods.py b/policyengine_uk/tests/test_structural_reform_periods.py
new file mode 100644
index 0000000000..a66ba01801
--- /dev/null
+++ b/policyengine_uk/tests/test_structural_reform_periods.py
@@ -0,0 +1,1108 @@
+"""Parameter-gated structural reforms follow their calculation periods.
+
+Issue #1800 concerns constructor dispatch, which the policy YAML runner's
+explicit reform classes bypass. These public Simulation cases therefore use
+dated dictionaries and equivalent Scenario construction. They preserve the
+existing contributed formulas, rather than asserting new statutory policy.
+
+The current expanded-MA formula deducts its transfer from earned taxable
+income: at GBP80,000 the GBP1,260 transfer saves GBP504 at 40%, not a GBP252
+basic-rate credit. The reported-only cases declare synthetic survey reports;
+their isolated payments are not claims of natural statutory entitlement.
+"""
+
+import copy
+import math
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from policyengine_uk import Simulation
+from policyengine_uk.data import UKMultiYearDataset, UKSingleYearDataset
+from policyengine_uk.reforms.cps.marriage_tax_reforms import (
+    expanded_ma_reform,
+    marriage_neutral_it_reform,
+)
+from policyengine_uk.utils.scenario import Scenario
+
+
+YEARS = (2025, 2026, 2027)
+CPS = "gov.contrib.cps.marriage_tax_reforms"
+EXPAND = f"{CPS}.expanded_ma.remove_income_condition"
+MA_RATE = f"{CPS}.expanded_ma.ma_rate"
+MA_CHILD_AGE = f"{CPS}.expanded_ma.max_child_age"
+NEUTRAL = f"{CPS}.marriage_neutral_it.neutralise_income_tax"
+HOUSEHOLD_HICBC = "gov.contrib.conservatives.cb_hitc_household"
+SCP = "gov.contrib.scotland.scottish_child_payment.in_effect"
+REPORTED_ONLY = "gov.contrib.policyengine.disable_simulated_benefits"
+FY2026 = "2026-04-06.2027-04-05"
+
+
+@pytest.fixture(params=("dict", "scenario"))
+def route(request):
+    return request.param
+
+
+def annual(value, years=YEARS):
+    return {year: value for year in years}
+
+
+def construct(situation=None, changes=None, route="dict", dataset=None):
+    kwargs = (
+        {"situation": copy.deepcopy(situation)} if situation else {"dataset": dataset}
+    )
+    if changes:
+        if route == "dict":
+            kwargs["reform"] = changes
+        else:
+            kwargs["scenario"] = Scenario.from_reform(changes)
+    simulation = Simulation(**kwargs)
+    assert simulation.default_input_period == 2025
+    return simulation
+
+
+def money(simulation, variable, year):
+    return np.asarray(simulation.calculate(variable, year), dtype=float)
+
+
+def equals(actual, expected):
+    np.testing.assert_allclose(actual, expected, rtol=0, atol=0.01)
+
+
+def couple(earnings=(80_000, 0), child=False, region="LONDON", years=YEARS):
+    people = {}
+    for index, income in enumerate(earnings):
+        people[f"person{index + 1}"] = {
+            "age": {year: year - (1985 + index) for year in years},
+            "date_of_birth": annual((1985 + index) * 10_000 + 1006, years),
+            "employment_income": annual(income, years),
+            "marital_status": annual("MARRIED", years),
+            "would_claim_marriage_allowance": annual(True, years),
+        }
+    if child:
+        people["child1"] = {
+            "age": {year: year - 2015 for year in years},
+            "date_of_birth": annual(20151006, years),
+            "employment_income": annual(0, years),
+            "marital_status": annual("SINGLE", years),
+            "would_claim_scp": annual(True, years),
+        }
+    members = list(people)
+    return {
+        "people": people,
+        "benunits": {
+            "benunit": {
+                "members": members,
+                "is_married": annual(True, years),
+                "would_claim_child_benefit": annual(True, years),
+                "child_benefit_opts_out": annual(False, years),
+            }
+        },
+        "households": {
+            "household": {"members": members, "region": annual(region, years)}
+        },
+    }
+
+
+def scp_family(year, age=0, region="SCOTLAND", takes_up=True):
+    """An explicit October6 birthday agrees with the supplied annual age."""
+    return {
+        "people": {
+            "person1": {
+                "age": {year: 40},
+                "date_of_birth": {year: (year - 40) * 10_000 + 1006},
+            },
+            "child1": {
+                "age": {year: age},
+                "date_of_birth": {year: (year - age) * 10_000 + 1006},
+                "would_claim_scp": {year: takes_up},
+            },
+        },
+        "benunits": {
+            "benunit": {
+                "members": ["person1", "child1"],
+                # A declared qualifying award; SCP itself is not overridden.
+                "universal_credit": {year: 5_000},
+            }
+        },
+        "households": {
+            "household": {"members": ["person1", "child1"], "region": {year: region}}
+        },
+    }
+
+
+def reported_dataset():
+    """Three public adults, with all ten supported years supplied explicitly."""
+    datasets = []
+    for year in range(2025, 2035):
+        roles = [True, False, True] if year == 2026 else [True, True, False]
+        multiplier = 1 if year == 2025 else 10
+        person = pd.DataFrame(
+            {
+                "person_id": [1, 2, 3],
+                "person_benunit_id": [1, 1, 1],
+                "person_household_id": [1, 1, 1],
+                "age": [40.0, 39.0, 30.0],
+                "is_claimant_or_partner": roles,
+                "child_benefit_reported": np.array([111.0, 222.0, 333.0]) * multiplier,
+                "esa_income_reported": np.array([100.0, 200.0, 300.0]) * multiplier,
+                "carers_allowance_reported": [4_000.0, 0.0, 0.0],
+                "care_hours": [40.0, 0.0, 0.0],
+            }
+        )
+        datasets.append(
+            UKSingleYearDataset(
+                person=person,
+                benunit=pd.DataFrame({"benunit_id": [1]}),
+                household=pd.DataFrame(
+                    {
+                        "household_id": [1],
+                        "region": ["SCOTLAND" if year == 2026 else "LONDON"],
+                        "household_weight": [1.0],
+                        "tenure_type": ["OWNED_OUTRIGHT"],
+                        "rent": [0.0],
+                        "council_tax": [0.0],
+                    }
+                ),
+                fiscal_year=year,
+            )
+        )
+    return UKMultiYearDataset(datasets=datasets)
+
+
+def test_future_expanded_ma_applies_only_inside_interval(route):
+    """12570 x10%, rounded up to10, gives1260; higher-rate tax falls504."""
+    simulation = construct(couple(), {EXPAND: {FY2026: True}}, route)
+    for year in YEARS:
+        active = year == 2026
+        equals(
+            money(simulation, "marriage_allowance", year), [1_260 if active else 0, 0]
+        )
+        # Existing earned-income deduction:7540 + (80000-12570-37700-MA)*40%.
+        equals(money(simulation, "income_tax", year), [18_928 if active else 19_432, 0])
+
+
+def test_expanded_ma_child_limit_follows_calculation_period():
+    changes = {
+        EXPAND: {"2025-04-06.2028-04-05": True},
+        MA_CHILD_AGE: {"2025": 5, "2026": 15, "2027": 5},
+    }
+    simulation = construct(couple(child=True), changes)
+    for year in YEARS:
+        equals(
+            money(simulation, "marriage_allowance", year),
+            [1_260 if year == 2026 else 0, 0, 0],
+        )
+
+
+def test_expanded_ma_rate_only_later_change_preserves_off_year_statutory_cap(route):
+    simulation = construct(couple(), {MA_RATE: {FY2026: 0.2}}, route)
+    for year in YEARS:
+        active = year == 2026
+        # The reform's own20% cap is2520 after rounding, saving1008 at40%.
+        equals(
+            money(simulation, "marriage_allowance", year), [2_520 if active else 0, 0]
+        )
+        equals(money(simulation, "income_tax", year), [18_424 if active else 19_432, 0])
+
+
+def test_statutory_marriage_allowance_rate_alone_does_not_activate_cps():
+    statutory = "gov.hmrc.income_tax.allowances.marriage_allowance.max"
+    simulation = construct(couple(), {statutory: {FY2026: 0.0}})
+    for year in YEARS:
+        equals(money(simulation, "marriage_allowance", year), [0, 0])
+        equals(money(simulation, "income_tax", year), [19_432, 0])
+
+
+def test_marriage_neutral_income_tax_applies_only_inside_interval(route):
+    simulation = construct(couple(), {NEUTRAL: {FY2026: True}}, route)
+    for year in YEARS:
+        active = year == 2026
+        equals(
+            money(simulation, "net_income", year),
+            [40_000, 40_000] if active else [80_000, 0],
+        )
+        # Each spouse pays(40000-12570)*20%=5486 after an advantageous split.
+        equals(
+            money(simulation, "income_tax", year),
+            [5_486, 5_486] if active else [19_432, 0],
+        )
+
+
+def test_household_hicbc_applies_only_inside_interval(route):
+    simulation = construct(
+        couple((45_000, 45_000), child=True), {HOUSEHOLD_HICBC: {FY2026: True}}, route
+    )
+    for year in YEARS:
+        child_benefit = float(money(simulation, "child_benefit", year).sum())
+        assert child_benefit > 0
+        # Each individual is below60000; combined90000 exceeds80000, so the
+        # household reform charges100% of CB, rounded down to whole pounds.
+        expected = math.floor(round(child_benefit, 2)) if year == 2026 else 0
+        assert money(simulation, "CB_HITC", year).sum() == pytest.approx(
+            expected, abs=0.01
+        )
+
+
+def test_scp_activation_beyond_old_discovery_window(route):
+    changes = {SCP: {"2035-04-06.2036-04-05": True}}
+    for year in (2034, 2035, 2036):
+        # Each independent household contains an explicitly dated under-one
+        # child that year; no future DOB is assigned to an earlier year.
+        simulation = construct(scp_family(year), changes, route)
+        p = simulation.tax_benefit_system.parameters(year)
+        weekly = (
+            p.gov.contrib.scotland.scottish_child_payment.under_one
+            if year == 2035
+            else p.gov.social_security_scotland.scottish_child_payment.amount
+        )
+        equals(money(simulation, "scottish_child_payment", year), [0, 52 * weekly])
+
+
+@pytest.mark.parametrize(
+    "control", ("false", "age_one", "outside_scotland", "no_takeup")
+)
+def test_scp_off_age_location_and_takeup_controls(control):
+    year = 2035
+    changes = {SCP: {"2035-04-06.2036-04-05": control != "false"}}
+    situation = scp_family(
+        year,
+        age=1 if control == "age_one" else 0,
+        region="LONDON" if control == "outside_scotland" else "SCOTLAND",
+        takes_up=control != "no_takeup",
+    )
+    simulation = construct(situation, changes)
+    weekly = simulation.tax_benefit_system.parameters(
+        year
+    ).gov.social_security_scotland.scottish_child_payment.amount
+    expected = 0 if control in ("outside_scotland", "no_takeup") else 52 * weekly
+    equals(money(simulation, "scottish_child_payment", year), [0, expected])
+
+
+def test_reported_only_switch_writes_only_active_dataset_years(route):
+    simulation = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, route=route, dataset=reported_dataset()
+    )
+    for year in YEARS:
+        # No children: the natural CB entitlement is0. The mode substitutes
+        # only2026's666 base-year report, not its6660 later-year reports.
+        equals(money(simulation, "child_benefit", year), [666 if year == 2026 else 0])
+    recorded = {
+        str(period)
+        for variable, _, period in simulation._user_input_keys
+        if variable == "child_benefit"
+    }
+    assert recorded == {"2026"}
+
+
+def test_reported_only_period_role_and_country_mapping_is_preserved():
+    changes = {REPORTED_ONLY: {FY2026: True}}
+    simulation = construct(changes=changes, dataset=reported_dataset())
+    # Current-year roles choose members1+3, but the amounts come from2025.
+    equals(money(simulation, "esa_income", 2026), [600])
+    equals(money(simulation, "claimant_or_partner_esa_income", 2026), [400])
+    # The single4000 report moves to Scotland's CSP only in the active year.
+    equals(money(simulation, "carers_allowance", 2026), [0, 0, 0])
+    equals(money(simulation, "carer_support_payment", 2026), [4_000, 0, 0])
+    baseline = construct(dataset=reported_dataset())
+    for year in (2025, 2027):
+        for variable in (
+            "esa_income",
+            "claimant_or_partner_esa_income",
+            "carers_allowance",
+            "carer_support_payment",
+        ):
+            equals(money(simulation, variable, year), money(baseline, variable, year))
+
+
+def test_disjoint_true_false_true_intervals_restore_original_outputs():
+    changes = {EXPAND: {"2025": True, "2026": False, "2027": True}}
+    simulation = construct(couple(), changes)
+    for year in YEARS:
+        active = year != 2026
+        equals(
+            money(simulation, "marriage_allowance", year), [1_260 if active else 0, 0]
+        )
+        equals(money(simulation, "income_tax", year), [18_928 if active else 19_432, 0])
+
+
+def test_reverse_query_order_baseline_instances_and_clones_are_isolated():
+    simulation = construct(couple(), {EXPAND: {FY2026: True}})
+    clone = simulation.clone()
+    branch = simulation.get_branch("period_control", clone_system=True)
+    for year in (2027, 2025, 2026):
+        expected = [1_260 if year == 2026 else 0, 0]
+        for item in (simulation, clone, branch):
+            equals(money(item, "marriage_allowance", year), expected)
+        equals(money(simulation.baseline, "marriage_allowance", year), [0, 0])
+    fresh = construct(couple())
+    equals(money(fresh, "marriage_allowance", 2026), [0, 0])
+    # A branch's genuine user override stays branch-local.
+    branch.set_input("employment_income", 2026, [60_000, 0])
+    equals(money(simulation, "employment_income", 2026), [80_000, 0])
+    equals(money(clone, "employment_income", 2026), [80_000, 0])
+
+
+def test_independent_dated_switches_compose_without_losing_either():
+    interval = "2027-04-06.2028-04-05"
+    changes = {HOUSEHOLD_HICBC: {interval: True}, SCP: {interval: True}}
+    for year in (2026, 2027, 2028):
+        situation = couple(
+            (45_000, 45_000), child=True, region="SCOTLAND", years=(year,)
+        )
+        situation["people"]["child1"].update(
+            age={year: 0}, date_of_birth={year: year * 10_000 + 1006}
+        )
+        situation["benunits"]["benunit"]["universal_credit"] = {year: 5_000}
+        simulation = construct(situation, changes)
+        cb = float(money(simulation, "child_benefit", year).sum())
+        expected_charge = math.floor(round(cb, 2)) if year == 2027 else 0
+        assert money(simulation, "CB_HITC", year).sum() == pytest.approx(
+            expected_charge, abs=0.01
+        )
+        p = simulation.tax_benefit_system.parameters(year)
+        weekly = (
+            p.gov.contrib.scotland.scottish_child_payment.under_one
+            if year == 2027
+            else p.gov.social_security_scotland.scottish_child_payment.amount
+        )
+        equals(money(simulation, "scottish_child_payment", year), [0, 0, 52 * weekly])
+
+
+@pytest.mark.parametrize("kind", ("expanded_ma", "neutral"))
+def test_direct_bypass_reform_interfaces_keep_their_existing_behavior(kind):
+    reform = expanded_ma_reform if kind == "expanded_ma" else marriage_neutral_it_reform
+    simulation = Simulation(situation=couple(), reform=reform)
+    if kind == "expanded_ma":
+        equals(money(simulation, "marriage_allowance", 2026), [1_260, 0])
+        equals(money(simulation, "income_tax", 2026), [18_928, 0])
+    else:
+        equals(money(simulation, "net_income", 2026), [40_000, 40_000])
+        equals(money(simulation, "income_tax", 2026), [5_486, 5_486])
+
+
+# Additional public ownership/context controls for the period-scoped input
+# overlay. The original twenty-four cases above remain unchanged.
+def test_reported_mode_equal_value_genuine_replacement_survives_off_year():
+    from policyengine_core.periods import period as period_
+
+    for direct_holder, same_object in ((False, False), (True, True)):
+        simulation = construct(
+            changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+        )
+        value = simulation.calculate("child_benefit", 2026)
+        equals(value, [666])
+        genuine = value if same_object else np.array([666], dtype=value.dtype)
+        if direct_holder:
+            simulation.get_holder("child_benefit").set_input(
+                period_(2026),
+                genuine,
+                branch_name=simulation.branch_name,
+            )
+        else:
+            simulation.set_input("child_benefit", 2026, genuine)
+        Scenario(parameter_changes={REPORTED_ONLY: False}).apply(simulation)
+        equals(money(simulation, "child_benefit", 2027), [0])
+        equals(money(simulation, "child_benefit", 2026), [666])
+        assert any(
+            name == "child_benefit" and str(period) == "2026"
+            for name, _, period in simulation._user_input_keys
+        )
+        simulation.set_input("child_benefit", 2026, np.array([0.0]))
+        equals(money(simulation, "child_benefit", 2026), [0])
+        assert any(
+            name == "child_benefit" and str(period) == "2026"
+            for name, _, period in simulation._user_input_keys
+        )
+
+
+def test_reported_mode_nested_requested_periods_restore_caller_state():
+    from policyengine_uk.model_api import BenUnit, Reform, Variable, YEAR
+
+    class active_period_probe(Variable):
+        label = "Public active-period transition probe"
+        entity = BenUnit
+        definition_period = YEAR
+        value_type = float
+
+        def formula(benunit, period):
+            simulation = benunit.simulation
+            # A real active -> inactive -> active transition within the
+            # inactive outer request. All people in this fixture are adults.
+            active = simulation.calculate("child_benefit", 2026)
+            inactive = simulation.calculate("child_benefit", 2027)
+            return active + inactive + simulation.calculate("child_benefit", 2026)
+
+    class inactive_period_probe(Variable):
+        label = "Public inactive-period transition probe"
+        entity = BenUnit
+        definition_period = YEAR
+        value_type = float
+
+        def formula(benunit, period):
+            return benunit.simulation.calculate("active_period_probe", 2026) + benunit(
+                "child_benefit", period
+            )
+
+    class probes(Reform):
+        def apply(self):
+            self.add_variable(active_period_probe)
+            self.add_variable(inactive_period_probe)
+
+    simulation = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+    )
+    simulation.apply_reform(probes)
+    # 666 + 0 + 666 from declared reports, plus natural no-child CB0.
+    for _ in range(2):
+        equals(money(simulation, "inactive_period_probe", 2027), [1332])
+        equals(money(simulation, "child_benefit", 2027), [0])
+        equals(money(simulation, "child_benefit", 2026), [666])
+
+
+def test_reported_mode_clone_exception_finally_keeps_genuine_inputs():
+    from policyengine_uk.model_api import BenUnit, Reform, Variable, YEAR
+
+    class failing_period_probe(Variable):
+        label = "Public cross-year exception probe"
+        entity = BenUnit
+        definition_period = YEAR
+        value_type = float
+
+        def formula(benunit, period):
+            simulation = benunit.simulation
+            equals(money(simulation, "child_benefit", 2026), [666])
+            equals(money(simulation, "child_benefit", 2027), [0])
+            # This genuine replacement occurs WHILE the inactive outer
+            # context is suspended, not merely before the failing query.
+            simulation.set_input("child_benefit", 2026, [999])
+            raise ValueError("reported-mode public cross-year probe")
+
+    class probe(Reform):
+        def apply(self):
+            self.add_variable(failing_period_probe)
+
+    parent = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+    )
+    clone = parent.clone()
+    clone.apply_reform(probe)
+    with pytest.raises(ValueError, match="^reported-mode public cross-year probe$"):
+        clone.calculate("failing_period_probe", 2027)
+    equals(money(clone, "child_benefit", 2026), [999])
+    equals(money(parent, "child_benefit", 2026), [666])
+    for simulation in (parent, clone):
+        equals(money(simulation, "child_benefit", 2027), [0])
+    # An equal genuine write before cloning must also cease being model-owned.
+    parent.set_input("child_benefit", 2026, [666])
+    branch = parent.get_branch("genuine_reported_control", clone_system=True)
+    Scenario(parameter_changes={REPORTED_ONLY: False}).apply(branch)
+    equals(money(branch, "child_benefit", 2026), [666])
+    equals(money(parent, "child_benefit", 2026), [666])
+
+
+# DRAFT supplemental engineering regressions. These are not adopted or run.
+@pytest.mark.parametrize(
+    "history,expected",
+    (
+        ({"0000-01-01": False, "2026-01-01": True, "2027-01-01": None}, None),
+        ({"0000-01-01": False, "2026-01-01": False}, False),
+        ({"0000-01-01": False, "2026-01-01": True, "2027-01-01": False}, True),
+    ),
+    ids=("true_then_null_rejected", "all_false", "true_then_false_available"),
+)
+def test_structural_discovery_validates_complete_real_parameter_history(
+    history, expected
+):
+    from policyengine_core.parameters import Parameter
+    from policyengine_uk.reforms.reforms import finite_parameter_activation
+
+    # Real Parameter parsing preserves the literal None and raw sentinel.
+    # The literal oracle requires complete supported history, not just one
+    # available interval. No Simulation/converter null-survival is assumed.
+    leaf = Parameter("structural_history_probe", data=history)
+    assert [(entry.instant_str, entry.value) for entry in leaf.values_list] == sorted(
+        history.items(), reverse=True
+    )
+    if expected is None:
+        assert leaf._get_at_instant("2027-01-01") is None
+        with pytest.raises(
+            ValueError, match="^Structural reform parameter has no value at 2027-01-01$"
+        ):
+            finite_parameter_activation([leaf], lambda values: values[0])
+    else:
+        assert finite_parameter_activation([leaf], lambda values: values[0]) is expected
+
+
+@pytest.mark.parametrize(
+    "completion,delete",
+    (("return", True), ("exception", True), ("return", False), ("exception", False)),
+    ids=("delete_return", "delete_exception", "control_return", "control_exception"),
+)
+def test_reported_mode_direct_holder_absent_delete_is_not_resurrected(
+    completion, delete
+):
+    from policyengine_core.periods import period as period_
+    from policyengine_uk.model_api import BenUnit, Reform, Variable, YEAR
+
+    class absent_slot_delete_probe(Variable):
+        label = "Public absent-slot deletion probe"
+        entity = BenUnit
+        definition_period = YEAR
+        value_type = float
+
+        def formula(benunit, period):
+            simulation = benunit.simulation
+            equals(money(simulation, "child_benefit", 2027), [0])
+            holder = simulation.get_holder("child_benefit")
+            target = period_(2026)
+            branch = simulation.branch_name
+            # Prove this exact physical slot contains no genuine input before
+            # arranging the original no-op-delete premise. Missing slots need
+            # no derived mark; every present memory/disk slot must be derived.
+            assert ("child_benefit", branch, target) not in simulation._user_input_keys
+            assert holder._stores_user_input(target, branch) is False
+            for storage in (holder._memory_storage, holder._disk_storage):
+                if storage is not None:
+                    present = storage.has(target, branch)
+                    assert type(present) is bool
+                    if present:
+                        assert storage.is_derived(target, branch) is True
+            simulation.drop_computed_arrays()
+            # This is the actual physical absence that tuple-only tracking
+            # misses: no input record and no array in either physical tier.
+            assert not holder._memory_storage.has(target, branch)
+            assert holder._disk_storage is None or not holder._disk_storage.has(
+                target, branch
+            )
+            assert ("child_benefit", branch, target) not in simulation._user_input_keys
+            if delete:
+                holder.delete_arrays(target, branch_name=branch)
+            if completion == "exception":
+                raise ValueError("reported-mode absent-slot deletion probe")
+            return np.zeros(benunit.count)
+
+    class probe(Reform):
+        def apply(self):
+            self.add_variable(absent_slot_delete_probe)
+
+    simulation = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+    )
+    simulation.apply_reform(probe)
+    if completion == "exception":
+        with pytest.raises(
+            ValueError, match="^reported-mode absent-slot deletion probe$"
+        ):
+            simulation.calculate("absent_slot_delete_probe", 2027)
+    else:
+        equals(money(simulation, "absent_slot_delete_probe", 2027), [0])
+    equals(money(simulation, "child_benefit", 2026), [0 if delete else 666])
+    equals(money(simulation, "child_benefit", 2027), [0])
+    key = ("child_benefit", simulation.branch_name, period_(2026))
+    assert (key in simulation._user_input_keys) is (not delete)
+    # Retire only the deleted slot; the unrelated declared report remains.
+    equals(money(simulation, "esa_income", 2026), [600])
+
+
+@pytest.mark.parametrize(
+    "completion,action",
+    (
+        ("return", "delete"),
+        ("exception", "delete"),
+        ("return", "none"),
+        ("exception", "none"),
+        ("return", "genuine"),
+        ("exception", "genuine"),
+    ),
+    ids=(
+        "delete_return",
+        "delete_exception",
+        "control_return",
+        "control_exception",
+        "genuine_return",
+        "genuine_exception",
+    ),
+)
+def test_reported_mode_inactive_clone_delete_before_first_calculate(completion, action):
+    from policyengine_core.periods import period as period_
+    from policyengine_uk.model_api import BenUnit, Reform, Variable, YEAR
+
+    created = []
+
+    class inactive_clone_escape_probe(Variable):
+        label = "Public inactive clone escape probe"
+        entity = BenUnit
+        definition_period = YEAR
+        value_type = float
+
+        def formula(benunit, period):
+            parent = benunit.simulation
+            equals(money(parent, "child_benefit", 2027), [0])
+            target = period_(2026)
+            physical_branch = parent.branch_name
+            parent_holder = parent.get_holder("child_benefit")
+            # Prove this exact physical slot contains no genuine input before
+            # arranging the original no-op-delete premise. Missing slots need
+            # no derived mark; every present memory/disk slot must be derived.
+            assert (
+                "child_benefit",
+                physical_branch,
+                target,
+            ) not in parent._user_input_keys
+            assert parent_holder._stores_user_input(target, physical_branch) is False
+            for storage in (parent_holder._memory_storage, parent_holder._disk_storage):
+                if storage is not None:
+                    present = storage.has(target, physical_branch)
+                    assert type(present) is bool
+                    if present:
+                        assert storage.is_derived(target, physical_branch) is True
+            parent.drop_computed_arrays()
+            assert not parent_holder._memory_storage.has(target, physical_branch)
+            assert (
+                parent_holder._disk_storage is None
+                or not parent_holder._disk_storage.has(target, physical_branch)
+            )
+            clone = parent.clone()
+            created.append((clone, physical_branch))
+            # No clone.calculate call precedes this direct physical operation;
+            # the clone has no own Core calculation frame here.
+            holder = clone.get_holder("child_benefit")
+            if action == "delete":
+                holder.delete_arrays(target, branch_name=physical_branch)
+            elif action == "genuine":
+                holder.set_input(target, np.array([999.0]), branch_name=physical_branch)
+            if completion == "exception":
+                raise ValueError("reported-mode inactive clone escape probe")
+            return np.zeros(benunit.count)
+
+    class probe(Reform):
+        def apply(self):
+            self.add_variable(inactive_clone_escape_probe)
+
+    parent = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+    )
+    parent.apply_reform(probe)
+    if completion == "exception":
+        with pytest.raises(
+            ValueError, match="^reported-mode inactive clone escape probe$"
+        ):
+            parent.calculate("inactive_clone_escape_probe", 2027)
+    else:
+        equals(money(parent, "inactive_clone_escape_probe", 2027), [0])
+    assert created
+    clone, physical_branch = created[-1]
+    expected = {"delete": 0, "none": 666, "genuine": 999}[action]
+    equals(money(clone, "child_benefit", 2026), [expected])
+    key = ("child_benefit", physical_branch, period_(2026))
+    assert (key in clone._user_input_keys) is (action != "delete")
+    equals(money(clone, "esa_income", 2026), [600])
+    equals(money(parent, "child_benefit", 2026), [666])
+    for item in (clone, parent):
+        equals(money(item, "child_benefit", 2027), [0])
+
+
+@pytest.mark.parametrize("target", ("self", "clone", "branch"))
+@pytest.mark.parametrize("replay", ("variable", "noop", "failure_before_replay"))
+def test_reported_owned_inputs_survive_public_reform_replay(target, replay):
+    from policyengine_uk.model_api import BenUnit, Reform, Variable, YEAR
+
+    class public_replay_probe(Variable):
+        label = "Public reported-input replay probe"
+        entity = BenUnit
+        definition_period = YEAR
+        value_type = float
+
+        def formula(benunit, period):
+            return np.zeros(benunit.count)
+
+    class public_replay(Reform):
+        def apply(self):
+            if replay == "failure_before_replay":
+                raise ValueError("public reform failed before cache replay")
+            if replay == "variable":
+                self.add_variable(public_replay_probe)
+
+    parent = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+    )
+    simulation = (
+        parent
+        if target == "self"
+        else parent.clone()
+        if target == "clone"
+        else parent.get_branch("reported_public_replay", clone_system=True)
+    )
+    equals(money(simulation, "child_benefit", 2026), [666])
+    if replay == "failure_before_replay":
+        with pytest.raises(
+            ValueError, match="^public reform failed before cache replay$"
+        ):
+            simulation.apply_reform(public_replay)
+    else:
+        simulation.apply_reform(public_replay)
+    equals(money(simulation, "child_benefit", 2027), [0])
+    equals(money(simulation, "child_benefit", 2026), [666])
+    equals(money(simulation, "esa_income", 2026), [600])
+    equals(money(parent, "child_benefit", 2026), [666])
+    if replay == "variable":
+        equals(money(simulation, "public_replay_probe", 2027), [0])
+
+
+@pytest.mark.parametrize("mutation", ("equal", "same_array", "new", "delete"))
+def test_public_reform_replay_preserves_genuine_holder_write_or_delete(mutation):
+    from policyengine_core.periods import period as period_
+    from policyengine_uk.model_api import Reform
+
+    parent = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+    )
+    simulation = parent.clone()
+    holder = simulation.get_holder("child_benefit")
+    target = period_(2026)
+    if mutation == "delete":
+        expected = 0
+    else:
+        value = (
+            holder.get_array(target, simulation.branch_name)
+            if mutation == "same_array"
+            else np.array([999 if mutation == "new" else 666.0])
+        )
+        expected = 999 if mutation == "new" else 666
+
+    class no_policy_change(Reform):
+        def apply(self):
+            if mutation == "delete":
+                holder.delete_arrays(target, branch_name=simulation.branch_name)
+            else:
+                holder.set_input(target, value, branch_name=simulation.branch_name)
+
+    simulation.apply_reform(no_policy_change)
+    equals(money(simulation, "child_benefit", 2026), [expected])
+    equals(money(simulation, "child_benefit", 2027), [0])
+    equals(money(simulation, "esa_income", 2026), [600])
+    equals(money(parent, "child_benefit", 2026), [666])
+
+
+def _traceback_tail(traceback):
+    while traceback.tb_next is not None:
+        traceback = traceback.tb_next
+    return traceback
+
+
+@pytest.mark.parametrize(
+    "failure_stages",
+    (
+        ("core", "recording", "forget", "prune"),
+        ("recording", "forget", "prune"),
+        ("forget", "prune"),
+        ("prune",),
+        ("core",),
+        (),
+    ),
+    ids=(
+        "all_fail",
+        "exit_first",
+        "forget_first",
+        "prune_only",
+        "core_only",
+        "success",
+    ),
+)
+def test_input_cleanup_preserves_first_failure_and_attempts_remaining_steps(
+    monkeypatch, failure_stages
+):
+    """A failed input operation still accounts for its store and all cleanup.
+
+    Mock the Core store boundary rather than build a dataset: this tests the
+    composition of UK cleanup, not the storage implementation or policy math.
+    """
+    from contextlib import contextmanager
+
+    import policyengine_uk.simulation as uk_simulation
+
+    simulation = object.__new__(Simulation)
+    simulation.input_variables = []
+    simulation._external_input_change_pending = False
+    calls, stored, stored_values, originals = [], set(), {}, {}
+    errors = {
+        stage: (
+            KeyboardInterrupt("synthetic Core input failure")
+            if stage == "core"
+            else ValueError(f"synthetic {stage} failure")
+        )
+        for stage in failure_stages
+    }
+    for index, error in enumerate(errors.values()):
+        error.__cause__ = RuntimeError("original explicit cause")
+        error.__context__ = LookupError("original implicit context")
+        error.__suppress_context__ = bool(index % 2)
+
+    def fail_if_requested(stage):
+        if stage in errors:
+            try:
+                raise errors[stage]
+            except BaseException as error:
+                originals[stage] = (
+                    error.__traceback__,
+                    error.__cause__,
+                    error.__context__,
+                    error.__suppress_context__,
+                )
+                raise
+
+    @contextmanager
+    def record_stores():
+        calls.append("recording_enter")
+        try:
+            yield stored
+        finally:
+            calls.append("recording_exit")
+            fail_if_requested("recording")
+
+    def core_store(owner, name, period, value):
+        calls.append("core")
+        stored_values[(name, period)] = value
+        stored.add((name, "default", period))
+        fail_if_requested("core")
+
+    def forget(owner, name, records):
+        calls.append("forget")
+        fail_if_requested("forget")
+
+    def prune():
+        calls.append("prune")
+        # Even failed Core/context/forget cleanup must retain the pending
+        # invalidation for a store that completed before the error.
+        assert simulation._external_input_change_pending is True
+        fail_if_requested("prune")
+
+    monkeypatch.setattr(simulation, "_record_stored_inputs", record_stores)
+    monkeypatch.setattr(simulation, "_prune_reported_benefit_inputs", prune)
+    monkeypatch.setattr(uk_simulation.CoreSimulation, "set_input", core_store)
+    monkeypatch.setattr(uk_simulation, "model_value_input_records", lambda *args: {})
+    monkeypatch.setattr(uk_simulation, "forget_replaced_model_values", forget)
+    monkeypatch.setattr(
+        uk_simulation, "hold_identifiers_in", lambda *args: calls.append("identifiers")
+    )
+    monkeypatch.setattr(
+        simulation,
+        "_capture_cgt_supplied_inputs",
+        lambda *args, **kwargs: calls.append("cgt"),
+    )
+    value = object()
+    if failure_stages:
+        with pytest.raises(BaseException) as caught:
+            simulation.set_input("capital_gains", 2026, value)
+        primary = failure_stages[0]
+        error = errors[primary]
+        traceback, cause, context, suppression = originals[primary]
+        assert caught.value is error
+        assert _traceback_tail(error.__traceback__) is _traceback_tail(traceback)
+        assert error.__cause__ is cause
+        assert error.__context__ is context
+        assert error.__suppress_context__ is suppression
+        secondary = error._policyengine_input_cleanup_failures
+        assert type(secondary) is tuple
+        assert [record[0] for record in secondary] == [
+            errors[stage] for stage in failure_stages[1:]
+        ]
+        for stage, (failure, original_tb, cause, context, suppression) in zip(
+            failure_stages[1:], secondary, strict=True
+        ):
+            expected_tb, expected_cause, expected_context, expected_suppression = (
+                originals[stage]
+            )
+            assert failure is errors[stage]
+            assert _traceback_tail(original_tb) is _traceback_tail(expected_tb)
+            assert cause is expected_cause
+            assert context is expected_context
+            assert suppression is expected_suppression
+    else:
+        simulation.set_input("capital_gains", 2026, value)
+    expected_calls = ["recording_enter", "core", "recording_exit", "forget", "prune"]
+    if not failure_stages:
+        expected_calls += ["identifiers", "cgt"]
+    assert calls == expected_calls
+    assert stored_values[("capital_gains", 2026)] is value
+    assert simulation._external_input_change_pending is True
+
+
+def test_reported_observer_refusal_precedes_pending_cache_invalidation(monkeypatch):
+    import policyengine_uk.simulation as uk_simulation
+
+    simulation = object.__new__(Simulation)
+    simulation._reported_benefit_observer_error = "synthetic prior observer refusal"
+    simulation._external_input_change_pending = True
+    simulation._cgt_calculation_depth = 0
+    invalidated = []
+    monkeypatch.setattr(
+        uk_simulation, "drop_calculated_values", lambda owner: invalidated.append(owner)
+    )
+    with pytest.raises(RuntimeError, match="synthetic prior observer refusal"):
+        simulation.calculate("child_benefit", 2026)
+    assert invalidated == []
+    assert simulation._external_input_change_pending is True
+
+
+@pytest.mark.parametrize("reported_overlay", (False, True), ids=("plain", "reported"))
+@pytest.mark.parametrize("fails", (False, True), ids=("return", "exception"))
+def test_bsp_projection_routing_remains_inside_reported_observer(
+    monkeypatch, reported_overlay, fails
+):
+    """The split BSP route and reported-input attempt observer both survive."""
+    from types import SimpleNamespace
+
+    from policyengine_core.periods import period as period_
+    from policyengine_core.simulations.simulation import _calculation_frames
+
+    simulation = object.__new__(Simulation)
+    simulation.branch_name = "default"
+    simulation.built_from_dataset = True
+    simulation._reported_benefit_inputs = (
+        {"present": object()} if reported_overlay else {}
+    )
+    simulation._reported_benefit_observers = []
+    source, target = period_(2025), period_(2027)
+    holder = SimpleNamespace(get_input_periods=lambda branch: [source])
+    variable = SimpleNamespace(
+        formulas={},
+        is_neutralized=False,
+        uprating="gov.economic_assumptions.indices.obr.consumer_price_index",
+    )
+    simulation.tax_benefit_system = SimpleNamespace(get_variable=lambda name: variable)
+    monkeypatch.setattr(simulation, "get_holder", lambda name: holder)
+    calls, observed = [], []
+    result, error = object(), ValueError("synthetic BSP source failure")
+
+    def calculate_source(name, period):
+        calls.append((name, period))
+        observed.append(bool(simulation._reported_benefit_observers))
+        if name == "bsp_reported" and fails:
+            raise error
+        return result
+
+    monkeypatch.setattr(simulation, "calculate", calculate_source)
+    frame = SimpleNamespace(simulation=simulation, changes=[])
+    token = _calculation_frames.set((frame,))
+    try:
+        if fails:
+            with pytest.raises(
+                ValueError, match="synthetic BSP source failure"
+            ) as caught:
+                simulation._calculate("bsp_reported", target)
+            assert caught.value is error
+        else:
+            assert simulation._calculate("bsp_reported", target) is result
+        # The two dependency names form a set; their evaluation order is not
+        # part of the BSP contract. Both precede the retained source amount.
+        assert len(calls) == 3
+        assert set(calls[:2]) == {
+            ("widowed_parents_allowance_reported", source),
+            ("war_widows_pension_reported", source),
+        }
+        assert calls[-1] == ("bsp_reported", source)
+        assert observed == [reported_overlay] * 3
+        assert simulation._reported_benefit_observers == []
+    finally:
+        _calculation_frames.reset(token)
+
+
+def test_clone_copies_isa_contract_before_synchronizing_reported_inputs(monkeypatch):
+    """Reported-input synchronization sees the clone's declared dataset metadata."""
+    import policyengine_uk.simulation as uk_simulation
+    from policyengine_uk.data.savings_income_projection import (
+        OBR_SAVINGS_INCOME_2022_2025,
+    )
+
+    parent = object.__new__(Simulation)
+    parent._reported_benefit_inputs = {}
+    parent._savings_income_projection = OBR_SAVINGS_INCOME_2022_2025
+    parent._isa_income_contracts = {
+        2025: {"version": 1, "representation": "additive", "projection": "legacy_gross"}
+    }
+    original = copy.deepcopy(parent._isa_income_contracts)
+    calls = []
+    monkeypatch.setattr(
+        parent, "_prune_reported_benefit_inputs", lambda: calls.append("prune")
+    )
+
+    def core_clone(owner, *args, **kwargs):
+        calls.append("core_clone")
+        clone = object.__new__(Simulation)
+        clone.__dict__.update(owner.__dict__)
+        # Require the UK clone boundary to copy its own metadata, rather than
+        # rely on the generic Core clone carrying this subclass attribute.
+        clone.__dict__.pop("_savings_income_projection")
+        return clone
+
+    def synchronize(clone, active):
+        calls.append(("synchronize", active))
+        assert clone._isa_income_contracts == original
+        assert clone._isa_income_contracts is not parent._isa_income_contracts
+        assert (
+            clone._isa_income_contracts[2025] is not parent._isa_income_contracts[2025]
+        )
+        assert clone._savings_income_projection == OBR_SAVINGS_INCOME_2022_2025
+        clone._isa_income_contracts[2025]["projection"] = "additive"
+
+    monkeypatch.setattr(uk_simulation.CoreSimulation, "clone", core_clone)
+    monkeypatch.setattr(uk_simulation, "copy_model_values", lambda *args: None)
+    monkeypatch.setattr(Simulation, "_sync_reported_benefit_inputs", synchronize)
+    clone = parent.clone()
+    assert calls == ["prune", "core_clone", ("synchronize", True)]
+    assert parent._isa_income_contracts == original
+    assert clone._isa_income_contracts[2025]["projection"] == "additive"
+
+
+@pytest.mark.parametrize("model_value", (False, True), ids=("supplied", "model"))
+def test_population_rebuild_retires_reports_and_keeps_donor_input_provenance(
+    model_value,
+):
+    """A new population drops the old reports and keeps its own values' origin."""
+    from policyengine_core.periods import period as period_
+    from policyengine_uk.utils.supplied_inputs import set_model_value, supplied_input
+
+    simulation = construct(
+        changes={REPORTED_ONLY: {FY2026: True}}, dataset=reported_dataset()
+    )
+    equals(money(simulation, "child_benefit", 2026), [666])
+    assert simulation._reported_benefit_inputs
+    simulation.set_input("adjusted_net_income", 2026, [1, 2, 3])
+
+    donor = construct(dataset=reported_dataset())
+    original = np.array([10_000.0, 20_000.0, 30_000.0])
+    if model_value:
+        set_model_value(donor, "adjusted_net_income", period_(2026), original)
+    else:
+        donor.set_input("adjusted_net_income", 2026, original)
+    donor_population = donor.populations["person"]
+
+    simulation.build_from_populations(donor.populations)
+
+    assert simulation._reported_benefit_inputs == {}
+    assert simulation._reported_benefit_requests == []
+    assert simulation._reported_benefit_observers == []
+    assert simulation._reported_benefit_observer_error is None
+    assert donor.populations["person"] is donor_population
+    assert donor_population.simulation is donor
+    assert simulation.populations["person"] is not donor_population
+    equals(money(simulation, "adjusted_net_income", 2026), original)
+    supplied = supplied_input(
+        simulation.populations["person"], "adjusted_net_income", period_(2026)
+    )
+    if model_value:
+        assert supplied is None
+    else:
+        equals(supplied, original)
+    # The replacement population has adults and no supplied Child Benefit.
+    # The old population's reported award must not return in the active year.
+    equals(money(simulation, "child_benefit", 2026), [0])
+
+    simulation.set_input("adjusted_net_income", 2026, [40_000, 50_000, 60_000])
+    equals(money(donor, "adjusted_net_income", 2026), original)
+    equals(money(simulation, "adjusted_net_income", 2026), [40_000, 50_000, 60_000])
+    donor.set_input("adjusted_net_income", 2026, [70_000, 80_000, 90_000])
+    equals(money(simulation, "adjusted_net_income", 2026), [40_000, 50_000, 60_000])
diff --git a/pyproject.toml b/pyproject.toml
index 2d24e5d468..dd8d38f758 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -24,7 +24,7 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "policyengine-core>=3.33.0",
+    "policyengine-core>=3.33.2",
     "microdf-python>=1.2.1",
     "pydantic>=2.11.7",
     "tables>=3.10.2",
diff --git a/uv.lock b/uv.lock
index b0f7c84413..6321b90cdf 100644
--- a/uv.lock
+++ b/uv.lock
@@ -2259,7 +2259,7 @@ requires-dist = [
     { name = "hypothesis", marker = "extra == 'dev'" },
     { name = "jupyter-book", marker = "extra == 'dev'", specifier = ">=2.0.0a0" },
     { name = "microdf-python", specifier = ">=1.2.1" },
-    { name = "policyengine-core", specifier = ">=3.33.0" },
+    { name = "policyengine-core", specifier = ">=3.33.2" },
     { name = "pydantic", specifier = ">=2.11.7" },
     { name = "pytest-cov", marker = "extra == 'dev'" },
     { name = "pytest-xdist", marker = "extra == 'dev'" },