Never publish a plan computed from inputs that have since changed

Confirmed: the check run after the optimizer compared only zone sensors,
outdoor temperature, thermostat and relay. A window opening, a UPS load
change or a wind change during the calculation went unnoticed; nothing
prevented an older update that finished late from overwriting a newer
result; and an update in flight during unload still saved.

- Every input a prediction uses is captured as a normalized snapshot
  (zone and outdoor temperatures, wind speed/direction/gust, heat-source
  power, window states, relay state and switch time, thermostat mode and
  target) and compared after the calculation with tolerances (0.25 °C,
  50 W, 1 m/s, 2 m/s gusts, 20° direction around the circle; exact for
  relay, thermostat and windows). A meaningful change discards the plan
  and lists the inputs in changed_inputs; noise does not. No immediate
  recalculation: an off-schedule poll would break the 5-minute learning
  interval; the next regular poll computes a fresh plan.
- Updates are numbered; one overtaken by a newer update, or running when
  the entry unloads, keeps the newer data and skips saving and
  optimizing. Configuration changes reload the entry, so they are covered.
- A per-entry storage lock makes a reloaded coordinator restore only
  after the previous one's save finished, and integration removal waits
  for it too, so a late save cannot recreate a deleted model file.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

custom_components/heating_mpc/__init__.py

@@ -12,7 +12,7 @@ from homeassistant.helpers.event import async_track_state_change_event
 
 from .const import CONF_HEATER, CONF_ROOMS, CONF_WEIGHTS, DOMAIN, CONF_OUTSIDE, WEATHER_SOURCES
 from .weather_sources import sources_for
-from .coordinator import HeatingMPCCoordinator
+from .coordinator import HeatingMPCCoordinator, storage_lock
 from .zones import SUBENTRY_ZONE, ZONE_KEY, ZONE_NAME, legacy_zone_data
 
 PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.BINARY_SENSOR]
@@ -62,7 +62,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
     """Remove the learned model when this integration instance is deleted."""
     store = Store(hass, 1, f"{DOMAIN}.{entry.entry_id}")
+    # Wait for a save still in progress, which would otherwise recreate the file.
+    async with storage_lock(hass, entry.entry_id):
         await store.async_remove()
+    hass.data.get(f"{DOMAIN}_storage_locks", {}).pop(entry.entry_id, None)
 
 
 async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

custom_components/heating_mpc/coordinator.py

@@ -1,6 +1,7 @@
 """Read existing HA entities, learn thermal response, produce shadow MPC decisions."""
 from __future__ import annotations
 
+import asyncio
 import logging
 from collections import deque
 from datetime import datetime, timedelta, timezone
@@ -21,7 +22,7 @@ from .const import (
     CONF_VENT_WIND, CONF_WINDOW_OPEN_MINUTES, DEFAULTS, READY_OFF_SAMPLES, READY_ON_SAMPLES, READY_SAMPLES,
 )
 from .compatibility import FINGERPRINT_VERSION, baseline_fingerprints, changed, reason, zone_fingerprints
-from .data_quality import ReportAgeHistory, freshness
+from .data_quality import ReportAgeHistory, freshness, inputs_differ
 from .evaluation import ForecastEvaluator, model_snapshot
 from .heat_sources import SUBENTRY_HEAT_SOURCE, HeatSource, HeatSourceTracker, build_sources, normalize_power
 from .learning import FeatureLearner
@@ -30,7 +31,9 @@ from .physics import (
     lag_from_time_in_state, optimize, relay_segments,
 )
 from .weather_sources import choose_weather_sources, sources_for, normalize
-from .windows import SUBENTRY_WINDOW, VentilationTracker, Window, build_windows, facing_cos, ventilation_vector
+from .windows import (
+    SUBENTRY_WINDOW, VentilationTracker, Window, build_windows, contact_open, facing_cos, ventilation_vector,
+)
 from .zone_model import ZoneModel, compile_step_model, extended_features, feature_vector
 from .zones import SUBENTRY_ZONE, Zone, build_zones, legacy_zone_data, zone_temperature
 
@@ -43,6 +46,14 @@ MAX_MODEL_CHANGES = 20
 BASE_TERMS = ("loss", "heat", "bias")
 
 
+def storage_lock(hass: HomeAssistant, entry_id: str) -> asyncio.Lock:
+    """One lock per entry, shared by successive coordinators across reloads.
+
+    A reload's new coordinator restores only after the old one's save finished.
+    """
+    return hass.data.setdefault(f"{DOMAIN}_storage_locks", {}).setdefault(entry_id, asyncio.Lock())
+
+
 def entry_zone_list(entry: ConfigEntry, config: dict[str, Any]) -> list[Zone]:
     """Zones from the entry's zone subentries; one zone per legacy room sensor otherwise."""
     subentries = getattr(entry, "subentries", None) or {}
@@ -130,8 +141,22 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._relay_events_dropped = False
         self._on_samples = 0
         self._off_samples = 0
+        # Increases with every update; an older update finishing late never publishes.
+        self._run = 0
 
     async def async_restore_model(self) -> None:
+        async with storage_lock(self.hass, self.entry.entry_id):
+            await self._async_restore_model()
+
+    async def _async_save(self, outside_source: str | None) -> None:
+        async with storage_lock(self.hass, self.entry.entry_id):
+            await self._store.async_save(self._payload(outside_source))
+
+    def _stale(self, run: int) -> bool:
+        """True when a newer update started or the entry is unloading."""
+        return run != self._run or getattr(self, "_shutdown_requested", False)
+
+    async def _async_restore_model(self) -> None:
         stored = await self._store.async_load() or {}
         if not isinstance(stored, dict) or stored.get("schema_version", 1) not in (1, 2):
             _LOGGER.warning("Ignoring unsupported Heating MPC model data")
@@ -189,7 +214,7 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             # Save at once: a recalibration is not redone after a restart, and data
             # from before fingerprints gets them now, so the next edit is detected
             # even if it happens before the next learned sample.
-            await self._store.async_save(self._payload(stored.get("outside_source")))
+            await self._store.async_save(self._payload(stored.get("outside_source")))  # lock held
 
     def _check_compatibility(self, stored: dict[str, Any] | None, base_prints: dict[str, str | None]) -> bool:
         """Recalibrate exactly the components whose configuration changed."""
@@ -318,6 +343,11 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._relay_events_dropped = False
         return steady
 
+    def _superseded(self, status: dict[str, Any]) -> dict[str, Any]:
+        """Result of an update that a newer one (or unloading) overtook: keep what is shown."""
+        current = getattr(self, "data", None)
+        return current if current is not None else {**status, "status": "inputs_changed", "proposal": None}
+
     def _count_result(self, status: dict[str, Any], result: str) -> None:
         """Record how this poll was handled for learning, and the learning progress."""
         status["sample_result"] = result
@@ -435,22 +465,36 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         return {"windows": windows, "ventilation": zones,
                 "ventilation_events": list(self._ventilation.events)[-5:]}
 
-    def _prediction_inputs(self) -> tuple:
-        """Values the prediction depends on, used to detect changes during calculation."""
-        heater = self.config[CONF_HEATER]
-        entities = sorted(set(self.sensors) | set(sources_for(self.config, "temperature")) |
-                          {self.config[CONF_CLIMATE], heater})
-        values = []
-        for entity_id in entities:
+    def _input_snapshot(self) -> dict[tuple[str, str], Any]:
+        """Normalized value of every input a prediction uses, keyed by (kind, entity).
+
+        Compared (with tolerances, see data_quality.inputs_differ) before publishing,
+        so a result computed from inputs that have since changed is not shown.
+        """
+        def value(entity_id: str, metric: str):
             state = self.hass.states.get(entity_id)
-            if state is None:
-                values.append((entity_id, None))
-                continue
-            attrs = state.attributes
-            values.append((entity_id, state.state, attrs.get("unit_of_measurement"),
-                           attrs.get("temperature"), attrs.get("temperature_unit"),
-                           state.last_changed if entity_id == heater else None))
-        return tuple(values)
+            return normalize(metric, state.state, state.attributes.get("unit_of_measurement")) if state else None
+
+        snapshot: dict[tuple[str, str], Any] = {}
+        for entity_id in set(self.sensors) | set(sources_for(self.config, "temperature")):
+            snapshot[("temperature", entity_id)] = value(entity_id, "temperature")
+        for metric in ("wind", "wind_direction", "wind_gust"):
+            for entity_id in sources_for(self.config, metric):
+                snapshot[(metric, entity_id)] = value(entity_id, metric)
+        for source in self.sources:
+            state = self.hass.states.get(source.sensor)
+            snapshot[("power", source.sensor)] = normalize_power(
+                state.state, state.attributes.get("unit_of_measurement")) if state else None
+        for window in self.windows:
+            state = self.hass.states.get(window.sensor)
+            snapshot[("window", window.sensor)] = contact_open(state.state if state else None, window.inverted)
+        heater = self.hass.states.get(self.config[CONF_HEATER])
+        snapshot[("relay", self.config[CONF_HEATER])] = (heater.state, heater.last_changed) if heater else None
+        climate = self.hass.states.get(self.config[CONF_CLIMATE])
+        snapshot[("thermostat", self.config[CONF_CLIMATE])] = (
+            climate.state, climate.attributes.get("temperature"), climate.attributes.get("temperature_unit")
+        ) if climate else None
+        return snapshot
 
     async def _async_update_data(self) -> dict[str, Any]:
         now = datetime.now(timezone.utc)
@@ -464,7 +508,9 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         }
         snapshot = {entity: self.hass.states.get(entity)
                     for entity in set(self.sensors) | weather_entity_ids}
-        inputs_before = self._prediction_inputs()
+        self._run += 1
+        run = self._run
+        inputs_before = self._input_snapshot()
         sensor_quality = {}
         readings: dict[str, float | None] = {}
         keep_stale = self.config.get(CONF_ROOM_STALE_KEEP) is True
@@ -649,9 +695,11 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                                                    self.ventilation.get(room))) for room in vals},
             inputs={room: {"internal": internal[room], "ventilation": vent_inputs.get(room)} for room in vals})
         status["forecast_accuracy"] = self._evaluator.summary(self.rooms)
-        if learned:
+        if learned and not self._stale(run):
             # Persist at each sample so restart does not erase hours of training.
-            await self._store.async_save(self._payload(active_sources["temperature"]))
+            await self._async_save(active_sources["temperature"])
+        if self._stale(run):
+            return self._superseded(status)
         # Comfort weights are validated when zones are parsed (invalid -> 1.0).
         weights = {r: self.zones[r].comfort_weight for r in self.rooms}
         # Left-out zones take their weight with them; the rest are re-normalized.
@@ -718,12 +766,17 @@ class HeatingMPCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 step_model=self._step_model(vals, outside, int(self.config[CONF_HORIZON]),
                                             weather, now.timestamp()),
             )
-            # Discard a result if a value used by the prediction changed while
-            # saving or while the worker calculated the hypothetical plan. Optional
-            # diagnostic weather sensors are not prediction inputs.
             output = await self.hass.async_add_executor_job(job)
-            if self._prediction_inputs() != inputs_before:
+            if self._stale(run):
+                return self._superseded(status)
+            # Discard a result if an input it used changed meaningfully while saving
+            # or while the worker calculated the hypothetical plan. Not recomputed at
+            # once: an off-schedule poll would break the 5-minute learning interval;
+            # the next regular poll (at most 5 minutes) computes a fresh plan.
+            changed = inputs_differ(inputs_before, self._input_snapshot())
+            if changed:
                 status["status"] = "inputs_changed"
+                status["changed_inputs"] = [f"{kind}: {entity}" for kind, entity in changed[:10]]
                 return status
         except (ValueError, RuntimeError, OverflowError) as err:
             _LOGGER.warning("MPC prediction failed: %s", err)

custom_components/heating_mpc/data_quality.py

@@ -37,3 +37,37 @@ class ReportAgeHistory:
         else:
             buckets.append([hour, age_seconds])
         return max(age for start, age in buckets if hour - 24 < start <= hour)
+
+
+# How much an input may move while a prediction is computed before the result no
+# longer describes the current situation. Below these, changes are sensor noise.
+INPUT_TOLERANCES = {
+    "temperature": 0.25,   # °C, zone sensors and outdoor sources
+    "power": 50.0,         # W, heat-source sensors
+    "wind": 1.0,           # m/s
+    "wind_gust": 2.0,      # m/s
+    "wind_direction": 20.0,  # degrees, compared around the circle
+}
+
+
+def inputs_differ(before: dict, after: dict) -> list:
+    """Inputs whose change is meaningful: keys are (kind, entity), values normalized.
+
+    Kinds without a tolerance (relay, thermostat mode and target, window states,
+    configuration revision) must match exactly; a value appearing or disappearing
+    (for example a sensor becoming unavailable) always counts.
+    """
+    changed = []
+    for key in sorted(set(before) | set(after), key=str):
+        old, new = before.get(key), after.get(key)
+        tolerance = INPUT_TOLERANCES.get(key[0])
+        if old is None or new is None or tolerance is None:
+            if old != new:
+                changed.append(key)
+            continue
+        difference = abs(new - old)
+        if key[0] == "wind_direction":
+            difference = min(difference, 360.0 - difference)
+        if difference > tolerance:
+            changed.append(key)
+    return changed

tests/test_coordinator.py

@@ -2,6 +2,7 @@
 import importlib.util
 import sys
 import types
+import asyncio
 import unittest
 import unittest.mock
 from datetime import datetime, timedelta, timezone
@@ -1193,3 +1194,105 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase):
         saved = without._payload('sensor.outside')
         back = await self.restored(saved, zones=[{**self.ZONES[0], 'sensors': ['sensor.other']}, self.ZONES[1]])
         self.assertEqual((back.learners['living'].theta, back.learners['living'].count), ([0.004, 0.12, 0.001], 0))
+
+    def ready_office(self):
+        """Office with a UPS and a window, trained enough to optimize."""
+        self.entry.subentries = {
+            'z': types.SimpleNamespace(subentry_id='z', subentry_type='zone', data={
+                'key': 'office', 'name': 'Office', 'sensors': ['sensor.room']}),
+            'h': types.SimpleNamespace(subentry_id='h', subentry_type='heat_source', data={
+                'key': 'ups', 'name': 'UPS', 'sensor': 'sensor.ups_power', 'zone': 'office'}),
+            'w': types.SimpleNamespace(subentry_id='w', subentry_type='window', data={
+                'key': 'w', 'name': 'Window', 'sensor': 'binary_sensor.office_window', 'zone': 'office'})}
+        self.set_state('sensor.ups_power', '200', unit_of_measurement='W')
+        self.set_state('binary_sensor.office_window', 'off')
+        self.c = coordinator.HeatingMPCCoordinator(self.hass, self.entry)
+        base = self.c.learners['office']
+        base.count, base.on_count, base.off_count = 200, 100, 100
+
+    async def change_during_optimization(self, change):
+        async def worker(job):
+            output = job()
+            change()
+            return output
+        self.hass.async_add_executor_job = worker
+        return await self.sample()
+
+    async def test_meaningful_input_changes_during_optimization_discard_the_plan(self):
+        cases = {
+            'window: binary_sensor.office_window': lambda: setattr(self.states['binary_sensor.office_window'], 'state', 'on'),
+            'power: sensor.ups_power': lambda: setattr(self.states['sensor.ups_power'], 'state', '600'),
+            'thermostat: climate.reference': lambda: self.states['climate.reference'].attributes.update(temperature=21),
+            'temperature: sensor.room': lambda: setattr(self.states['sensor.room'], 'state', '20.6'),
+        }
+        for expected, change in cases.items():
+            with self.subTest(change=expected):
+                self.setUp()
+                self.ready_office()
+                data = await self.change_during_optimization(change)
+                self.assertEqual((data['status'], data['proposal'], data['changed_inputs']),
+                                 ('inputs_changed', None, [expected]))
+
+    async def test_sensor_noise_during_optimization_does_not_discard_the_plan(self):
+        self.ready_office()
+        def noise():
+            self.states['sensor.room'].state = '20.1'
+            self.states['sensor.ups_power'].state = '230'
+        self.assertEqual((await self.change_during_optimization(noise))['status'], 'ready')
+
+    async def test_an_older_run_finishing_late_never_overwrites_a_newer_result(self):
+        self.ready_office()
+        release = asyncio.Event()
+        async def slow_then_fast(job):
+            if not release.is_set() and not getattr(self, '_first_started', False):
+                self._first_started = True
+                await release.wait()
+            return job()
+        self.hass.async_add_executor_job = slow_then_fast
+        with patch.object(coordinator, 'datetime') as clock:
+            clock.now.return_value = self.now
+            older = asyncio.ensure_future(self.c._async_update_data())
+            await asyncio.sleep(0)
+            clock.now.return_value = self.now + timedelta(seconds=60)
+            newer = await self.c._async_update_data()
+            self.c.data = newer   # what DataUpdateCoordinator publishes
+            release.set()
+            result = await older
+        self.assertEqual(newer['status'], 'ready')
+        self.assertIs(result, newer)
+
+    async def test_unloading_during_save_or_optimization_publishes_nothing_new(self):
+        self.ready_office()
+        await self.sample()                       # a first sample to learn from
+        previous = self.c.data = {'status': 'learning', 'proposal': None}
+        self.fresh('sensor.room', 'sensor.outside', 'sensor.ups_power')
+        async def unload_while_saving(payload):
+            self.c._shutdown_requested = True
+        self.c._store.async_save.side_effect = unload_while_saving
+        worker = AsyncMock(side_effect=AssertionError('no optimization after unload'))
+        self.hass.async_add_executor_job = worker
+        self.assertIs(await self.sample(), previous)
+        worker.assert_not_called()
+
+    async def test_restore_waits_for_a_save_still_in_progress(self):
+        self.ready_office()
+        order = []
+        release = asyncio.Event()
+        async def slow_save(payload):
+            order.append('save started')
+            await release.wait()
+            order.append('save finished')
+        self.c._store.async_save.side_effect = slow_save
+        saving = asyncio.ensure_future(self.c._async_save('sensor.outside'))
+        await asyncio.sleep(0)
+        successor = coordinator.HeatingMPCCoordinator(self.hass, self.entry)
+        async def load():
+            order.append('load')
+            return None
+        successor._store.async_load.side_effect = load
+        restoring = asyncio.ensure_future(successor.async_restore_model())
+        await asyncio.sleep(0)
+        self.assertEqual(order, ['save started'])  # the new coordinator waits
+        release.set()
+        await asyncio.gather(saving, restoring)
+        self.assertEqual(order, ['save started', 'save finished', 'load'])

tests/test_data_quality.py

@@ -46,6 +46,22 @@ class FreshnessTests(unittest.TestCase):
         self.assertEqual(history.observe('sensor.a', later, 300), 300)
         self.assertEqual(history.observe('sensor.b', later, 30), 30)
 
+    def test_inputs_differ_ignores_noise_but_not_meaningful_changes(self):
+        before = {('temperature', 'sensor.room'): 20.0, ('power', 'sensor.ups'): 210.0,
+                  ('wind_direction', 'sensor.dir'): 355.0, ('wind', 'sensor.wind'): 4.0,
+                  ('window', 'binary_sensor.w'): False, ('thermostat', 'climate.t'): ('heat', 22, None)}
+        noise = {**before, ('temperature', 'sensor.room'): 20.1, ('power', 'sensor.ups'): 240.0,
+                 ('wind_direction', 'sensor.dir'): 5.0, ('wind', 'sensor.wind'): 4.6}
+        self.assertEqual(quality.inputs_differ(before, noise), [])  # 10° across north is small
+        for key, value in ((('temperature', 'sensor.room'), 20.5), (('power', 'sensor.ups'), 610.0),
+                           (('wind_direction', 'sensor.dir'), 40.0), (('window', 'binary_sensor.w'), True),
+                           (('window', 'binary_sensor.w'), None), (('thermostat', 'climate.t'), ('heat', 21, None)),
+                           (('temperature', 'sensor.room'), None)):
+            with self.subTest(key=key, value=value):
+                self.assertEqual(quality.inputs_differ(before, {**before, key: value}), [key])
+        self.assertEqual(quality.inputs_differ(before, {**before, ('power', 'sensor.new'): 5.0}),
+                         [('power', 'sensor.new')])
+
     def test_freshness_settings_translated(self):
         for name in ('strings.json', 'translations/en.json', 'translations/ro.json'):
             data = json.loads((ROOT / name).read_text())