fix: restart the MQTT session when one device stops receiving messages

docs/DEVICES.md

@@ -496,18 +496,52 @@ For each V1 command:
 | **Channel Class** | `V1Channel` | `MqttChannel` directly |
 | **RPC Abstraction** | `RpcChannel` with strategies | Helper functions |
 | **Strategy Pattern** | ✅ Multi-strategy (Local → MQTT) | ❌ Direct MQTT only |
-| **Health Manager** | ✅ Tracks local/MQTT health | ❌ Not needed |
+| **MQTT Health Reporting** | ✅ MQTT requests | ✅ A01 requests; ❌ B01 not yet |
 | **Code Location** | `v1_channel.py` | `a01_channel.py`, `b01_q7_channel.py` |
 
-#### Health Management (V1 Only)
+#### Local Connection Recovery (V1 Only)
 
-Each V1 RPC strategy can have a `HealthManager` that tracks success/failure:
+`V1Channel` keeps the local connection alive in the background:
 
-- **Exponential Backoff**: After failures, wait before retrying
-- **Automatic Recovery**: Periodically attempt to restore failed connections
-- **Network Info Refresh**: Refresh local IP addresses after extended periods
+- **Exponential Backoff**: After a failed local connection, wait before retrying (1 to 10 minutes)
+- **Automatic Recovery**: Check the local connection periodically and reconnect when it drops
+- **Network Info Refresh**: Refresh the device's local IP address after 12 hours
 
-A01/B01 devices don't need health management since they only use MQTT (no fallback).
+#### MQTT Health Management
+
+The broker can stop delivering one device's messages while the MQTT connection itself stays up and
+other devices on the shared session keep working. The client keepalive can't detect this, so the
+outcome of MQTT requests is tracked instead. Subscribing to the device's topic again on the same
+connection doesn't bring its messages back; a new connection does.
+
+The session's `HealthManager` receives the result of every MQTT request through a per-device
+`DeviceHealthManager`, and restarts the whole session (a new connection with all subscriptions
+re-established) when:
+
+- **One device fails**: 3 of its requests time out within 30 minutes. A single response in between
+  doesn't reset the count, because a failing connection may still deliver some of them. The device
+  counts as recovered after 3 responses in a row, or at its first response once its timeouts are
+  older than 30 minutes.
+- **The session fails**: 3 requests in a row time out across more than one device.
+
+Only timeouts on the current connection count: every new connection, whether forced or after the
+session reconnected on its own, starts counting from zero.
+
+Restarts are paced so that a device that is really offline doesn't keep interrupting the others:
+
+- At most one restart every 30 minutes.
+- While the same device keeps failing, the interval between the restarts it triggers doubles
+  (30 minutes, 1, 2, 4 hours, then every 6 hours). It is reset by the device's next response.
+- A forced restart reconnects right away; other disconnects keep the exponential backoff.
+
+V1 devices answer most requests over the local connection, so their MQTT topic may carry no traffic
+for hours. A broken topic is then noticed when the map (only available over MQTT) is requested
+again, and the session is restarted after the first few map requests time out.
+
+Each device's diagnostics include an `mqtt_health` section: when it started failing, its last
+success and timeout, request counts, and the last 10 health events (newest first): `timeouts_started`,
+`session_restart` (with `cause`: `this_device`, `other_device` or `several_devices`) and `recovered`.
+B01 devices don't report their requests yet, so they have no `mqtt_health` section.
 
 ### Protocol Versions
 

roborock/devices/device.py

@@ -15,6 +15,7 @@
 from roborock.data import HomeDataDevice, HomeDataProduct
 from roborock.diagnostics import redact_device_data
 from roborock.exceptions import RoborockException
+from roborock.mqtt.health_manager import DeviceHealthManager
 from roborock.roborock_message import RoborockMessage
 from roborock.util import RoborockLoggerAdapter
 
@@ -59,12 +60,15 @@ def __init__(
         product: HomeDataProduct,
         channel: Channel,
         trait: Trait,
+        mqtt_health: DeviceHealthManager | None = None,
     ) -> None:
         """Initialize the RoborockDevice.
 
         The device takes ownership of the channel for communication with the device.
         Use `connect()` to establish the connection, which will set up the appropriate
         protocol channel. Use `close()` to clean up all connections.
+
+        The optional `mqtt_health` adds the device's MQTT request history to its diagnostics.
         """
         TraitsMixin.__init__(self, trait)
         self._duid = device_info.duid
@@ -73,6 +77,7 @@ def __init__(
         self._device_info = device_info
         self._product = product
         self._channel = channel
+        self._mqtt_health = mqtt_health
         self._connect_task: asyncio.Task[None] | None = None
         self._unsub: Callable[[], None] | None = None
         self._ready_callbacks = CallbackList["RoborockDevice"]()
@@ -257,6 +262,8 @@ def diagnostic_data(self) -> dict[str, Any]:
             extra["traits"] = self.v1_properties.as_dict()
         elif self.b01_q10_properties:
             extra["traits"] = self.b01_q10_properties.as_dict()
+        if self._mqtt_health:
+            extra["mqtt_health"] = self._mqtt_health.diagnostic_data()
         return redact_device_data(
             {
                 "device": self.device_info.as_dict(),

roborock/devices/device_manager.py

@@ -21,6 +21,7 @@
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig
 from roborock.map.map_parser import MapParserConfig
+from roborock.mqtt.health_manager import DeviceHealthManager
 from roborock.mqtt.roborock_session import create_lazy_mqtt_session
 from roborock.mqtt.session import MqttSession, SessionUnauthorizedHook
 from roborock.protocol import create_mqtt_params
@@ -290,7 +291,11 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat
                     f"Device {device.name} has unsupported version {device.pv} {product.model}"
                 )
 
-        dev = RoborockDevice(device, product, channel, trait)
+        # B01 channels don't report their MQTT requests to the health manager yet.
+        mqtt_health = (
+            DeviceHealthManager(mqtt_session.health_manager, device.duid) if device.pv != DeviceVersion.B01 else None
+        )
+        dev = RoborockDevice(device, product, channel, trait, mqtt_health)
         if ready_callback:
             dev.add_ready_callback(ready_callback)
         return dev

roborock/devices/rpc/v1_channel.py

@@ -18,7 +18,7 @@
 from roborock.devices.transport.local_channel import LocalChannel, LocalSession, create_local_session
 from roborock.devices.transport.mqtt_channel import MqttChannel
 from roborock.exceptions import RoborockException
-from roborock.mqtt.health_manager import HealthManager
+from roborock.mqtt.health_manager import DeviceHealthManager
 from roborock.mqtt.session import MqttParams, MqttSession
 from roborock.protocols.v1_protocol import (
     CommandType,
@@ -67,7 +67,7 @@ class RpcStrategy:
     channel: LocalChannel | MqttChannel
     encoder: Callable[[RequestMessage], RoborockMessage]
     decoder: Callable[[RoborockMessage], ResponseMessage | MapResponse | None]
-    health_manager: HealthManager | None = None
+    health_manager: DeviceHealthManager | None = None
 
 
 class RpcChannel(V1RpcChannel):

roborock/devices/transport/mqtt_channel.py

@@ -7,7 +7,7 @@
 from roborock.callbacks import decoder_callback
 from roborock.data import HomeDataDevice, RRiot, UserData
 from roborock.exceptions import RoborockException
-from roborock.mqtt.health_manager import HealthManager
+from roborock.mqtt.health_manager import DeviceHealthManager
 from roborock.mqtt.session import MqttParams, MqttQos, MqttSession, MqttSessionException
 from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder
 from roborock.roborock_message import RoborockMessage
@@ -45,9 +45,9 @@ def is_connected(self) -> bool:
         return self._mqtt_session.connected
 
     @property
-    def health_manager(self) -> HealthManager:
-        """Return the health manager for the session."""
-        return self._mqtt_session.health_manager
+    def health_manager(self) -> DeviceHealthManager:
+        """Return the health manager that reports this device's events to the session."""
+        return DeviceHealthManager(self._mqtt_session.health_manager, self._duid)
 
     @property
     def is_local_connected(self) -> bool: