@@ -2654,6 +2654,44 @@ async def _update_charge_slot(self, entity_id, hhmm, serial):
         await self._send_control(entity_id, "set_charge_slot", schedule_json=json.dumps(schedule), serial=serial)
         self.log(f"Info: GatewayMQTT: Charge slot update: {schedule}")
 
+    def discharge_window_uses_staging(self, start_entity, end_entity):
+        """Identify a coordinator synchronously, before crossing an async bridge."""
+        if not isinstance(start_entity, str) or not start_entity.endswith("_discharge_slot1_start"):
+            return False
+        if end_entity != start_entity.removesuffix("_start") + "_end":
+            return False
+        serial = self._serial_from_entity_id(start_entity)
+        record = getattr(self, "_discovery", {}).get((serial or "").upper(), {})
+        return bool(serial) and record.get("type") in GATEWAY_COORDINATOR_TYPES
+
+    async def update_discharge_window(self, start_entity, end_entity, start, end):
+        """Send an engine discharge window as one full-serial command, not two partial slots.
+
+        True means publication was attempted, not that the inverter applied it. The
+        caller must verify both endpoint entities, which only an applied ACK or
+        physical telemetry updates. Coordinators use their existing separate path:
+        those devices can stage a disabled slot and apply it only on enable.
+        Manual single-endpoint selects keep their path.
+        """
+        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":
+            return False
+        serial = self._serial_from_entity_id(start_entity)
+        if not serial or self._serial_from_entity_id(end_entity) != serial:
+            return False
+        if self.discharge_window_uses_staging(start_entity, end_entity):
+            return False
+        try:
+            times = [datetime.datetime.strptime(value, "%H:%M:%S") for value in (start, end)]
+        except (ValueError, TypeError):
+            return False
+        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
+
     async def _update_discharge_slot(self, entity_id, hhmm, serial):
         """Send set_discharge_slot command with updated start or end time."""
         if "_start" in entity_id:
 
@@ -2927,6 +2927,74 @@ def adjust_idle_time(self, charge_start=None, charge_end=None, discharge_start=N
                 self.write_and_poll_option("idle_end_time", idle_end_time_id, idle_end)
                 self.idle_end_minutes = idle_end_minutes
 
+    def write_and_poll_gateway_discharge_window(self, new_start, new_end):
+        """Verify a complete Hub slot together, or return None for another provider.
+
+        A start equal to the old end cannot be sent alone while discharging is enabled:
+        it temporarily disables the slot and the Hub correctly refuses it. Sending
+        both bounds also handles swapping the bounds of a midnight-spanning slot.
+        """
+        from ha import run_async
+
+        entities = [self.base.get_arg(name, indirect=False, index=self.id) for name in ("discharge_start_time", "discharge_end_time")]
+        owners = [getattr(self.base, "dashboard_index_app", {}).get(entity) for entity in entities]
+        if "gateway" not in owners:
+            return None
+        components = getattr(self.base, "components", None)
+        gateway = components.get_component("gateway") if components else None
+        names = ("discharge_start_time", "discharge_end_time")
+        target = (new_start, new_end)
+        if owners != ["gateway", "gateway"] or gateway is None:
+            self.base.record_status(f"Warn: Inverter {self.id} Hub discharge window has inconsistent entity ownership", had_errors=True)
+            return False
+        if not all(self.check_write_entity("write_and_poll_gateway_discharge_window", name, entity, value) for name, entity, value in zip(names, entities, target)):
+            return False
+        if gateway.discharge_window_uses_staging(*entities):
+            return None  # Coordinator slot staging must precede its enable write.
+
+        def read_pair():
+            """Read both endpoint entities without substituting desired values."""
+            return tuple(self.base.get_state_wrapper(entity, refresh=True) for entity in entities)
+
+        before = read_pair()
+        ledger = self.base.control_ledger
+        if ledger is not None:
+            for name, entity, value, wanted in zip(names, entities, before, target):
+                self._ledger_observe(ledger, name, entity, value)
+                if value != wanted:
+                    ledger.note_write_attempt(entity)
+        key = "|".join(entities)
+        verified = before == target
+        for _retry in range(0 if verified else self._write_attempts("discharge_window", key, target)):
+            try:
+                published = run_async(gateway.update_discharge_window(*entities, *target))
+            except Exception as error:
+                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)
+            verified = read_pair() == target
+            if verified:
+                self.count_register_writes += 1
+                self.registers_moved += sum(old != wanted for old, wanted in zip(before, target))
+                break
+        self._write_backoff_result("discharge_window", key, target, verified)
+        if ledger is not None:
+            for name, entity, value in zip(names, entities, target):
+                if verified:
+                    if before == target:
+                        ledger.record_ownership_from_read(entity, name, value, now=time.time(), generation=self._ledger_generation(entity))
+                    else:
+                        ledger.record_write(entity, name, value, now=time.time(), generation=self._ledger_generation(entity))
+                else:
+                    ledger.clear(entity)
+        if not verified:
+            message = f"Warn: Inverter {self.id} write to Hub discharge window failed: wanted {target}, got {read_pair()}"
+            self.base.log(message)
+            self.base.record_status(message, had_errors=True)
+        return verified
+
     def adjust_force_export(self, force_export, new_start_time=None, new_end_time=None):
         """
         Adjust force export on/off and set the time window correctly
@@ -3024,8 +3092,20 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No
         else:
             self.adjust_idle_time(discharge_start="00:00:00", discharge_end="00:00:00")
 
+        # Direct Hub batteries accept the complete slot as one physical write. Do not
+        # enable or enter export mode after a refused, staged or incomplete read-back.
+        # Coordinators retain their existing staged-endpoint then enable sequence.
+        gateway_window = None
+        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:
+                self.record_commit("export_window", (new_start, new_end, force_export), False)
+                return
+            if gateway_window is True:
+                changed_start_end = True
+
         # Change start time
-        if new_start and (new_start != old_start or is_hm_format):
+        if gateway_window is None and new_start and (new_start != old_start or is_hm_format):
             self.base.log("Inverter {} set new export start time to {}".format(self.id, new_start))
             if "discharge_start_time" in self.base.args:
                 # Always write to this as it is the GE default
@@ -3047,7 +3127,7 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No
                 self.log("Warn: Inverter {} unable write export start time as neither REST or discharge_start_time are set".format(self.id))
 
         # Change end time
-        if new_end and (new_end != old_end or is_hm_format):
+        if gateway_window is None and new_end and (new_end != old_end or is_hm_format):
             self.base.log("Inverter {} Set new export end time to {} was {}".format(self.id, new_end, old_end))
             if "discharge_end_time" in self.base.args:
                 # Always write to this as it is the GE default
 
@@ -7404,8 +7404,10 @@ def test_counters_restart_at_local_midnight(self):
 def run_gateway_tests(my_predbat=None):
     """Run all GatewayMQTT tests. Returns True on failure, False on success."""
     from tests.test_gateway_token_refresh import TestIsAuthFailure, TestApplyRefreshResponse, TestMaybeRefreshOnAuthError
+    from tests.test_gateway_discharge_window import TestGatewayDischargeWindow
 
     test_classes = [
+        TestGatewayDischargeWindow,
         TestProtobufDecode,
         TestPlanSerialization,
         TestCommandFormat,