@@ -241,6 +241,7 @@ class GatewayMQTT(ComponentBase):
     _slots_without_capacity = frozenset()
     _last_ratings = {}  # replaced, never changed in place
     _assumed_ratings = frozenset()
+    _reserve_before_zero = {}  # replaced, never changed in place
     gateway_device_id = None
 
     def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqtt_token=None, gateway_inverter_serial=None, gateway_evc_automatic=False, gateway_evc_control=False, gateway_shared_ct=True, gateway_integrate_power=False, **kwargs):
@@ -349,6 +350,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt
         self._slots_without_capacity = frozenset()  # slots bound with no battery size sensor, re-bound once the hub reports a size
         self._last_ratings = {}  # (entity suffix, rating name) -> the last real value the hub reported
         self._assumed_ratings = frozenset()  # entity suffixes of inverters whose maximum rate has never been reported
+        self._reserve_before_zero = {}  # entity suffix -> the last real reserve, set aside when Predbat wrote a reserve of 0
         self._discovery = {}  # upper-case serial -> what the hub last reported it to be (see _update_discovery)
         self._discovery_dirty = False
         self._discovery_save_retry_at = 0
@@ -881,6 +883,8 @@ def _process_telemetry(self, data):
                 for inv in status.inverters:
                     if not inv.primary and self._is_published(inv):
                         self._publish_inverter(inv)
+                # The integration above ran on the slots as they were before this status changed them
+                self._update_integrated_energy(status, rebase=True)
                 self._inject_unreported_inverters(status)
                 self._inject_shared_site_metering(status)
         self._check_configured_inverters(status)
@@ -1077,7 +1081,7 @@ def _inject_shared_site_metering(self, status):
         for name, wh in today.items():
             self.dashboard_item(f"sensor.{pfx}_{name}_today", round(wh / 1000.0, 2), attributes=GATEWAY_ATTRIBUTE_TABLE.get(f"{name}_today", {}), app="gateway")
 
-    def _update_integrated_energy(self, status):
+    def _update_integrated_energy(self, status, rebase=False):
         """Integrate grid and battery power into today's import, export and load (gateway_integrate_power).
 
         Some inverters do not keep usable grid energy counters. GivEnergy AC inverters sharing one
@@ -1097,6 +1101,12 @@ def _update_integrated_energy(self, status):
         Each interval uses the mean of the readings at its two ends. The figures return to zero at
         local midnight. They are kept in memory only, so they also restart from zero when Predbat
         restarts, which Predbat reads the same way as the midnight reset.
+
+        Args:
+            status: A decoded GatewayStatus protobuf message.
+            rebase: True when auto-config has just run on a status that was already integrated.
+                Nothing is added: its readings become the baseline for the inverters now in the
+                slots, so one that has just joined is counted from this status and not the next.
         """
         if not self.gateway_integrate_power:
             return
@@ -1143,7 +1153,7 @@ def _update_integrated_energy(self, status):
             state = {"day": day, "time": None, **zeroed}
             self._integrated_energy = state
         previous = state["time"]
-        if previous is not None and now <= previous:
+        if previous is not None and now <= previous and not rebase:
             # The same status again (a re-inject), or the hub's clock stepping back
             return
 
@@ -1158,6 +1168,9 @@ def _update_integrated_energy(self, status):
 
         if previous is None:
             state["pv"] = {serial: {"base": wh, "high": None, "high_time": 0} for serial, wh in pv_wh.items()}
+        elif rebase:
+            for serial, wh in pv_wh.items():
+                state["pv"].setdefault(serial, {"base": wh, "high": None, "high_time": 0})
         else:
             elapsed = now - previous
             hours = elapsed / 3600.0
@@ -2676,11 +2689,16 @@ async def number_event(self, entity_id, value):
         elif "_charge_rate" in entity_id:
             await self._send_control(entity_id, "set_charge_rate", power_w=val, serial=serial)
         elif "_reserve" in entity_id:
+            await self._send_control(entity_id, "set_reserve", target_soc=val, serial=serial)
             if val == 0:
                 # A reserve of 0% asked for here is real, so the next 0 from the hub is not a gap
-                # to fill with the last value (see _last_rating)
-                self._last_ratings = {key: kept for key, kept in self._last_ratings.items() if key != (_serial_suffix(serial), "reserve")}
-            await self._send_control(entity_id, "set_reserve", target_soc=val, serial=serial)
+                # to fill with the last value (see _last_rating). Only once the write has gone
+                # out, and the value is set aside in case the hub refuses it (see _process_ack).
+                key = (_serial_suffix(serial), "reserve")
+                self._reserve_before_zero = {suffix: kept for suffix, kept in self._reserve_before_zero.items() if suffix != key[0]}
+                if key in self._last_ratings:
+                    self._reserve_before_zero = {**self._reserve_before_zero, key[0]: self._last_ratings[key]}
+                    self._last_ratings = {kept_key: kept for kept_key, kept in self._last_ratings.items() if kept_key != key}
         elif "_target_soc" in entity_id:
             await self._send_control(entity_id, "set_target_soc", target_soc=val, serial=serial)
 
@@ -2880,6 +2898,7 @@ def _process_ack(self, data):
                 apply_cache = ok and entry["cached"] is None
             command = entry["command"]
             serial = entry["kwargs"].get("serial")
+            zero_reserve = entry["kwargs"].get("target_soc") == 0
 
         if ok:
             self.log(f"Info: GatewayMQTT: {command} for {serial} acknowledged by hub ({command_id}) applied={ack.get('applied')}")
@@ -2893,6 +2912,13 @@ def _process_ack(self, data):
             self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: replay ({command_id} already used), {action}")
         else:
             self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}")
+            if command == "set_reserve" and zero_reserve and serial:
+                # The reserve of 0 was not applied, so a 0 from the hub is a gap again and the
+                # reserve it had before is still the last real one
+                key = (_serial_suffix(serial), "reserve")
+                if key[0] in self._reserve_before_zero and key not in self._last_ratings:
+                    self._last_ratings = {**self._last_ratings, key: self._reserve_before_zero[key[0]]}
+                    restore_status = True
             if restore_status and self._last_status is not None:
                 self._inject_entities(self._last_status)
 
 
@@ -4275,6 +4275,78 @@ def test_reserve_the_hub_stops_reporting_keeps_its_last_good_value(self):
         # The other inverter's reserve is its own
         assert gw._states["number.predbat_gateway_23g800_reserve_soc"] == 4
 
+    def test_reserve_is_kept_when_the_write_of_zero_fails(self):
+        """A reserve of 0% that never went out is not real: the hub's 0 is still a gap filled with the last value."""
+        gw = self._writable(self._listed_gateway([]))
+        gw._process_telemetry(self._rated_status().SerializeToString())
+
+        async def failing_send(entity_id, command, **kwargs):
+            raise RuntimeError("publish failed")
+
+        gw._send_control = failing_send
+        try:
+            asyncio.run(gw.number_event("number.predbat_gateway_25g400_reserve_soc", 0))
+            raise AssertionError("the failed publish should have been raised")
+        except RuntimeError:
+            pass
+
+        blank = self._rated_status()
+        blank.inverters[1].control.reserve_soc = 0
+        gw._process_telemetry(blank.SerializeToString())
+        assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 4
+
+    def test_reserve_is_put_back_when_the_hub_refuses_the_write_of_zero(self):
+        """A refused reserve of 0% was not applied, so the reserve the inverter had is published again."""
+        import threading
+
+        gw = self._writable(self._listed_gateway([]))
+        gw._ack_subscribed = True
+        gw._acks_seen = True
+        gw._command_id = 0
+        gw._pending_commands = {}
+        gw._command_lock = threading.Lock()
+        gw._process_telemetry(self._rated_status().SerializeToString())
+        blank = self._rated_status()
+        blank.inverters[1].control.reserve_soc = 0
+
+        asyncio.run(gw.number_event("number.predbat_gateway_25g400_reserve_soc", 0))
+        command, sent = gw._published[-1]
+        assert command == "set_reserve" and sent["target_soc"] == 0
+        gw._process_telemetry(blank.SerializeToString())
+        assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 0
+
+        gw._process_ack(json.dumps({"command_id": f"PBAT{sent['command_id']}", "command": "set_reserve", "ok": False, "error": "modbus_write_failed"}).encode("utf-8"))
+        assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 4
+        # The other inverter is untouched, and an accepted 0 stays 0
+        assert gw._states["number.predbat_gateway_23g800_reserve_soc"] == 4
+
+        asyncio.run(gw.number_event("number.predbat_gateway_23g800_reserve_soc", 0))
+        command, sent = gw._published[-1]
+        blank.inverters[0].control.reserve_soc = 0
+        gw._process_telemetry(blank.SerializeToString())
+        gw._process_ack(json.dumps({"command_id": f"PBAT{sent['command_id']}", "command": "set_reserve", "ok": True}).encode("utf-8"))
+        gw._process_telemetry(blank.SerializeToString())
+        assert gw._states["number.predbat_gateway_23g800_reserve_soc"] == 0
+
+    def test_inverter_that_becomes_usable_is_the_integration_baseline_on_that_status(self):
+        """_process_telemetry re-baselines the integration after auto-config changes the slots."""
+        import time
+
+        gw = self._make_handler_gateway()
+        gw.gateway_integrate_power = True
+        gw.gateway_shared_ct = True
+        gw._integrated_energy = None
+
+        gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString())
+        assert sorted(gw._integrated_energy["pv"]) == ["CE2223G800"]
+
+        joining = self._two_ac_status("ok")
+        joining.timestamp = int(time.time()) + 60
+        joining.inverters[1].energy.pv_today_wh = 7000
+        gw._process_telemetry(joining.SerializeToString())
+        assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"]
+        assert gw._integrated_energy["pv"]["CE2225G400"]["base"] == 7000
+
     def test_reserve_never_reported_is_published_as_zero(self):
         """An inverter that has never reported a reserve still publishes 0, as before."""
         gw = self._listed_gateway([])
@@ -7028,6 +7100,23 @@ def status(timestamp, pv_only_wh, has_battery=False):
         self._feed(gw, status(self.BASE_TIME + 1800, 3500, has_battery=True))
         assert approx_equal(gw._integrated_energy["load_wh"], 500.0)
 
+    def test_rebase_takes_the_new_fleet_power_and_adds_nothing(self):
+        """A rebase on a status already integrated adds no energy and starts the next interval from the new slots' power."""
+        gw = self._make_gateway(shared_ct=False)
+        self._feed(gw, self._status(self.BASE_TIME, grid_w=-1000, battery_w=(0,), serials=("CE2225G400",)))
+        joining = self._status(self.BASE_TIME + 600, grid_w=-1000, battery_w=(0, 0))
+        gw._last_status = joining
+        gw._update_integrated_energy(joining)
+        before = dict((name, gw._integrated_energy[name]) for name in ("import_wh", "export_wh", "load_wh"))
+        assert gw._integrated_energy["power"]["import"] == 1000
+
+        gw.automatic_config()
+        gw._update_integrated_energy(joining, rebase=True)
+
+        assert dict((name, gw._integrated_energy[name]) for name in before) == before
+        assert gw._integrated_energy["power"]["import"] == 2000
+        assert gw._integrated_energy["time"] == self.BASE_TIME + 600
+
     def test_inverter_given_a_slot_later_has_its_generation_counted_from_then(self):
         """An inverter that joins the slots after the integration began is counted from its first status, not left out."""
         gw = self._make_gateway()
@@ -7040,13 +7129,19 @@ def both(timestamp, first_wh, second_wh):
             built.inverters[1].energy.pv_today_wh = second_wh
             return built
 
-        # CE2223G800 appears and takes slot 0; its counter already stands at 7000
-        gw._last_status = both(self.BASE_TIME + 1200, 7000, 1200)
+        # CE2223G800 appears and takes slot 0; its counter already stands at 7000. As in
+        # _process_telemetry, the status is integrated on the old slots, auto-config re-runs,
+        # and the status then becomes the baseline for the new ones
+        joining = both(self.BASE_TIME + 1200, 7000, 1200)
+        gw._last_status = joining
+        gw._update_integrated_energy(joining)
         gw.automatic_config()
         assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"]
-
-        self._feed(gw, both(self.BASE_TIME + 1200, 7000, 1200))
+        gw._update_integrated_energy(joining, rebase=True)
         assert approx_equal(gw._integrated_energy["load_wh"], 200.0)
+        assert gw._integrated_energy["pv"]["CE2223G800"]["base"] == 7000
+
+        # Its generation since the status it joined on is counted, not only from the one after
         self._feed(gw, both(self.BASE_TIME + 1800, 7300, 1250))
         assert approx_equal(gw._integrated_energy["load_wh"], 550.0)