From 4046a9ae26f90d06370b7354059bc73c272eece1 Mon Sep 17 00:00:00 2001
From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com>
Date: Thu, 17 Sep 2026 03:47:30 +0200
Subject: [PATCH 1/6] Stop using the EZSP gateway once its transport has closed

When the serial/TCP transport closes, `Gateway.connection_lost()` resolves
the "connection done" future, which stops the gateway's worker thread and
closes its event loop. Anything the application still routes through the
`ThreadsafeProxy` after that point lands on a dead loop: a bare `None` that
is then awaited (`TypeError: 'NoneType' object can't be awaited`), a
dispatch that never runs, or a `CancelledError` from the teardown
cancelling the dispatched task.

`EZSP.startup_reset()` made this common on network coordinators: a
connection dropped while waiting for the startup reset frame raised the
transport error on the first attempt, and the two retries then went to the
torn-down gateway -- a two-second hang, `TypeError`s from `reset()` and
`wait_for_startup_reset()`, and finally `disconnect()` itself failing the
same way, masking the real error all the way up through
`ControllerApplication.connect()` and `shutdown()`.

`EZSP` now records that its transport closed (`Gateway.connection_lost()`
only -- an NCP failed state keeps the transport open and is still reset
and closed normally) and:

- `startup_reset()` stops retrying and raises the transport's error;
  a `CancelledError` caused by the worker teardown (not by the caller) is
  reported as a `ConnectionResetError` so setup fails instead of looking
  cancelled
- `disconnect()` no longer calls into the gateway, which has nothing left
  to close, and always drops the reference
- `_command()` fails immediately instead of waiting out the command
  timeout

`ControllerApplication.connect()` and `disconnect()` drop the `EZSP`
reference even if disconnecting fails.
---
 bellows/ezsp/__init__.py      |  55 ++++++++--
 bellows/zigbee/application.py |  12 ++-
 tests/test_application.py     |  25 +++++
 tests/test_ezsp.py            | 190 ++++++++++++++++++++++++++++++++++
 4 files changed, 272 insertions(+), 10 deletions(-)

diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py
index a158018e..9f786d38 100644
--- a/bellows/ezsp/__init__.py
+++ b/bellows/ezsp/__init__.py
@@ -78,6 +78,11 @@ def __init__(self, device_config: dict, application: Any | None = None):
         self._ezsp_version = v4.EZSPv4.VERSION
         self._xncp_features = FirmwareFeatures.NONE
         self._gw = None
+        # Set once the gateway reports that its transport closed. The gateway then has
+        # nothing left to close and, when it runs in its own thread, its event loop is
+        # being torn down: nothing may be sent through it and nothing needs to be
+        # awaited on it.
+        self._transport_closed = False
         self._protocol = None
         self._application = application
 
@@ -148,8 +153,23 @@ async def startup_reset(self) -> None:
             try:
                 await self._startup_reset()
                 break
+            except asyncio.CancelledError:
+                task = asyncio.current_task()
+                if not self._transport_closed or (
+                    task is not None and task.cancelling()
+                ):
+                    raise
+
+                # A gateway in its own thread cancels the work dispatched to it when its
+                # transport closes: a lost connection, not our caller cancelling us
+                await self.disconnect()
+                raise ConnectionResetError(
+                    "Connection was lost during startup"
+                ) from None
             except Exception as exc:
-                if attempt + 1 < RESET_ATTEMPTS:
+                # Retrying is for an NCP that did not answer. Once the transport itself
+                # is gone, every further attempt would go to a dead gateway.
+                if attempt + 1 < RESET_ATTEMPTS and not self._transport_closed:
                     LOGGER.debug(
                         "EZSP startup/reset failed, retrying (%d/%d): %r",
                         attempt + 1,
@@ -163,6 +183,7 @@ async def startup_reset(self) -> None:
 
     async def connect(self, *, use_thread: bool = True) -> None:
         assert self._gw is None
+        self._transport_closed = False
         self._gw = await bellows.uart.connect(self._config, self, use_thread=use_thread)
         await self.startup_reset()
 
@@ -227,13 +248,29 @@ async def get_xncp_features(self) -> xncp.FirmwareFeatures:
 
     async def disconnect(self):
         self.stop_ezsp()
-        if self._gw:
-            await self._gw.disconnect()
+        if self._gw is None:
+            return
+
+        try:
+            # A gateway whose transport already closed has nothing left to close, and
+            # its worker loop may be gone: don't call into it
+            if not self._transport_closed:
+                await self._gw.disconnect()
+        finally:
             self._gw = None
 
     async def _command(self, name: str, *args: Any, **kwargs: Any) -> Any:
         command = getattr(self._protocol, name)
 
+        if self._transport_closed:
+            LOGGER.debug(
+                "Couldn't send command %s(%s, %s). Connection was lost",
+                name,
+                args,
+                kwargs,
+            )
+            raise EzspError("Connection was lost")
+
         if not self.is_ezsp_running:
             LOGGER.debug(
                 "Couldn't send command %s(%s, %s). EZSP is not running",
@@ -330,12 +367,18 @@ async def leaveNetwork(self, timeout: float | int = NETWORK_OPS_TIMEOUT) -> None
 
     def connection_lost(self, exc):
         """Lost serial connection."""
-        if self._application is not None:
-            self._application.connection_lost(exc)
+        self._transport_closed = True
+        self._notify_connection_lost(exc)
 
     def enter_failed_state(self, code: t.NcpResetCode) -> None:
         """UART received reset code."""
-        self.connection_lost(NcpFailure(code=code))
+        # The transport is still open here: a reset can recover the NCP, and
+        # `disconnect()` still has a transport to close
+        self._notify_connection_lost(NcpFailure(code=code))
+
+    def _notify_connection_lost(self, exc: Exception) -> None:
+        if self._application is not None:
+            self._application.connection_lost(exc)
 
     def __getattr__(self, name: str) -> Callable:
         if name not in self._protocol.COMMANDS:
diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py
index c16f43eb..a814c917 100644
--- a/bellows/zigbee/application.py
+++ b/bellows/zigbee/application.py
@@ -195,8 +195,10 @@ async def connect(self) -> None:
             await self.register_endpoints()
         except Exception:
             if self._ezsp is not None:
-                await self._ezsp.disconnect()
-                self._ezsp = None
+                try:
+                    await self._ezsp.disconnect()
+                finally:
+                    self._ezsp = None
             raise
 
     async def _ensure_network_running(self) -> bool:
@@ -612,8 +614,10 @@ async def disconnect(self):
         # TODO: how do you shut down the stack?
         self.controller_event.clear()
         if self._ezsp is not None:
-            await self._ezsp.disconnect()
-            self._ezsp = None
+            try:
+                await self._ezsp.disconnect()
+            finally:
+                self._ezsp = None
 
     async def force_remove(self, dev):
         # This should probably be delivered to the parent device instead
diff --git a/tests/test_application.py b/tests/test_application.py
index 784f86c8..b0387b03 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2194,6 +2194,31 @@ async def test_connect_failure(app: ControllerApplication) -> None:
     assert len(ezsp.disconnect.mock_calls) == 1
 
 
+async def test_connect_failure_disconnect_failure(app: ControllerApplication) -> None:
+    """Test that EZSP is dropped even when disconnecting after a failure fails."""
+    ezsp = app._ezsp
+    app._ezsp.write_config = AsyncMock(side_effect=OSError())
+    app._ezsp.connect = AsyncMock()
+    app._ezsp.disconnect = AsyncMock(side_effect=RuntimeError("Uh oh"))
+    app._ezsp = None
+
+    with patch("bellows.ezsp.EZSP", return_value=ezsp):
+        with pytest.raises(RuntimeError):
+            await app.connect()
+
+    assert app._ezsp is None
+
+
+async def test_disconnect_failure(app: ControllerApplication) -> None:
+    """Test that EZSP is dropped even when disconnecting fails."""
+    app._ezsp.disconnect = AsyncMock(side_effect=RuntimeError("Uh oh"))
+
+    with pytest.raises(RuntimeError):
+        await app.disconnect()
+
+    assert app._ezsp is None
+
+
 async def test_repair_tclk_partner_ieee(
     app: ControllerApplication, ieee: zigpy_t.EUI64
 ) -> None:
diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py
index e2e26e1d..75089e77 100644
--- a/tests/test_ezsp.py
+++ b/tests/test_ezsp.py
@@ -4,6 +4,7 @@
 from asyncio import timeout as asyncio_timeout
 import functools
 import logging
+import threading
 from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 import pytest
@@ -79,6 +80,54 @@ async def test_disconnect(ezsp_f):
     gw_disconnect = ezsp_f._gw.disconnect
     await ezsp_f.disconnect()
     assert len(gw_disconnect.mock_calls) == 1
+    assert ezsp_f._gw is None
+
+
+async def test_disconnect_after_transport_closed(ezsp_f):
+    """A gateway whose transport closed is not called into: it has nothing to close."""
+    ezsp_f._application = MagicMock()
+    gw_disconnect = ezsp_f._gw.disconnect
+    exc = ConnectionResetError("Remote server closed connection")
+
+    ezsp_f.connection_lost(exc)
+    await ezsp_f.disconnect()
+
+    assert gw_disconnect.mock_calls == []
+    assert ezsp_f._gw is None
+    assert ezsp_f._application.connection_lost.mock_calls == [call(exc)]
+
+
+async def test_disconnect_after_failed_state(ezsp_f):
+    """An NCP failure leaves the transport open, so disconnecting still closes it."""
+    ezsp_f._application = MagicMock()
+    gw_disconnect = ezsp_f._gw.disconnect
+
+    ezsp_f.enter_failed_state(t.NcpResetCode.RESET_SOFTWARE)
+    await ezsp_f.disconnect()
+
+    assert len(gw_disconnect.mock_calls) == 1
+    assert ezsp_f._gw is None
+
+
+async def test_disconnect_drops_gateway_on_failure(ezsp_f):
+    """The gateway reference is dropped even when closing it fails."""
+    ezsp_f._gw.disconnect = AsyncMock(side_effect=RuntimeError("Uh oh"))
+
+    with pytest.raises(RuntimeError):
+        await ezsp_f.disconnect()
+
+    assert ezsp_f._gw is None
+
+
+async def test_command_after_transport_closed(ezsp_f):
+    """Commands fail immediately once the transport closed, instead of timing out."""
+    ezsp_f._protocol = MagicMock()
+    ezsp_f.connection_lost(ConnectionResetError("Remote server closed connection"))
+
+    with pytest.raises(EzspError, match="Connection was lost"):
+        await EZSP._command(ezsp_f, "version")
+
+    assert ezsp_f._protocol.version.mock_calls == []
 
 
 def test_attr(ezsp_f):
@@ -326,6 +375,147 @@ async def startup_reset_mock():
     assert disconnect_mock.call_count == 0
 
 
+async def test_ezsp_connect_no_retry_after_transport_closed():
+    """A startup reset that failed because the transport closed is not retried."""
+    exc = ConnectionResetError("Remote server closed connection")
+
+    with patch("bellows.uart.connect") as conn_mock:
+        ezsp = make_ezsp(version=4)
+        ezsp._transport_closed = True  # stale state from an earlier connection
+
+        async def startup_reset_mock():
+            # The gateway reports the loss before the failure reaches the caller
+            ezsp.connection_lost(exc)
+            raise exc
+
+        with patch.object(
+            ezsp, "_startup_reset", side_effect=startup_reset_mock
+        ) as startup_reset:
+            with pytest.raises(ConnectionResetError):
+                await ezsp.connect()
+
+    assert startup_reset.await_count == 1
+    assert conn_mock.return_value.disconnect.mock_calls == []
+    assert ezsp._gw is None
+
+
+async def test_ezsp_connect_cancelled_by_transport_closed():
+    """A threaded gateway cancels dispatched work when its transport closes.
+
+    That cancellation is reported as a lost connection, not propagated as-is.
+    """
+    with patch("bellows.uart.connect") as conn_mock:
+        ezsp = make_ezsp(version=4)
+
+        async def startup_reset_mock():
+            ezsp.connection_lost(
+                ConnectionResetError("Remote server closed connection")
+            )
+            raise asyncio.CancelledError()
+
+        with patch.object(
+            ezsp, "_startup_reset", side_effect=startup_reset_mock
+        ) as startup_reset:
+            with pytest.raises(ConnectionResetError, match="lost during startup"):
+                await ezsp.connect()
+
+    assert startup_reset.await_count == 1
+    assert conn_mock.return_value.disconnect.mock_calls == []
+    assert ezsp._gw is None
+
+
+async def test_ezsp_connect_cancelled_without_transport_closed():
+    """A cancellation not explained by a closed transport propagates as-is."""
+    with patch("bellows.uart.connect"):
+        ezsp = make_ezsp(version=4)
+
+        with patch.object(ezsp, "_startup_reset", side_effect=asyncio.CancelledError()):
+            with pytest.raises(asyncio.CancelledError):
+                await ezsp.connect()
+
+
+async def test_ezsp_connect_cancelled_by_caller_after_transport_closed():
+    """The caller cancelling `connect()` still wins, even after the transport closed."""
+    loop = asyncio.get_running_loop()
+    blocked = loop.create_future()
+
+    with patch("bellows.uart.connect"):
+        ezsp = make_ezsp(version=4)
+
+        async def startup_reset_mock():
+            ezsp.connection_lost(
+                ConnectionResetError("Remote server closed connection")
+            )
+            blocked.set_result(None)
+            await loop.create_future()
+
+        with patch.object(ezsp, "_startup_reset", side_effect=startup_reset_mock):
+            connect_task = asyncio.ensure_future(ezsp.connect())
+            await blocked
+            connect_task.cancel()
+
+            with pytest.raises(asyncio.CancelledError):
+                await connect_task
+
+
+async def test_ezsp_connect_transport_closed_during_startup_reset():
+    """The bridge drops the TCP connection while EZSP waits for the startup reset.
+
+    Connecting fails with the transport's error right away: no retry against the
+    torn-down gateway, no call into its worker loop, and the worker thread exits.
+    """
+    loop = asyncio.get_running_loop()
+    waiting_for_reset = loop.create_future()
+    clients = []
+
+    async def handle_client(reader, writer):
+        clients.append(writer)
+
+    server = await asyncio.start_server(handle_client, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+
+    wait_for_startup_reset = uart.Gateway.wait_for_startup_reset
+
+    async def wait_for_startup_reset_mock(self):
+        # Runs on the worker loop
+        loop.call_soon_threadsafe(waiting_for_reset.set_result, None)
+        await wait_for_startup_reset(self)
+
+    ezsp = EZSP(
+        {
+            **DEVICE_CONFIG,
+            zigpy.config.CONF_DEVICE_PATH: f"socket://127.0.0.1:{port}",
+            zigpy.config.CONF_DEVICE_FLOW_CONTROL: None,
+        },
+        application=MagicMock(),
+    )
+
+    try:
+        with patch.object(
+            uart.Gateway, "wait_for_startup_reset", wait_for_startup_reset_mock
+        ):
+            connect_task = asyncio.ensure_future(ezsp.connect(use_thread=True))
+
+            async with asyncio_timeout(5):
+                await waiting_for_reset
+
+            # The bridge drops the connection
+            clients[0].close()
+
+            with pytest.raises(ConnectionResetError):
+                async with asyncio_timeout(1):
+                    await connect_task
+    finally:
+        server.close()
+        await server.wait_closed()
+
+    assert ezsp._gw is None
+    assert ezsp._application.connection_lost.mock_calls == [call(ANY)]
+
+    [t.join(1) for t in threading.enumerate() if "bellows" in t.name]
+    assert [t for t in threading.enumerate() if "bellows" in t.name] == []
+
+
 async def test_ezsp_newer_version(ezsp_f):
     """Test newer version of ezsp."""
     with patch.object(

From dde9252e86ca3c695b24dfe38fa8223cfd8daaf0 Mon Sep 17 00:00:00 2001
From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com>
Date: Thu, 17 Sep 2026 04:13:59 +0200
Subject: [PATCH 2/6] Don't dispatch to a closed transport that was reported
 late or early

Two windows around the startup retry loop, from review:

- The transport-closed notification from a threaded gateway is queued onto
  this loop with `call_soon_threadsafe()`. A proxy call that fails
  synchronously (the check-then-dispatch window of `ThreadsafeProxy`) can
  reach `startup_reset()`'s `except` clause before that notification is
  delivered, so the retry guard would not see it. Yield once before deciding.
- If the transport closes between `uart.connect()` returning and the first
  attempt, or `startup_reset()` is entered with the flag already set, the
  attempt itself would dispatch into the dead gateway. Check the flag at the
  top of every attempt.

`ControllerApplication.connect()` now logs a failing follow-up disconnect
instead of letting it replace the connection error, which is the masking
this series is about. The end-to-end test synchronises on the server side
accepting the client and cancels the connect task on failure.
---
 bellows/ezsp/__init__.py      | 13 +++++++
 bellows/zigbee/application.py |  5 +++
 tests/test_application.py     | 11 ++++--
 tests/test_ezsp.py            | 73 ++++++++++++++++++++++++++++++-----
 4 files changed, 88 insertions(+), 14 deletions(-)

diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py
index 9f786d38..26bdb4dd 100644
--- a/bellows/ezsp/__init__.py
+++ b/bellows/ezsp/__init__.py
@@ -148,12 +148,19 @@ async def _startup_reset(self) -> None:
 
     async def startup_reset(self) -> None:
         for attempt in range(RESET_ATTEMPTS):
+            if self._transport_closed:
+                await self.disconnect()
+                raise ConnectionResetError("Connection was lost during startup")
+
             self._protocol = v4.EZSPv4(self.handle_callback, self._gw)
 
             try:
                 await self._startup_reset()
                 break
             except asyncio.CancelledError:
+                # See below
+                await asyncio.sleep(0)
+
                 task = asyncio.current_task()
                 if not self._transport_closed or (
                     task is not None and task.cancelling()
@@ -167,6 +174,12 @@ async def startup_reset(self) -> None:
                     "Connection was lost during startup"
                 ) from None
             except Exception as exc:
+                # A gateway in its own thread reports a closed transport by queueing
+                # `connection_lost()` onto this loop. The failure it caused may reach us
+                # first, without a chance to run anything else: yield once so the
+                # notification is delivered before deciding what to do.
+                await asyncio.sleep(0)
+
                 # Retrying is for an NCP that did not answer. Once the transport itself
                 # is gone, every further attempt would go to a dead gateway.
                 if attempt + 1 < RESET_ATTEMPTS and not self._transport_closed:
diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py
index a814c917..f9178fed 100644
--- a/bellows/zigbee/application.py
+++ b/bellows/zigbee/application.py
@@ -197,6 +197,11 @@ async def connect(self) -> None:
             if self._ezsp is not None:
                 try:
                     await self._ezsp.disconnect()
+                except Exception:
+                    # Don't let cleanup failures mask why connecting failed
+                    LOGGER.warning(
+                        "Failed to disconnect after a connection failure", exc_info=True
+                    )
                 finally:
                     self._ezsp = None
             raise
diff --git a/tests/test_application.py b/tests/test_application.py
index b0387b03..b1a74703 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2194,19 +2194,22 @@ async def test_connect_failure(app: ControllerApplication) -> None:
     assert len(ezsp.disconnect.mock_calls) == 1
 
 
-async def test_connect_failure_disconnect_failure(app: ControllerApplication) -> None:
-    """Test that EZSP is dropped even when disconnecting after a failure fails."""
+async def test_connect_failure_disconnect_failure(
+    app: ControllerApplication, caplog
+) -> None:
+    """Test that a failing disconnect after a connection failure doesn't mask it."""
     ezsp = app._ezsp
-    app._ezsp.write_config = AsyncMock(side_effect=OSError())
+    app._ezsp.write_config = AsyncMock(side_effect=OSError("Connection failed"))
     app._ezsp.connect = AsyncMock()
     app._ezsp.disconnect = AsyncMock(side_effect=RuntimeError("Uh oh"))
     app._ezsp = None
 
     with patch("bellows.ezsp.EZSP", return_value=ezsp):
-        with pytest.raises(RuntimeError):
+        with pytest.raises(OSError, match="Connection failed"):
             await app.connect()
 
     assert app._ezsp is None
+    assert "Failed to disconnect after a connection failure" in caplog.text
 
 
 async def test_disconnect_failure(app: ControllerApplication) -> None:
diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py
index 75089e77..055a8004 100644
--- a/tests/test_ezsp.py
+++ b/tests/test_ezsp.py
@@ -399,6 +399,56 @@ async def startup_reset_mock():
     assert ezsp._gw is None
 
 
+async def test_ezsp_connect_no_retry_with_queued_connection_lost():
+    """The transport-closed notification may still be queued when the attempt fails.
+
+    A threaded gateway queues `connection_lost()` onto this loop; a proxy call that
+    fails synchronously reaches `startup_reset()` before it is delivered.
+    """
+    loop = asyncio.get_running_loop()
+    exc = ConnectionResetError("Remote server closed connection")
+
+    with patch("bellows.uart.connect") as conn_mock:
+        ezsp = make_ezsp(version=4)
+
+        async def startup_reset_mock():
+            # Queued, not yet delivered
+            loop.call_soon(ezsp.connection_lost, exc)
+            raise TypeError("'NoneType' object can't be awaited")
+
+        with patch.object(
+            ezsp, "_startup_reset", side_effect=startup_reset_mock
+        ) as startup_reset:
+            with pytest.raises(TypeError):
+                await ezsp.connect()
+
+    assert startup_reset.await_count == 1
+    assert conn_mock.return_value.disconnect.mock_calls == []
+    assert ezsp._gw is None
+
+
+async def test_ezsp_connect_transport_closed_before_startup_reset():
+    """A transport that closed before the first attempt is not dispatched to at all."""
+    with patch("bellows.uart.connect") as conn_mock:
+        ezsp = make_ezsp(version=4)
+
+        async def connect_mock(*args, **kwargs):
+            ezsp.connection_lost(
+                ConnectionResetError("Remote server closed connection")
+            )
+            return conn_mock.return_value
+
+        conn_mock.side_effect = connect_mock
+
+        with patch.object(ezsp, "_startup_reset") as startup_reset:
+            with pytest.raises(ConnectionResetError, match="lost during startup"):
+                await ezsp.connect()
+
+    assert startup_reset.await_count == 0
+    assert conn_mock.return_value.disconnect.mock_calls == []
+    assert ezsp._gw is None
+
+
 async def test_ezsp_connect_cancelled_by_transport_closed():
     """A threaded gateway cancels dispatched work when its transport closes.
 
@@ -466,10 +516,10 @@ async def test_ezsp_connect_transport_closed_during_startup_reset():
     """
     loop = asyncio.get_running_loop()
     waiting_for_reset = loop.create_future()
-    clients = []
+    client_connected = loop.create_future()
 
     async def handle_client(reader, writer):
-        clients.append(writer)
+        client_connected.set_result(writer)
 
     server = await asyncio.start_server(handle_client, "127.0.0.1", 0)
     port = server.sockets[0].getsockname()[1]
@@ -496,18 +546,21 @@ async def wait_for_startup_reset_mock(self):
         ):
             connect_task = asyncio.ensure_future(ezsp.connect(use_thread=True))
 
-            async with asyncio_timeout(5):
-                await waiting_for_reset
+            try:
+                async with asyncio_timeout(5):
+                    client = await client_connected
+                    await waiting_for_reset
 
-            # The bridge drops the connection
-            clients[0].close()
+                # The bridge drops the connection
+                client.close()
 
-            with pytest.raises(ConnectionResetError):
-                async with asyncio_timeout(1):
-                    await connect_task
+                with pytest.raises(ConnectionResetError):
+                    async with asyncio_timeout(1):
+                        await connect_task
+            finally:
+                connect_task.cancel()
     finally:
         server.close()
-        await server.wait_closed()
 
     assert ezsp._gw is None
     assert ezsp._application.connection_lost.mock_calls == [call(ANY)]

From f6bb448b31559e145279f4fa4898c1d23b5dc7f9 Mon Sep 17 00:00:00 2001
From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com>
Date: Thu, 17 Sep 2026 04:13:59 +0200
Subject: [PATCH 3/6] Await the ASH failure in test_ash_end_to_end instead of
 sleeping for it

The two "request fails due to a connectivity issue" steps created the send
task and slept `15 * t_rx_ack` (a few milliseconds at the test's scaled-down
timeouts) before unpausing the transport, expecting all `ACK_TIMEOUTS`
retries to have expired inside that window. Any scheduling hiccup of a few
milliseconds -- a garbage collection pass triggered by an earlier test is
enough -- leaves a retry pending, the unpaused transport delivers it, and
the test fails with `DID NOT RAISE TimeoutError`. Seen deterministically
when `tests/test_application.py` gained a test that leaves a reference
cycle behind.

A paused `FakeTransport` drops writes rather than buffering them, so the
failure can simply be awaited inside the paused block.
---
 tests/test_ash.py | 14 ++++----------
 1 file changed, 4 insertions(+), 10 deletions(-)

diff --git a/tests/test_ash.py b/tests/test_ash.py
index f97b2c4f..f36bbd0a 100644
--- a/tests/test_ash.py
+++ b/tests/test_ash.py
@@ -543,11 +543,8 @@ async def test_ash_end_to_end(transport_cls: type[FakeTransport]) -> None:
 
     # Let's let a request fail due to a connectivity issue
     with patch.object(ncp_transport, "paused", True):
-        send_task = asyncio.create_task(host.send_data(b"host failure"))
-        await asyncio.sleep(host._t_rx_ack * 15)
-
-    with pytest.raises(TimeoutError):
-        await send_task
+        with pytest.raises(TimeoutError):
+            await host.send_data(b"host failure")
 
     ncp_ezsp.data_received.reset_mock()
     host_ezsp.data_received.reset_mock()
@@ -569,11 +566,8 @@ async def test_ash_end_to_end(transport_cls: type[FakeTransport]) -> None:
     assert ncp._ncp_reset_code is None
 
     with patch.object(host_transport, "paused", True):
-        send_task = asyncio.create_task(ncp.send_data(b"ncp failure"))
-        await asyncio.sleep(ncp._t_rx_ack * 15)
-
-    with pytest.raises(TimeoutError):
-        await send_task
+        with pytest.raises(TimeoutError):
+            await ncp.send_data(b"ncp failure")
 
     assert (
         host._ncp_reset_code is t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT

From ce00ae099e14f01d19fe5920c302889a42c7ac10 Mon Sep 17 00:00:00 2001
From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com>
Date: Thu, 17 Sep 2026 04:39:13 +0200
Subject: [PATCH 4/6] Report the transport error uniformly, pin the
 cancellation yield

When the transport closed but the failure that surfaced is incidental to
it (a proxy `TypeError`/`RuntimeError` from the closing worker loop), raise
`ConnectionResetError` chained from it instead of the incidental error, so
every window reports the same thing to the caller.

Add the test for the `except CancelledError` branch's yield -- the
notification queued but not yet delivered when the cancellation arrives --
which was not pinned, and disconnect in the end-to-end test's cleanup so a
failing run doesn't leave the worker thread to the garbage collector.
---
 bellows/ezsp/__init__.py |  7 +++++++
 tests/test_ezsp.py       | 29 ++++++++++++++++++++++++++++-
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py
index 26bdb4dd..12812ea0 100644
--- a/bellows/ezsp/__init__.py
+++ b/bellows/ezsp/__init__.py
@@ -192,6 +192,13 @@ async def startup_reset(self) -> None:
                     continue
 
                 await self.disconnect()
+
+                if self._transport_closed and not isinstance(exc, ConnectionError):
+                    # What failed is incidental to the transport closing
+                    raise ConnectionResetError(
+                        "Connection was lost during startup"
+                    ) from exc
+
                 raise
 
     async def connect(self, *, use_thread: bool = True) -> None:
diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py
index 055a8004..995bd971 100644
--- a/tests/test_ezsp.py
+++ b/tests/test_ezsp.py
@@ -419,7 +419,32 @@ async def startup_reset_mock():
         with patch.object(
             ezsp, "_startup_reset", side_effect=startup_reset_mock
         ) as startup_reset:
-            with pytest.raises(TypeError):
+            with pytest.raises(ConnectionResetError, match="lost during startup") as e:
+                await ezsp.connect()
+
+    assert isinstance(e.value.__cause__, TypeError)
+    assert startup_reset.await_count == 1
+    assert conn_mock.return_value.disconnect.mock_calls == []
+    assert ezsp._gw is None
+
+
+async def test_ezsp_connect_cancelled_with_queued_connection_lost():
+    """Same as above, for a teardown cancellation reaching us before the notification."""
+    loop = asyncio.get_running_loop()
+    exc = ConnectionResetError("Remote server closed connection")
+
+    with patch("bellows.uart.connect") as conn_mock:
+        ezsp = make_ezsp(version=4)
+
+        async def startup_reset_mock():
+            # Queued, not yet delivered
+            loop.call_soon(ezsp.connection_lost, exc)
+            raise asyncio.CancelledError()
+
+        with patch.object(
+            ezsp, "_startup_reset", side_effect=startup_reset_mock
+        ) as startup_reset:
+            with pytest.raises(ConnectionResetError, match="lost during startup"):
                 await ezsp.connect()
 
     assert startup_reset.await_count == 1
@@ -559,6 +584,8 @@ async def wait_for_startup_reset_mock(self):
                         await connect_task
             finally:
                 connect_task.cancel()
+                # Tear down the gateway (and its thread) even if the test failed
+                await ezsp.disconnect()
     finally:
         server.close()
 

From b541aa3ad117f9876dea6de00334267e3e342731 Mon Sep 17 00:00:00 2001
From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com>
Date: Thu, 17 Sep 2026 04:39:13 +0200
Subject: [PATCH 5/6] Attach the worker-stop callback on the worker loop

`uart.connect()` attached the callback that stops the worker thread to the
connection-done future only after `_connect()` had returned to the calling
thread. If the connection was lost before that -- the bridge closing the
socket right after accepting it -- the future was already resolved, and
`add_done_callback()` queued the callback onto the worker loop with a plain
`call_soon()` from the wrong thread, which does not wake the loop. Nothing
in the worker's own queue would ever run it, so the thread stayed asleep in
`select()` until something else was dispatched to it.

Until now the startup retries always were that something. With `EZSP` no
longer dispatching to a gateway whose transport closed, the worker thread
leaked: seen as a hang at interpreter exit joining a `bellows.thread_`
thread still in `run_forever()`.

Attach the callback inside `_connect()`, on the loop that resolves the
future, before the connection can be lost.
---
 bellows/uart.py    | 14 ++++++++---
 tests/test_uart.py | 62 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 72 insertions(+), 4 deletions(-)

diff --git a/bellows/uart.py b/bellows/uart.py
index af274dc8..eeddd139 100644
--- a/bellows/uart.py
+++ b/bellows/uart.py
@@ -105,11 +105,18 @@ async def reset(self):
             return await self._reset_future
 
 
-async def _connect(config, api):
+async def _connect(config, api, thread=None):
     loop = asyncio.get_event_loop()
 
     connection_done_future = loop.create_future()
 
+    if thread is not None:
+        # Attach the callback here, on the loop that resolves the future. Attaching
+        # from the caller's thread once the connection is already lost would queue it
+        # with a plain `call_soon` that never wakes this loop: the thread would then
+        # sleep in `select()` until something else happens to be dispatched to it.
+        connection_done_future.add_done_callback(lambda _: thread.force_stop())
+
     gateway = Gateway(api, connection_done_future)
     protocol = AshProtocol(gateway)
 
@@ -139,13 +146,12 @@ async def connect(config, api, use_thread=True):
         thread = EventLoopThread()
         await thread.start()
         try:
-            protocol, connection_done = await thread.run_coroutine_threadsafe(
-                _connect(config, api)
+            protocol, _ = await thread.run_coroutine_threadsafe(
+                _connect(config, api, thread)
             )
         except Exception:
             thread.force_stop()
             raise
-        connection_done.add_done_callback(lambda _: thread.force_stop())
     else:
         protocol, _ = await _connect(config, api)
     return protocol
diff --git a/tests/test_uart.py b/tests/test_uart.py
index 7cc44b27..fd8e339d 100644
--- a/tests/test_uart.py
+++ b/tests/test_uart.py
@@ -1,4 +1,5 @@
 import asyncio
+from asyncio import timeout as asyncio_timeout
 import threading
 from unittest.mock import AsyncMock, MagicMock, call, patch, sentinel
 
@@ -138,6 +139,67 @@ async def mock_connect(loop, protocol_factory, *args, **kwargs):
     assert len(threads) == 0
 
 
+async def test_connect_threaded_connection_lost_before_connect_returns():
+    """The connection is lost before `connect()` resumes: the worker must still stop.
+
+    The callback that stops the worker thread used to be attached once `connect()`
+    resumed. A future that had already completed by then queued it onto the worker
+    loop from the wrong thread, without waking it: the thread slept forever.
+    """
+    loop = asyncio.get_running_loop()
+    lost = loop.create_future()
+
+    async def handle_client(reader, writer):
+        writer.close()
+
+    server = await asyncio.start_server(handle_client, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+
+    appmock = MagicMock()
+    appmock.connection_lost.side_effect = lambda exc: lost.set_result(exc)
+
+    run_coroutine_threadsafe = uart.EventLoopThread.run_coroutine_threadsafe
+
+    def resume_after_loss(self, coroutine):
+        future = run_coroutine_threadsafe(self, coroutine)
+
+        async def inner():
+            result = await future
+            # The application has been told the connection is lost, so the worker
+            # loop has already resolved the connection-done future. Give it time to
+            # go idle in `select()` as well: nothing else will be dispatched to it.
+            async with asyncio_timeout(1):
+                await lost
+            await asyncio.sleep(0.05)
+            return result
+
+        return inner()
+
+    try:
+        with patch.object(
+            uart.EventLoopThread, "run_coroutine_threadsafe", resume_after_loss
+        ):
+            gw = await uart.connect(
+                conf.SCHEMA_DEVICE(
+                    {
+                        conf.CONF_DEVICE_PATH: f"socket://127.0.0.1:{port}",
+                        conf.CONF_DEVICE_BAUDRATE: 115200,
+                    }
+                ),
+                appmock,
+                use_thread=True,
+            )
+    finally:
+        server.close()
+
+    assert gw is not None
+
+    # Ensure all threads are cleaned up
+    [t.join(1) for t in threading.enumerate() if "bellows" in t.name]
+    threads = [t for t in threading.enumerate() if "bellows" in t.name]
+    assert len(threads) == 0
+
+
 @pytest.fixture
 async def gw():
     gw = uart.Gateway(MagicMock())

From c5f14fce593b4d13c6fd6a10a38cd988bc957070 Mon Sep 17 00:00:00 2001
From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com>
Date: Thu, 17 Sep 2026 04:46:37 +0200
Subject: [PATCH 6/6] Report a worker teardown during `uart.connect()` as a
 lost connection

With the worker-stop callback attached before `_connect()` returns, a
transport lost while `_connect()` is still running would have the worker
cancel `_connect()` itself, and the resulting `CancelledError` would cross
`run_coroutine_threadsafe()` past `connect()`'s `except Exception` and reach
the application looking like a caller cancellation. Not reachable with
asyncio's `create_connection()` (its waiter resolves after
`connection_made()`, so `wait_until_connected()` never suspends), but cheap
to rule out: a `CancelledError` that is not our caller's is raised as
`ConnectionResetError`.

A caller cancelling `connect()` now also stops the worker thread, which
`except Exception` never covered.

The leak regression test stops a leaked worker itself before asserting, so
a regression fails the test instead of hanging the interpreter at exit.
---
 bellows/uart.py    |  9 +++++
 tests/test_uart.py | 82 +++++++++++++++++++++++++++++++++++++++++++---
 2 files changed, 87 insertions(+), 4 deletions(-)

diff --git a/bellows/uart.py b/bellows/uart.py
index eeddd139..54545637 100644
--- a/bellows/uart.py
+++ b/bellows/uart.py
@@ -149,6 +149,15 @@ async def connect(config, api, use_thread=True):
             protocol, _ = await thread.run_coroutine_threadsafe(
                 _connect(config, api, thread)
             )
+        except asyncio.CancelledError:
+            task = asyncio.current_task()
+            if task is not None and task.cancelling():
+                thread.force_stop()
+                raise
+
+            # Not our caller: the worker stopped itself, cancelling `_connect()`,
+            # because the connection was lost before it returned
+            raise ConnectionResetError("Connection was lost while connecting") from None
         except Exception:
             thread.force_stop()
             raise
diff --git a/tests/test_uart.py b/tests/test_uart.py
index fd8e339d..2e4f550d 100644
--- a/tests/test_uart.py
+++ b/tests/test_uart.py
@@ -139,6 +139,66 @@ async def mock_connect(loop, protocol_factory, *args, **kwargs):
     assert len(threads) == 0
 
 
+async def test_connect_threaded_cancelled_by_worker_teardown():
+    """The worker cancelling `_connect()` itself is reported as a lost connection."""
+
+    def cancelled(coroutine):
+        coroutine.close()
+        raise asyncio.CancelledError()
+
+    with patch.object(uart.EventLoopThread, "start", AsyncMock()), patch.object(
+        uart.EventLoopThread, "run_coroutine_threadsafe", side_effect=cancelled
+    ), patch.object(uart.EventLoopThread, "force_stop") as force_stop:
+        with pytest.raises(ConnectionResetError, match="lost while connecting"):
+            await uart.connect(
+                conf.SCHEMA_DEVICE(
+                    {
+                        conf.CONF_DEVICE_PATH: "/dev/serial",
+                        conf.CONF_DEVICE_BAUDRATE: 115200,
+                    }
+                ),
+                MagicMock(),
+                use_thread=True,
+            )
+
+    # The worker already stopped itself
+    assert force_stop.mock_calls == []
+
+
+async def test_connect_threaded_cancelled_by_caller_stops_thread():
+    """A caller cancelling `connect()` still stops the worker thread."""
+    loop = asyncio.get_running_loop()
+    started = loop.create_future()
+
+    async def blocked(coroutine):
+        coroutine.close()
+        started.set_result(None)
+        await loop.create_future()
+
+    with patch.object(uart.EventLoopThread, "start", AsyncMock()), patch.object(
+        uart.EventLoopThread, "run_coroutine_threadsafe", side_effect=blocked
+    ), patch.object(uart.EventLoopThread, "force_stop") as force_stop:
+        connect_task = asyncio.ensure_future(
+            uart.connect(
+                conf.SCHEMA_DEVICE(
+                    {
+                        conf.CONF_DEVICE_PATH: "/dev/serial",
+                        conf.CONF_DEVICE_BAUDRATE: 115200,
+                    }
+                ),
+                MagicMock(),
+                use_thread=True,
+            )
+        )
+        await started
+        connect_task.cancel()
+
+        with pytest.raises(asyncio.CancelledError):
+            await connect_task
+
+    assert len(force_stop.mock_calls) == 1
+
+
 async def test_connect_threaded_connection_lost_before_connect_returns():
     """The connection is lost before `connect()` resumes: the worker must still stop.
 
@@ -148,6 +208,13 @@ async def test_connect_threaded_connection_lost_before_connect_returns():
     """
     loop = asyncio.get_running_loop()
     lost = loop.create_future()
+    threads = []
+
+    thread_init = uart.EventLoopThread.__init__
+
+    def track_thread(self):
+        thread_init(self)
+        threads.append(self)
 
     async def handle_client(reader, writer):
         writer.close()
@@ -176,7 +243,7 @@ async def inner():
         return inner()
 
     try:
-        with patch.object(
+        with patch.object(uart.EventLoopThread, "__init__", track_thread), patch.object(
             uart.EventLoopThread, "run_coroutine_threadsafe", resume_after_loss
         ):
             gw = await uart.connect(
@@ -194,10 +261,17 @@ async def inner():
 
     assert gw is not None
 
-    # Ensure all threads are cleaned up
+    # The worker thread stops on its own
     [t.join(1) for t in threading.enumerate() if "bellows" in t.name]
-    threads = [t for t in threading.enumerate() if "bellows" in t.name]
-    assert len(threads) == 0
+    leaked = [t for t in threading.enumerate() if "bellows" in t.name]
+
+    # Never leave a stuck worker behind, even when this fails: it would hang the
+    # interpreter at exit
+    (thread,) = threads
+    thread.force_stop()
+    [t.join(1) for t in leaked]
+
+    assert leaked == []
 
 
 @pytest.fixture