fix: keep the WebSocket pool usable after an event-loop change (#2001)

* fix: detach the WebSocket pool before cleaning it up on an event-loop change

Pooled clients belong to the loop they were created on. When the loop
changed, get_client awaited disconnect() for every stale client from the
new loop, which raises RuntimeError ("attached to a different loop").
That class was not in the best-effort catch, so it escaped before
_clients.clear() and before the _current_loop update: the pool stayed
stale, every later call re-entered the same branch, and WebSocket-backed
tools kept failing until the process restarted.

The pool and the loop reference are now detached first, and cleanup runs
without awaiting anything across loops: a still-running owning loop gets
each disconnect scheduled on it, a closed or stopped one leaves nothing
to schedule so the connection is abandoned with the loop's resources.

WebSocketManager.disconnect gets the same detach-first order for the
shutdown path, and both remaining best-effort catches around
client.disconnect() now include RuntimeError.

* fix: report scheduled stale-disconnect outcomes and pin the detach order

Review follow-ups on the loop-change fix:

- A disconnect scheduled on a still-running owning loop was fire-and-forget
  into a discarded concurrent.futures.Future, which never warns about an
  exception nobody retrieved, so a failure there was invisible at every log
  level. A done-callback now reports cancellation and failure at debug level.
- The docstring claimed an abandoned connection's socket is released with the
  loop's resources. Closing a loop does not close its transports; the socket
  goes when the orphaned transport is garbage-collected, with a
  ResourceWarning. Reworded, and the residual scheduling window (a loop that
  accepts the callback and then stops before draining it) is now named
  instead of implied to be closed.
- The disconnect() regression test asserted only the final pool state, which
  the widened except alone satisfies, so the detach-before-cleanup order was
  untested. The stub now captures the pool from inside its disconnect call and
  the test asserts it is already empty.

* fix: release pooled clients when the old loop is only stopped

A loop that is stopped but not closed still owns its transports and can
be resumed with run_forever()/run_until_complete(), so treating it like a
closed loop dropped the manager's only reference to those clients without
scheduling their disconnect: the WebSocket and its background task stayed
rooted by the retained loop while a replacement connection was opened, and
repeated loop swaps leaked one connection each.

run_coroutine_threadsafe accepts a stopped, non-closed loop; the callback
is queued and runs when the loop is resumed. Narrow the abandon condition
to loops that are closed or unknown, which are the only ones with nothing
left to schedule on.

* test: pin the proxy branch in the journald window lines cases

The two proxy cases in TestJournaldWindowLinesParam asserted on the
HA-Core proxy path by mocking mock_client.httpx_client.request, but did
not declare which branch get_addon_logs / _get_system_service_logs should
take. Both branch on is_running_in_addon(), which keys off SUPERVISOR_TOKEN
in the environment, so with that variable set the call went down the
Supervisor-direct path instead, never touched the mocked attribute, and
issued a real request to http://supervisor. CI never sets the variable, so
the failure only appears when the unit suite runs inside an add-on
container.

Take the non_addon_install fixture, which the sibling cases in this file
already use, so the branch is an explicit input on every machine (#2000).

* test: cover the stale-disconnect callback and schedule-race branches

Pin the previously untested paths of the fire-and-forget cleanup added
for the loop-change fix:

- _log_stale_disconnect on a cancelled future and on a failed one. The
  cancelled() guard has to precede exception(), since calling exception()
  on a cancelled concurrent.futures.Future raises CancelledError; without
  the guard the callback itself would raise.
- _release_stale_clients when run_coroutine_threadsafe raises RuntimeError
  (the owning loop closing between the is_closed() check and the schedule
  call). Two clients pin the continue, and the orphaned coroutines are
  closed rather than left "never awaited".

Also correct the disconnect() comment, which claimed parity with
get_client()'s loop-change branch it does not have: the sole caller runs
on the pool's own loop and awaits each disconnect, and the RuntimeError
catch there is purely defensive.

Author: Patch76

src/ha_mcp/client/websocket_client.py

@@ -8,6 +8,7 @@
 """
 
 import asyncio
+import concurrent.futures
 import hashlib
 import json
 import logging
@@ -1036,6 +1037,21 @@ def last_connect_error(self) -> str | None:
 MAX_POOL_SIZE = 50
 
 
+def _log_stale_disconnect(future: "concurrent.futures.Future[None]") -> None:
+    """Report how a disconnect scheduled on a stale event loop ended.
+
+    Nothing awaits that future, so without this the outcome would be invisible
+    at every log level: a ``concurrent.futures.Future`` never warns about an
+    exception no one retrieved.
+    """
+    if future.cancelled():
+        logger.debug("Stale WebSocket disconnect was cancelled with its loop")
+        return
+    error = future.exception()
+    if error is not None:
+        logger.debug("Stale WebSocket disconnect failed: %s", error)
+
+
 class WebSocketManager:
     """Singleton manager for Home Assistant WebSocket connections.
 
@@ -1114,6 +1130,56 @@ def _effective_verify_ssl(verify_ssl: bool | None) -> bool:
             )
             return True
 
+    @staticmethod
+    def _release_stale_clients(
+        clients: list[HomeAssistantWebSocketClient],
+        loop: asyncio.AbstractEventLoop | None,
+    ) -> None:
+        """Best-effort cleanup of clients orphaned by an event-loop change.
+
+        Never awaits: the connections belong to ``loop``, and awaiting them
+        from the loop that replaced it is exactly the cross-loop failure this
+        avoids. Each disconnect is scheduled on the owning loop and
+        deliberately not waited for. A merely stopped loop is scheduled too:
+        it still owns live transports and still accepts callbacks, so the
+        disconnect runs once that loop is resumed, and otherwise stays an
+        unrun callback like any other the abandoned loop still holds. Only a
+        closed or unknown loop has nothing left to schedule on; its
+        connection is abandoned and the socket closes when the orphaned
+        transport is garbage-collected (with a ``ResourceWarning``), because
+        closing a loop does not close the transports it carried. Either way
+        the caller has already detached the pool, so a failure here cannot
+        make it stale again.
+        """
+        if not clients:
+            return
+        if loop is None or loop.is_closed():
+            logger.debug(
+                "Abandoning %d stale WebSocket client(s): the owning event "
+                "loop is gone",
+                len(clients),
+            )
+            return
+        for client in clients:
+            coro = client.disconnect()
+            try:
+                future = asyncio.run_coroutine_threadsafe(coro, loop)
+            except RuntimeError:
+                # The loop closed between the check above and now. Closing the
+                # coroutine keeps it from surfacing as "never awaited". The
+                # narrower window stays open by construction: a loop that
+                # accepts the callback and then stops before draining it never
+                # runs the coroutine at all, and ``run_coroutine_threadsafe``
+                # needs the coroutine object up front, so there is nothing left
+                # to reclaim from this side.
+                coro.close()
+                logger.debug(
+                    "Could not schedule disconnect of a stale WebSocket client",
+                    exc_info=True,
+                )
+                continue
+            future.add_done_callback(_log_stale_disconnect)
+
     async def get_client(
         self,
         url: str | None = None,
@@ -1143,22 +1209,21 @@ async def get_client(
         if not self._lock:
             raise Exception("Lock not initialized")
         async with self._lock:
-            if self._current_loop is not None and self._current_loop != current_loop:
-                # Event loop changed — disconnect all clients
-                for client in self._clients.values():
-                    try:
-                        await client.disconnect()
-                    except (OSError, asyncio.CancelledError):
-                        # Best-effort cleanup — failure is expected when the
-                        # event loop changed and connections are stale.
-                        logger.debug(
-                            "Ignoring error disconnecting stale WebSocket client",
-                            exc_info=True,
-                        )
+            previous_loop = self._current_loop
+            stale_clients: list[HomeAssistantWebSocketClient] = []
+            if previous_loop is not None and previous_loop is not current_loop:
+                # Event loop changed: detach the pool BEFORE cleaning it up.
+                # The pooled clients' futures belong to ``previous_loop``, so
+                # awaiting their ``disconnect()`` here raises ``RuntimeError:
+                # ... attached to a different loop``, which used to escape the
+                # best-effort catch and leave both the pool and the loop
+                # reference stale for every later call (issue #1994).
+                stale_clients = list(self._clients.values())
                 self._clients.clear()
                 self._last_used.clear()
 
             self._current_loop = current_loop
+            self._release_stale_clients(stale_clients, previous_loop)
 
             # Determine credentials to use
             if url and token:
@@ -1226,7 +1291,7 @@ async def _evict_lru_if_needed(self) -> None:
         if stale:
             try:
                 await stale.disconnect()
-            except (OSError, asyncio.CancelledError):
+            except (OSError, RuntimeError, asyncio.CancelledError):
                 logger.warning(
                     "Error disconnecting evicted WebSocket client",
                     exc_info=True,
@@ -1239,16 +1304,27 @@ async def disconnect(self) -> None:
         if not self._lock:
             raise Exception("Lock not initialized")
         async with self._lock:
-            for client in self._clients.values():
+            # Detach the pool before disconnecting so a raising disconnect
+            # cannot leave it populated (the bookkeeping half of the
+            # loop-change branch in ``get_client``). The only caller,
+            # ``__main__``'s shutdown, runs on the pool's own loop, so each
+            # ``disconnect()`` is awaited to completion here. ``RuntimeError``
+            # is caught purely defensively: were this ever driven from a
+            # different loop, the cross-loop future would raise instead of
+            # hanging the shutdown. Unlike ``get_client`` it does not
+            # reschedule onto the owning loop, because the process is going
+            # away and there is nothing left to resume it.
+            clients = list(self._clients.values())
+            self._clients.clear()
+            self._last_used.clear()
+            self._current_loop = None
+            for client in clients:
                 try:
                     await client.disconnect()
-                except (OSError, asyncio.CancelledError):
+                except (OSError, RuntimeError, asyncio.CancelledError):
                     logger.warning(
                         "Error disconnecting WebSocket client", exc_info=True
                     )
-            self._clients.clear()
-            self._last_used.clear()
-            self._current_loop = None
 
 
 # Global WebSocket manager instance

tests/src/unit/test_tools_utility_supervisor_logs.py

@@ -1502,7 +1502,9 @@ async def test_addon_direct_branch_omits_query_without_lines(
         assert kwargs["params"] is None
 
     @pytest.mark.asyncio
-    async def test_addon_proxy_branch_sends_lines_query(self, mock_client):
+    async def test_addon_proxy_branch_sends_lines_query(
+        self, mock_client, non_addon_install
+    ):
         mock_response = MagicMock()
         mock_response.status_code = 200
         mock_response.text = "x\n"
@@ -1514,7 +1516,9 @@ async def test_addon_proxy_branch_sends_lines_query(self, mock_client):
         assert kwargs["params"] == {"lines": 2000}
 
     @pytest.mark.asyncio
-    async def test_system_service_proxy_branch_sends_lines_query(self, mock_client):
+    async def test_system_service_proxy_branch_sends_lines_query(
+        self, mock_client, non_addon_install
+    ):
         mock_response = MagicMock()
         mock_response.status_code = 200
         mock_response.text = "x\n"