@@ -2675,22 +2675,46 @@ async def update_discharge_window(self, start_entity, end_entity, start, end):
         """
         if getattr(self.base, "set_read_only", False):
             return False
-        if not isinstance(start_entity, str) or not start_entity.endswith("_discharge_slot1_start"):
-            return False
-        if end_entity != start_entity.removesuffix("_start") + "_end":
+        if self.discharge_window_uses_staging(start_entity, end_entity):
             return False
-        serial = self._serial_from_entity_id(start_entity)
-        if not serial or self._serial_from_entity_id(end_entity) != serial:
+        kwargs = self._discharge_window_kwargs(start_entity, end_entity, start, end)
+        if kwargs is None:
             return False
-        if self.discharge_window_uses_staging(start_entity, end_entity):
+        if self.discharge_window_refused(start_entity, end_entity, start, end):
+            # _send_control would not re-send it inside the ack window, so polling again
+            # would only wait out the timeout for an answer that has already arrived
+            self.log(f"Info: GatewayMQTT: discharge window {start}-{end} for {kwargs['serial']} was just refused, not retrying yet")
             return False
+        await self._send_control(start_entity, "set_discharge_slot", **kwargs)
+        return True
+
+    def _discharge_window_kwargs(self, start_entity, end_entity, start, end):
+        """The paired set_discharge_slot fields for a slot1 start/end entity pair, or None."""
+        # The end entity name is derived from the start, so both carry the same serial
+        if not isinstance(start_entity, str) or not start_entity.endswith("_discharge_slot1_start"):
+            return None
+        if end_entity != start_entity.removesuffix("_start") + "_end":
+            return None
+        serial = self._serial_from_entity_id(start_entity)
+        if not serial:
+            return None
         try:
             times = [datetime.datetime.strptime(value, "%H:%M:%S") for value in (start, end)]
         except (ValueError, TypeError):
-            return False
+            return None
         schedule = {field: value.hour * 100 + value.minute for field, value in zip(("start", "end"), times)}
-        await self._send_control(start_entity, "set_discharge_slot", schedule_json=json.dumps(schedule), serial=serial)
-        return True
+        return {"schedule_json": json.dumps(schedule), "serial": serial}
+
+    def discharge_window_refused(self, start_entity, end_entity, start, end):
+        """Whether the hub refused this exact paired window within the ack window."""
+        kwargs = self._discharge_window_kwargs(start_entity, end_entity, start, end)
+        if kwargs is None or not (getattr(self, "_ack_subscribed", False) and self._mqtt_connected and self._acks_seen):
+            return False
+        with self._command_lock:
+            entry = self._pending_commands.get(start_entity)
+            if entry is None or entry["command"] != "set_discharge_slot" or entry["kwargs"] != kwargs:
+                return False
+            return entry["state"] == "refused" and _monotonic() - entry["sent_at"] < _COMMAND_ACK_WINDOW
 
     async def _update_discharge_slot(self, entity_id, hhmm, serial):
         """Send set_discharge_slot command with updated start or end time."""
 
@@ -2969,12 +2969,18 @@ def read_pair():
             try:
                 published = run_async(gateway.update_discharge_window(*entities, *target))
             except Exception as error:
+                # Only reached without a running event loop: inside one, run_async's thread hands
+                # the exception to the thread exception hook and returns None, handled just below
                 self.base.log(f"Warn: Inverter {self.id} Hub discharge window publication failed: {error}")
                 break
             if published is not True:
                 break
-            self._poll_after_write(entities[0], lambda state: state == new_start and self.base.get_state_wrapper(entities[1], refresh=True) == new_end)
+            # A refusal is final for this cycle (the identical command is not re-sent within the
+            # ack window), so stop waiting as soon as it arrives rather than polling to the timeout
+            self._poll_after_write(entities[0], lambda state: (state == new_start and self.base.get_state_wrapper(entities[1], refresh=True) == new_end) or gateway.discharge_window_refused(*entities, *target))
             verified = read_pair() == target
+            if not verified and gateway.discharge_window_refused(*entities, *target):
+                break
             if verified:
                 self.count_register_writes += 1
                 self.registers_moved += sum(old != wanted for old, wanted in zip(before, target))
@@ -3099,6 +3105,13 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No
         if (new_start and new_start != old_start) or (new_end and new_end != old_end):
             gateway_window = self.write_and_poll_gateway_discharge_window(new_start or old_start, new_end or old_end)
             if gateway_window is False:
+                if force_export and old_discharge_enable and not getattr(self.base, "set_read_only", False):
+                    # The previous window is still armed: leaving it would keep exporting on a
+                    # schedule the plan has moved away from, so stop exporting until a write lands.
+                    # (Without force_export the schedule was already turned off above.)
+                    self.log("Warn: Inverter {} Hub discharge window write failed, turning off scheduled export".format(self.id))
+                    self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), False)
+                    self.adjust_inverter_mode(False)
                 self.record_commit("export_window", (new_start, new_end, force_export), False)
                 return
             if gateway_window is True:
 
@@ -140,6 +140,36 @@ def test_pair_applied_before_enable_and_full_serial(self):
         self.assertEqual((gw.cache[base.START], gw.cache[base.END], gw.cache[base.ENABLE]), ("02:55:00", "03:01:00", "on"))
         self.assertFalse(base.errors)
 
+    def test_paired_write_counts_one_write_and_both_moved_registers(self):
+        """The pair is one physical write that moves both endpoints."""
+        inv, gw, base, wire, mode = self.fixture(enabled="on")
+        self.apply(inv)
+        self.assertEqual([command["command"] for command in wire], ["set_discharge_slot"])
+        self.assertEqual((inv.count_register_writes, inv.registers_moved), (1, 2))
+
+    def test_refused_window_disables_an_already_exporting_schedule(self):
+        """A refused move must not leave the previous enabled window exporting."""
+        inv, gw, base, wire, mode = self.fixture(outcome="refused", start="02:00:00", end="04:00:00", enabled="on")
+        self.apply(inv, start="02:30:00", end="03:00:00")
+        self.assertEqual(json.loads(wire[0]["schedule_json"]), {"start": 230, "end": 300})
+        self.assertEqual((gw.cache[base.START], gw.cache[base.END]), ("02:00:00", "04:00:00"))
+        self.assertEqual(gw.cache[base.ENABLE], "off")
+        self.assertEqual([command.get("enable") for command in wire if command["command"] == "set_discharge_enable"], [False])
+        inv.adjust_inverter_mode.assert_called_once_with(False)
+        self.assertTrue(inv.commit_pending["export_window"])
+
+    def test_refusal_is_not_polled_to_timeout_or_resent(self):
+        """A refusal ends the attempt at once; the identical command is not re-sent or re-polled."""
+        inv, gw, base, wire, mode = self.fixture(outcome="refused")
+        inv.inv_write_and_poll_sleep = 3600  # any poll to timeout would hang the test
+        polls = []
+        original = inv._poll_after_write
+        inv._poll_after_write = lambda *args, **kwargs: polls.append(args) or original(*args, **kwargs)
+        self.apply(inv)
+        self.assertEqual(len(polls), 1)
+        self.assertEqual([command["command"] for command in wire], ["set_discharge_slot"])
+        self.assertTrue(base.errors)
+
     def test_unverified_pair_clears_ledger_and_cannot_activate(self):
         """Refusal, incomplete or staged ACKs, timeout and exceptions fail closed."""
         for outcome in ("refused", "partial", "staged", "timeout", "exception"):
@@ -166,15 +196,16 @@ def test_retry_after_refusal_can_apply_and_enable(self):
         self.assertFalse(inv.commit_pending["export_window"])
         self.assertEqual(json.loads(wire[-2]["schedule_json"]), {"start": 255, "end": 301})
 
-    def test_read_only_owner_and_serial_refuse(self):
+    def test_read_only_owner_and_mismatched_pair_refuse(self):
         """Do not publish across read-only or mismatched endpoint boundaries."""
-        for reason in ("read_only", "owner", "serial"):
+        for reason in ("read_only", "owner", "mismatched_pair"):
             inv, gw, base, wire, mode = self.fixture()
             if reason == "read_only":
                 base.set_read_only = True
             elif reason == "owner":
                 base.dashboard_index_app[base.END] = "gecloud"
             else:
+                # An end slot of another serial is not this start's pair
                 entity = base.END.replace("456789", "654321")
                 base.args["discharge_end_time"] = entity
                 base.dashboard_index_app[entity] = "gateway"