From 7a79d8dbf8ea0f8a58571375cc0595167b3c5505 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Tue, 18 Aug 2026 10:43:54 +0200 Subject: [PATCH 1/3] Propagate unexpected API disconnects to the ESPHome async transport The ESPHome transport connected without an `on_stop` callback, so when the device restarted (or the API connection dropped for any other reason) the transport was never told: `connection_lost()` was never dispatched, readers blocked forever, and writes were silently discarded inside the client loop. Consumers saw a healthy port that simply never produced data again. Wire `APIClient.connect(on_stop=...)` for connections the transport owns and translate a stop into `connection_lost()`: a clean device-initiated disconnect surfaces as EOF and an unexpected drop as a `SerialException`, matching how the descriptor transport reports fatal errors. Co-Authored-By: Claude Fable 5 --- serialx/platforms/serial_esphome.py | 43 ++++++++++++++++++++++++++++- tests/test_serial_esphome.py | 24 ++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/serialx/platforms/serial_esphome.py b/serialx/platforms/serial_esphome.py index 85b1825..c9e9e13 100644 --- a/serialx/platforms/serial_esphome.py +++ b/serialx/platforms/serial_esphome.py @@ -182,6 +182,8 @@ def __init__( self._read_event = asyncio.Event() self._unsub: Callable[[], None] | None = None self._instance_subscribed = False + # Invoked (on the client's loop) when an API connection we own stops + self._on_stop_cb: Callable[[bool], None] | None = None self._last_line_state = LineStateFlag(0) @@ -317,11 +319,18 @@ async def _async_open(self) -> None: self._client_loop = self._api.loop self._disconnect_api = True - await self._call_on_client_loop(self._api.connect(login=True)) + await self._call_on_client_loop( + self._api.connect(on_stop=self._api_stopped, login=True) + ) else: # Don't disconnect an externally-passed API self._disconnect_api = False + async def _api_stopped(self, expected_disconnect: bool) -> None: + """Handle the API connection stopping; runs on the client's loop.""" + if self._on_stop_cb is not None: + self._on_stop_cb(expected_disconnect) + @translate_esphome_errors async def _async_list_serial_ports(self) -> list[SerialPortInfo]: assert self._api is not None @@ -636,6 +645,7 @@ async def _connect( self._extra["serial"] = self._serial assert self._serial is not None + self._serial._on_stop_cb = self._on_api_stop await self._serial._async_open() assert self._serial._api is not None @@ -668,6 +678,37 @@ def _on_data(self, msg: SerialProxyDataReceived) -> None: else: self._loop.call_soon_threadsafe(self._protocol.data_received, msg.data) + def _on_api_stop(self, expected_disconnect: bool) -> None: + """Dispatch an API connection stop to the transport's loop.""" + assert self._serial is not None + client_loop = self._serial._client_loop + if client_loop is None or client_loop is self._loop: + self._handle_api_stop(expected_disconnect) + else: + self._loop.call_soon_threadsafe(self._handle_api_stop, expected_disconnect) + + def _handle_api_stop(self, expected_disconnect: bool) -> None: + """Handle the API connection stopping without a local `close()`.""" + if self._closing: + return + self._closing = True + + # The data subscription and instance subscription died with the + # connection; there is nothing left to unsubscribe or disconnect. + self._unsub = None + + serial = self._serial + exc: Exception | None = None + if not expected_disconnect: + exc = SerialException("ESPHome API connection lost") + self._mark_broken(exc) + + if serial is not None: + serial._instance_subscribed = False + serial._api = None + + self._call_protocol_connection_lost(exc) + def write(self, data: bytes | bytearray | memoryview) -> None: """Write data to the serial proxy.""" if self._closing: diff --git a/tests/test_serial_esphome.py b/tests/test_serial_esphome.py index 2ea0094..3666017 100644 --- a/tests/test_serial_esphome.py +++ b/tests/test_serial_esphome.py @@ -163,6 +163,30 @@ async def test_externally_passed_api_close_after_disconnect() -> None: await serial.close() +@pytest.mark.skipif(not ESPHOME_HOST_BINARY, reason="esphome host binary not available") +async def test_daemon_death_propagates_connection_lost() -> None: + """Test that losing the API connection unblocks readers and closes the port.""" + with create_socat_pair() as (socat_left, socat_right, _, _): + with contextlib.ExitStack() as stack: + left, _right = stack.enter_context( + create_esphome_pair(socat_left, socat_right) + ) + + serial = async_serial_for_url(url=left, baudrate=115200) + await serial.open() + + # Simulate the ESPHome device restarting mid-connection + stack.close() + + # The reader unblocks: an unexpected disconnect surfaces as a + # SerialException, a clean device-initiated disconnect as EOF. + with pytest.raises((SerialException, asyncio.IncompleteReadError)): + await asyncio.wait_for(serial.readexactly(1), timeout=10) + + assert not serial.is_open + await serial.close() + + @pytest.mark.skipif(not ESPHOME_HOST_BINARY, reason="esphome host binary not available") async def test_connect_by_instance_id() -> None: """Test connecting to an ESPHome serial proxy by instance ID.""" From 7926a7bb01cb96ba64c34d20f17ff2302dda9d11 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Tue, 18 Aug 2026 12:19:21 +0200 Subject: [PATCH 2/3] Simplify on_stop wiring: constructor arg, no indirection, no marshalling The callback is now a constructor argument passed straight through to `APIClient.connect` (which accepts None), and the transport handler runs directly as the on_stop coroutine: it is only wired for owned API connections, which are created on the transport's loop, so no cross-loop dispatch is needed. Co-Authored-By: Claude Fable 5 --- serialx/platforms/serial_esphome.py | 35 +++++++++++++---------------- 1 file changed, 15 insertions(+), 20 deletions(-) diff --git a/serialx/platforms/serial_esphome.py b/serialx/platforms/serial_esphome.py index c9e9e13..cc08479 100644 --- a/serialx/platforms/serial_esphome.py +++ b/serialx/platforms/serial_esphome.py @@ -123,6 +123,7 @@ def __init__( key: str | None = None, password: str | None = None, noise_psk: str | None = None, + on_stop: Callable[[bool], Coroutine[Any, Any, None]] | None = None, **kwargs: Any, ) -> None: """Initialize ESPHome serial port. @@ -147,6 +148,10 @@ def __init__( password: The API password to use when creating an `aioesphomeapi.APIClient` instance. noise_psk: An alias for `key`. Both cannot be passed at once. + on_stop: Coroutine callback invoked (on the client's loop) when an + API connection we own stops; receives `expected_disconnect`. + Ignored for an externally-passed `api`, whose owner already + controls `APIClient.connect`. *args: Passed through to `BaseSerial`. **kwargs: Passed through to `BaseSerial`. @@ -182,8 +187,7 @@ def __init__( self._read_event = asyncio.Event() self._unsub: Callable[[], None] | None = None self._instance_subscribed = False - # Invoked (on the client's loop) when an API connection we own stops - self._on_stop_cb: Callable[[bool], None] | None = None + self._on_stop = on_stop self._last_line_state = LineStateFlag(0) @@ -320,17 +324,12 @@ async def _async_open(self) -> None: self._disconnect_api = True await self._call_on_client_loop( - self._api.connect(on_stop=self._api_stopped, login=True) + self._api.connect(on_stop=self._on_stop, login=True) ) else: # Don't disconnect an externally-passed API self._disconnect_api = False - async def _api_stopped(self, expected_disconnect: bool) -> None: - """Handle the API connection stopping; runs on the client's loop.""" - if self._on_stop_cb is not None: - self._on_stop_cb(expected_disconnect) - @translate_esphome_errors async def _async_list_serial_ports(self) -> list[SerialPortInfo]: assert self._api is not None @@ -641,11 +640,12 @@ def __init__( async def _connect( self, *, path: str | None = None, **kwargs: Unpack[ConnectKwargs] ) -> None: - self._serial = self._serial_cls(loop=self._loop, path=path, **kwargs) + self._serial = self._serial_cls( + loop=self._loop, path=path, on_stop=self._on_api_stop, **kwargs + ) self._extra["serial"] = self._serial assert self._serial is not None - self._serial._on_stop_cb = self._on_api_stop await self._serial._async_open() assert self._serial._api is not None @@ -678,17 +678,12 @@ def _on_data(self, msg: SerialProxyDataReceived) -> None: else: self._loop.call_soon_threadsafe(self._protocol.data_received, msg.data) - def _on_api_stop(self, expected_disconnect: bool) -> None: - """Dispatch an API connection stop to the transport's loop.""" - assert self._serial is not None - client_loop = self._serial._client_loop - if client_loop is None or client_loop is self._loop: - self._handle_api_stop(expected_disconnect) - else: - self._loop.call_soon_threadsafe(self._handle_api_stop, expected_disconnect) + async def _on_api_stop(self, expected_disconnect: bool) -> None: + """Handle the API connection stopping without a local `close()`. - def _handle_api_stop(self, expected_disconnect: bool) -> None: - """Handle the API connection stopping without a local `close()`.""" + Only wired for API connections the serial owns, which are created on + the transport's loop, so this always runs on `self._loop`. + """ if self._closing: return self._closing = True From 2dad1803f075d886293ea457b1bec0addec827a5 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Tue, 18 Aug 2026 12:24:09 +0200 Subject: [PATCH 3/3] Shorten _on_api_stop docstring Co-Authored-By: Claude Fable 5 --- serialx/platforms/serial_esphome.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/serialx/platforms/serial_esphome.py b/serialx/platforms/serial_esphome.py index cc08479..fb4849c 100644 --- a/serialx/platforms/serial_esphome.py +++ b/serialx/platforms/serial_esphome.py @@ -679,11 +679,7 @@ def _on_data(self, msg: SerialProxyDataReceived) -> None: self._loop.call_soon_threadsafe(self._protocol.data_received, msg.data) async def _on_api_stop(self, expected_disconnect: bool) -> None: - """Handle the API connection stopping without a local `close()`. - - Only wired for API connections the serial owns, which are created on - the transport's loop, so this always runs on `self._loop`. - """ + """Handle the API connection stopping.""" if self._closing: return self._closing = True