fix(gateway): dispatch MQTT publishes onto the owning event loop

Every gateway control write (set_reserve, set_charge_rate, slot times, ...)
reaches GatewayMQTT via ha.py::run_async(), which runs the call chain on a
fresh, throwaway event loop distinct from the persistent loop that owns
GatewayMQTT's aiomqtt Client. Calling client.publish() from that other loop
binds the QoS-1 publish-confirmation Future to the wrong loop, so every
control write stalls for the client's ~10s default timeout before
completing — confirmed with a live reproduction against a local broker
(10.001s cross-loop vs 0.001s same-loop).

_publish_raw() now detects when it's running on a different loop than the
one captured from run() and hands the actual publish off via
run_coroutine_threadsafe, so it executes on the owning loop instead.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: mgazza

apps/predbat/gateway.py

@@ -244,6 +244,13 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt
         self._mqtt_client = None
         self._mqtt_task = None
         self._mqtt_connected = False
+        # Event loop that owns self._mqtt_client (captured in run()). Control writes
+        # arrive via ha.py::run_async(), which runs on its own throwaway loop on the
+        # calling thread — publishing directly from there would bind the aiomqtt
+        # publish-confirmation Future to the wrong loop and stall for the client's
+        # ~10s default timeout instead of completing near-instantly. _publish_raw()
+        # dispatches onto this loop when called from elsewhere.
+        self._loop = None
         self._gateway_online = False
         self._last_telemetry_time = 0
         self._last_plan_data = None
@@ -536,6 +543,9 @@ async def run(self, seconds, first):
             return False
 
         if first:
+            # Capture the loop that will own the MQTT client/listener task, so
+            # cross-loop callers (ha.py::run_async()) can dispatch onto it later.
+            self._loop = asyncio.get_running_loop()
             # Start MQTT listener as a background task
             self._mqtt_task = asyncio.ensure_future(self._mqtt_loop())
             self.log("Info: GatewayMQTT: MQTT listener task started")
@@ -1622,12 +1632,30 @@ async def publish_command(self, command, **kwargs):
     async def _publish_raw(self, topic, payload, retain=False):
         """Publish raw bytes to an MQTT topic.
 
+        Must run the actual client.publish() on the event loop that owns
+        self._mqtt_client (self._loop). Callers reached via ha.py::run_async()
+        (i.e. every control write issued from the synchronous engine thread)
+        run on a different, throwaway loop — publishing directly from there
+        binds aiomqtt's publish-confirmation Future to the wrong loop and
+        stalls for the client's ~10s default timeout instead of completing
+        near-instantly. When we're already on the owning loop (e.g. internal
+        periodic publishes from run()/housekeeping), publish directly.
+
         Args:
             topic: MQTT topic string.
             payload: Bytes to publish.
             retain: Whether to set the retain flag.
         """
-        if self._mqtt_client and self._mqtt_connected:
+        if not (self._mqtt_client and self._mqtt_connected):
+            return
+
+        if self._loop is not None and self._loop is not asyncio.get_running_loop():
+            future = asyncio.run_coroutine_threadsafe(
+                self._mqtt_client.publish(topic, payload, qos=1, retain=retain),
+                self._loop,
+            )
+            await asyncio.wrap_future(future)
+        else:
             await self._mqtt_client.publish(topic, payload, qos=1, retain=retain)
 
     def is_alive(self):

apps/predbat/tests/test_gateway.py

@@ -4440,6 +4440,51 @@ def test_non_finite_returns_none(self):
         self.assertIsNone(extract_rate_anchors(8.0, float("inf"), 15.0, 12.0))
 
 
+class TestPublishRawLoopSafety:
+    """_publish_raw() must run the actual client.publish() on the event loop that
+    owns the aiomqtt Client (self._loop), even when invoked from a different loop —
+    otherwise the publish confirmation Future is bound to the wrong loop and stalls
+    for aiomqtt's ~10s default timeout instead of completing near-instantly. This is
+    what happens today when a control write reaches GatewayMQTT via
+    ha.py::run_async(), which runs the whole call chain on a fresh, throwaway
+    event loop distinct from GatewayMQTT's own persistent MQTT loop.
+    """
+
+    def test_publish_dispatches_to_owning_loop_when_called_cross_loop(self):
+        """A cross-loop _publish_raw() call must execute client.publish() on the
+        thread that owns self._loop, not on the calling thread."""
+        import asyncio
+        import threading
+        from gateway import GatewayMQTT
+
+        gw = GatewayMQTT.__new__(GatewayMQTT)
+        gw._mqtt_connected = True
+        observed = {}
+
+        class FakeClient:
+            async def publish(self, topic, payload, qos=0, retain=False):
+                observed["thread_ident"] = threading.get_ident()
+
+        gw._mqtt_client = FakeClient()
+
+        # Real second event loop on its own thread — mimics GatewayMQTT's
+        # persistent MQTT loop (owned by a dedicated thread via hass.py::create_task).
+        owner_loop = asyncio.new_event_loop()
+        owner_thread = threading.Thread(target=owner_loop.run_forever, daemon=True)
+        owner_thread.start()
+        gw._loop = owner_loop
+
+        try:
+            # Call _publish_raw from a DIFFERENT, freshly-created loop on THIS
+            # thread — mirrors ha.py::run_async()'s asyncio.run(coro) pattern.
+            asyncio.run(gw._publish_raw("predbat/devices/test/command", b"payload"))
+        finally:
+            owner_loop.call_soon_threadsafe(owner_loop.stop)
+            owner_thread.join(timeout=2)
+
+        assert observed.get("thread_ident") == owner_thread.ident, "client.publish() ran on the wrong thread/loop"
+
+
 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
@@ -4476,6 +4521,7 @@ def run_gateway_tests(my_predbat=None):
         TestApplyRefreshResponse,
         TestMaybeRefreshOnAuthError,
         TestRateAnchors,
+        TestPublishRawLoopSafety,
     ]
     for cls in test_classes:
         instance = cls()