@@ -1973,7 +1973,7 @@ async def _send_control(self, entity_id, command, **kwargs):
                     self._acks_seen = False
                     self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it")
             if not same:
-                entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": [], "attempt_ids": []}
+                entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": [], "attempt_ids": [], "acks": 0}
                 self._pending_commands[entity_id] = entry
             previous = {"command_ids": list(entry["command_ids"]), "attempt_ids": list(entry["attempt_ids"]), "outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]}
             if entry["state"] != "sent":
@@ -1988,9 +1988,9 @@ async def _send_control(self, entity_id, command, **kwargs):
             # Earlier ids of the same value stay matchable: an ack for any of them confirms it
             entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_ref]
             entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]}
-            entry["attempt_ids"] = entry["attempt_ids"] + [command_ref]
+            entry["attempt_ids"] = [ref for ref in entry["attempt_ids"] if ref in entry["command_ids"]] + [command_ref]
             entry["sent_at"] = now
-            outcomes_sent = dict(entry["outcomes"])
+            acks_before = entry["acks"]
 
         try:
             await self.publish_command(command, command_id=command_id, **kwargs)
@@ -2003,7 +2003,7 @@ async def _send_control(self, entity_id, command, **kwargs):
                 if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"] and command_ref not in entry["outcomes"]:
                     if not previous["command_ids"]:
                         del self._pending_commands[entity_id]
-                    elif entry["outcomes"] == outcomes_sent:
+                    elif entry["acks"] == acks_before:
                         # No ack arrived meanwhile: back to how the earlier attempt left it,
                         # including any id pruned to make room for this one
                         entry.update(previous)
@@ -2041,6 +2041,7 @@ def _process_ack(self, data):
             if entry is None or ack.get("command", entry["command"]) != entry["command"]:
                 return
             self._acks_seen = True
+            entry["acks"] += 1  # lets a failing publish tell whether any ack arrived meanwhile
             # A replay for the newest id, before any unit answered it, means the hub has seen that
             # id before and never considered this send. Drop the id; if nothing else has answered
             # the value, let the next attempt go out under a fresh id straight away. For an older
@@ -2051,6 +2052,7 @@ def _process_ack(self, data):
                 # Another send in this attempt already applied the value: only forget this id
                 entry["command_ids"].remove(command_id)
                 entry["attempt_ids"].remove(command_id)
+                entry["sent_at"] = _monotonic()  # still a refusal: the window runs from it
             elif drop_replay:
                 del self._pending_commands[entity_id]
             else:
 
@@ -5177,15 +5177,21 @@ def test_replay_for_the_newest_id_after_an_earlier_send_applied(self):
             self._run(gw.number_event(self.CHARGE_RATE, 2000))
             clock["now"] = 1030.0
             self._run(gw.number_event(self.CHARGE_RATE, 2000))
+            clock["now"] = 1031.0
             self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0})
+            clock["now"] = 1059.0
             self._ack(gw, "PBAT2", "set_charge_rate", ok=False, error="replay")
             entry = gw._pending_commands[self.CHARGE_RATE]
             assert entry["command_ids"] == ["PBAT1"]
             assert entry["state"] == "applied"
             assert self._logged(gw, "already applied by another send")
-            clock["now"] = 1040.0
+            # The replay is a refusal too: the window runs from it
+            clock["now"] = 1060.0
             self._run(gw.number_event(self.CHARGE_RATE, 2000))
-        assert len(gw._published) == 2
+            assert len(gw._published) == 2
+            clock["now"] = 1089.0
+            self._run(gw.number_event(self.CHARGE_RATE, 2000))
+        assert len(gw._published) == 3
 
     def test_replay_for_the_newest_id_with_only_an_older_attempts_ok_drops_the_entry(self):
         """An ok from an earlier attempt does not count: a replay of the fresh attempt's id re-sends straight away."""
@@ -5232,6 +5238,34 @@ async def failing_publish_command(command, command_id=None, **kwargs):
             self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0})
         assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied"
 
+    def test_repeated_refusal_during_a_failing_publish_is_not_rolled_back(self):
+        """A refusal repeated while a later publish awaits the broker keeps its restarted window when that publish fails."""
+        gw = self._make_gateway()
+        clock, patcher = self._clock()
+        with patcher:
+            self._run(gw.number_event(self.CHARGE_RATE, 2000))
+            self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="not_managed")
+
+            async def refusal_then_fail(command, command_id=None, **kwargs):
+                clock["now"] = 1035.0
+                self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="not_managed")
+                raise RuntimeError("broker gone")
+
+            real_publish = gw.publish_command
+            gw.publish_command = refusal_then_fail
+            clock["now"] = 1031.0
+            try:
+                self._run(gw.number_event(self.CHARGE_RATE, 2000))
+            except RuntimeError:
+                pass
+            gw.publish_command = real_publish
+            entry = gw._pending_commands[self.CHARGE_RATE]
+            assert entry["command_ids"] == ["PBAT1"]
+            assert entry["sent_at"] == 1035.0
+            clock["now"] = 1036.0
+            self._run(gw.number_event(self.CHARGE_RATE, 2000))
+        assert len(gw._published) == 1
+
     def test_outcomes_do_not_grow_past_the_kept_ids(self):
         """Per-id outcomes are pruned with the ids, so a value re-sent for a long time does not grow the entry."""
         gw = self._make_gateway()
@@ -5245,6 +5279,19 @@ def test_outcomes_do_not_grow_past_the_kept_ids(self):
         assert len(entry["command_ids"]) == 16
         assert set(entry["outcomes"]) <= set(entry["command_ids"])
 
+    def test_attempt_ids_do_not_grow_without_acks(self):
+        """Subscribed but no acks ever: every call re-sends, and the id lists stay bounded."""
+        gw = self._make_gateway(acks_seen=False)
+        clock, patcher = self._clock()
+        with patcher:
+            for step in range(50):
+                clock["now"] = 1000.0 + step * 2
+                self._run(gw.number_event(self.CHARGE_RATE, 2000))
+        entry = gw._pending_commands[self.CHARGE_RATE]
+        assert len(gw._published) == 50
+        assert len(entry["command_ids"]) == 16
+        assert len(entry["attempt_ids"]) == 16
+
     def test_second_ok_ack_does_not_overwrite_cache(self):
         """Only the first ok ack for a command updates the cached slot times."""
         gw = self._make_gateway()