Propagate unexpected API disconnects to the ESPHome async transport

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,6 +187,7 @@ def __init__(
         self._read_event = asyncio.Event()
         self._unsub: Callable[[], None] | None = None
         self._instance_subscribed = False
+        self._on_stop = on_stop
 
         self._last_line_state = LineStateFlag(0)
 
@@ -317,7 +323,9 @@ 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._on_stop, login=True)
+            )
         else:
             # Don't disconnect an externally-passed API
             self._disconnect_api = False
@@ -632,7 +640,9 @@ 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
@@ -668,6 +678,28 @@ def _on_data(self, msg: SerialProxyDataReceived) -> None:
         else:
             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."""
+        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:

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."""