@@ -2644,12 +2644,22 @@ async def select_event(self, entity_id, value):
                 # Read current charge slot times to send both start and end
                 await self._update_charge_slot(entity_id, hhmm, serial=serial)
 
+    def charge_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("_charge_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_charge_window(self, start_entity, end_entity, start, end):
         """Send an engine charge 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. None preserves the existing coordinator path:
+        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.
         """
@@ -2662,9 +2672,8 @@ async def update_charge_window(self, start_entity, end_entity, start, end):
         serial = self._serial_from_entity_id(start_entity)
         if not serial or self._serial_from_entity_id(end_entity) != serial:
             return False
-        record = getattr(self, "_discovery", {}).get(serial.upper(), {})
-        if record.get("type") in GATEWAY_COORDINATOR_TYPES:
-            return None
+        if self.charge_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):
 
@@ -3507,6 +3507,8 @@ def write_and_poll_gateway_charge_window(self, new_start, new_end):
             return False
         if not all(self.check_write_entity("write_and_poll_gateway_charge_window", name, entity, value) for name, entity, value in zip(names, entities, target)):
             return False
+        if gateway.charge_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."""
@@ -3527,9 +3529,7 @@ def read_pair():
             except Exception as error:
                 self.base.log(f"Warn: Inverter {self.id} Hub charge window publication failed: {error}")
                 break
-            if published is None:
-                return None  # Coordinator slot staging must precede its enable write.
-            if not published:
+            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
 
@@ -1,11 +1,12 @@
 """Joined engine window, Hub JSON and applied-ACK regressions."""
+# cspell:ignore excepthook
 
 import asyncio
 import datetime
 import json
 import types
 import unittest
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 from gateway import GatewayMQTT
 from inverter import Inverter
@@ -150,6 +151,23 @@ def test_success_records_both_controls_and_no_redundant_second_command(self):
         self.apply(inverter)
         self.assertEqual(len(wire), 1)
 
+    def test_async_timer_publication_exception_never_becomes_coordinator_fallback(self):
+        """RunThread returns None on error; that must remain a failed direct write."""
+        inverter, gateway, base, wire = self.fixture(outcome="exception")
+
+        async def timer():
+            """Match the standalone timer calling synchronous execution on its loop."""
+            self.apply(inverter)
+
+        with patch("threading.excepthook") as thread_error:
+            asyncio.run(timer())
+        thread_error.assert_called_once()
+        self.assertEqual(len(wire), 1)
+        self.assertTrue(base.errors)
+        inverter.write_and_poll_option.assert_not_called()
+        inverter.write_and_poll_switch.assert_not_called()
+        self.assertFalse(any("Updated start and end" in line for line in base.logs))
+
     def test_read_only_or_mismatched_entity_ownership_refuses_without_publication(self):
         """Never route a Cloud endpoint or user read-only window to the Hub."""
         for reason in ("read_only", "owner", "serial"):