diff --git a/docs/DEVICES.md b/docs/DEVICES.md
index cc7591e6b..6e148f25d 100644
--- a/docs/DEVICES.md
+++ b/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
 
diff --git a/roborock/devices/device.py b/roborock/devices/device.py
index 44c275f6c..56d75fa50 100644
--- a/roborock/devices/device.py
+++ b/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(),
diff --git a/roborock/devices/device_manager.py b/roborock/devices/device_manager.py
index f7d3a150a..f83c7b423 100644
--- a/roborock/devices/device_manager.py
+++ b/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
diff --git a/roborock/devices/rpc/v1_channel.py b/roborock/devices/rpc/v1_channel.py
index 8259994ad..2d7d6e6c2 100644
--- a/roborock/devices/rpc/v1_channel.py
+++ b/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):
diff --git a/roborock/devices/transport/mqtt_channel.py b/roborock/devices/transport/mqtt_channel.py
index 1a0f81e64..0c5156723 100644
--- a/roborock/devices/transport/mqtt_channel.py
+++ b/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:
diff --git a/roborock/mqtt/health_manager.py b/roborock/mqtt/health_manager.py
index 9e5a23ce6..f977ae9d3 100644
--- a/roborock/mqtt/health_manager.py
+++ b/roborock/mqtt/health_manager.py
@@ -1,28 +1,90 @@
 """A health manager for monitoring MQTT connections to Roborock devices.
 
 We observe a problem where sometimes the MQTT connection appears to be alive but
-no messages are being received. To mitigate this, we track consecutive timeouts
-and restart the connection if too many timeouts occur in succession.
+no messages are being received. To mitigate this, we track timeouts and restart
+the connection if too many of them occur.
+
+The broker can also stop delivering messages for a single device's topic while
+the rest of the session keeps working, so timeouts are tracked both for the
+session as a whole and for each device. Subscribing to the device's topic again
+on the same connection does not bring the messages back; only a new connection
+does, so the session is restarted. Such a failure can be partial, with some
+requests still answered in between, so a device's timeouts are counted within a
+time window rather than in a row.
 """
 
 import datetime
 import logging
+from collections import deque
 from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+from roborock.diagnostics import redact_device_uid
 
 _LOGGER = logging.getLogger(__name__)
 
-# Number of consecutive timeouts before considering the connection unhealthy.
+# Number of timeouts before considering the connection unhealthy: in a row across the
+# session, or within TIMEOUT_WINDOW for a single device.
 TIMEOUT_THRESHOLD = 3
 
+# Timeouts of a device older than this no longer count towards a restart. Successful
+# requests in between don't reset the count, since a failing connection may still
+# deliver some responses.
+TIMEOUT_WINDOW = datetime.timedelta(minutes=30)
+
+# A failing device counts as recovered after this many successful requests in a row, or at
+# its first successful request once its timeouts fell out of TIMEOUT_WINDOW.
+RECOVERY_THRESHOLD = 3
+
 # We won't restart the session more often than this interval.
 RESTART_COOLDOWN = datetime.timedelta(minutes=30)
 
+# A device that fails while the rest of the session works may simply be offline,
+# which a restart can't fix. Until it responds again, the interval between the
+# session restarts it triggers doubles after each restart, up to this maximum.
+MAX_DEVICE_RESTART_COOLDOWN = datetime.timedelta(hours=6)
+
+# Number of health changes kept per device for diagnostics.
+MAX_DEVICE_EVENTS = 10
+
+
+@dataclass
+class _DeviceHealth:
+    """The MQTT request history of a single device."""
+
+    recent_timeouts: deque[datetime.datetime] = field(default_factory=deque)
+    successes_in_a_row: int = 0
+    failing_since: datetime.datetime | None = None
+    restart_cooldown_start: datetime.datetime | None = None
+    restart_cooldown: datetime.timedelta | None = None
+    timeouts: int = 0
+    successes: int = 0
+    last_success: datetime.datetime | None = None
+    last_timeout: datetime.datetime | None = None
+    events: deque[dict[str, str]] = field(default_factory=lambda: deque(maxlen=MAX_DEVICE_EVENTS))
+
+    def add_event(self, time: datetime.datetime, event: str, **details: str) -> None:
+        """Record a change of the device's health for diagnostics."""
+        self.events.append({"time": time.isoformat(), "event": event, **details})
+
+    def forget_old_timeouts(self, now: datetime.datetime) -> None:
+        """Drop the timeouts that fell out of the time window."""
+        while self.recent_timeouts and now - self.recent_timeouts[0] >= TIMEOUT_WINDOW:
+            self.recent_timeouts.popleft()
+
+
+def _isoformat(value: datetime.datetime | None) -> str | None:
+    """Format an optional timestamp for diagnostics."""
+    return value.isoformat() if value else None
+
 
 class HealthManager:
     """Manager for monitoring the health of MQTT connections.
 
     This tracks communication timeouts and can trigger restarts of the MQTT
-    session if too many timeouts occur in succession.
+    session if too many timeouts occur, either in a row across the whole
+    session or within a time window for a single device.
     """
 
     def __init__(self, restart: Callable[[], Awaitable[None]]) -> None:
@@ -32,29 +94,154 @@ def __init__(self, restart: Callable[[], Awaitable[None]]) -> None:
             restart: A callable to restart the MQTT session.
         """
         self._consecutive_timeouts = 0
+        self._timed_out_devices: set[str] = set()
+        self._devices: dict[str, _DeviceHealth] = {}
         self._restart = restart
         self._last_restart: datetime.datetime | None = None
 
-    async def on_success(self) -> None:
-        """Record a successful communication event."""
+    def _device(self, device_id: str) -> _DeviceHealth:
+        """Return the history of a device, creating it on first use."""
+        return self._devices.setdefault(device_id, _DeviceHealth())
+
+    def on_connected(self) -> None:
+        """Forget the timeouts counted on a previous connection.
+
+        Timeouts on an old connection say nothing about a new one, whether the
+        session reconnected on its own or was restarted.
+        """
         self._consecutive_timeouts = 0
+        self._timed_out_devices.clear()
+        for device in self._devices.values():
+            device.recent_timeouts.clear()
 
-    async def on_timeout(self) -> None:
-        """Record a timeout event.
+    async def on_success(self, device_id: str) -> None:
+        """Record a successful communication event for a device."""
+        self._consecutive_timeouts = 0
+        self._timed_out_devices.clear()
+        device = self._device(device_id)
+        now = datetime.datetime.now(datetime.UTC)
+        device.successes += 1
+        device.successes_in_a_row += 1
+        device.last_success = now
+        device.forget_old_timeouts(now)
+        if device.failing_since is None:
+            return
+        # A single response in between timeouts doesn't mean the device recovered yet.
+        if device.recent_timeouts and device.successes_in_a_row < RECOVERY_THRESHOLD:
+            return
+        _LOGGER.info(
+            "Device %s responds over MQTT again after timing out since %s",
+            redact_device_uid(device_id),
+            device.failing_since.isoformat(timespec="seconds"),
+        )
+        device.add_event(now, "recovered")
+        device.recent_timeouts.clear()
+        device.failing_since = None
+        device.restart_cooldown_start = None
+        device.restart_cooldown = None
+
+    async def on_timeout(self, device_id: str) -> None:
+        """Record a timeout event for a device.
 
         This may trigger a restart of the MQTT session if too many timeouts
-        have occurred in succession.
+        have occurred.
         """
+        now = datetime.datetime.now(datetime.UTC)
         self._consecutive_timeouts += 1
-        if self._consecutive_timeouts >= TIMEOUT_THRESHOLD:
-            now = datetime.datetime.now(datetime.UTC)
-            since_last = (now - self._last_restart) if self._last_restart else None
-            if since_last is None or since_last >= RESTART_COOLDOWN:
+        self._timed_out_devices.add(device_id)
+        device = self._device(device_id)
+        device_name = redact_device_uid(device_id)
+        if device.failing_since is None:
+            device.failing_since = now
+            _LOGGER.info("Device %s started timing out over MQTT", device_name)
+            device.add_event(now, "timeouts_started")
+        device.forget_old_timeouts(now)
+        device.recent_timeouts.append(now)
+        device.successes_in_a_row = 0
+        device.timeouts += 1
+        device.last_timeout = now
+
+        # Only timeouts from several devices implicate the session itself. A single failing device
+        # may just be offline, even when it is the only device reporting to this session.
+        session_unhealthy = self._consecutive_timeouts >= TIMEOUT_THRESHOLD and len(self._timed_out_devices) > 1
+        if not session_unhealthy and len(device.recent_timeouts) < TIMEOUT_THRESHOLD:
+            return
+
+        since_last = (now - self._last_restart) if self._last_restart else None
+        if since_last is not None and since_last < RESTART_COOLDOWN:
+            _LOGGER.debug("Not restarting MQTT session for device %s, last restart was %s ago", device_name, since_last)
+            return
+
+        reason = f"{self._consecutive_timeouts} timeouts in a row across {len(self._timed_out_devices)} devices"
+        if not session_unhealthy:
+            cooldown_start, cooldown = device.restart_cooldown_start, device.restart_cooldown
+            if cooldown_start is not None and cooldown is not None and now - cooldown_start < cooldown:
                 _LOGGER.debug(
-                    "Restarting MQTT session after %d consecutive timeouts (duration since last restart %s)",
-                    self._consecutive_timeouts,
-                    since_last,
+                    "Not restarting MQTT session for device %s, it is still failing and may restart it again after %s",
+                    device_name,
+                    (cooldown_start + cooldown).isoformat(timespec="seconds"),
                 )
-                await self._restart()
-                self._last_restart = now
-                self._consecutive_timeouts = 0
+                return
+            device.restart_cooldown_start = now
+            device.restart_cooldown = (
+                min(cooldown * 2, MAX_DEVICE_RESTART_COOLDOWN) if cooldown is not None else RESTART_COOLDOWN
+            )
+            reason = (
+                f"device {device_name} timed out {len(device.recent_timeouts)} times in {TIMEOUT_WINDOW}; "
+                f"it may restart the session again in {device.restart_cooldown} if it keeps failing"
+            )
+
+        _LOGGER.info(
+            "Restarting MQTT session: %s (last restart: %s)",
+            reason,
+            f"{since_last} ago" if since_last is not None else "never",
+        )
+        # Update the state before awaiting, so timeouts arriving meanwhile don't restart again.
+        self._last_restart = now
+        # The restart affects every device, so each one records it along with what caused it.
+        several_devices = len(self._timed_out_devices) > 1
+        for tracked_id, tracked_device in self._devices.items():
+            if several_devices:
+                cause = "several_devices"
+            else:
+                cause = "this_device" if tracked_id == device_id else "other_device"
+            tracked_device.add_event(now, "session_restart", cause=cause)
+            # Timeouts on the old connection say nothing about the new one.
+            tracked_device.recent_timeouts.clear()
+        self._consecutive_timeouts = 0
+        self._timed_out_devices.clear()
+        await self._restart()
+
+    def device_diagnostic_data(self, device_id: str) -> dict[str, Any]:
+        """Return diagnostics about a device's MQTT requests."""
+        device = self._devices.get(device_id, _DeviceHealth())
+        return {
+            "failing_since": _isoformat(device.failing_since),
+            "last_success": _isoformat(device.last_success),
+            "last_timeout": _isoformat(device.last_timeout),
+            "successes": device.successes,
+            "timeouts": device.timeouts,
+            # Newest first, so the current state of the device is at the top.
+            "events": [dict(event) for event in reversed(device.events)],
+        }
+
+
+class DeviceHealthManager:
+    """Reports the outcome of a single device's MQTT requests to the session's health manager."""
+
+    def __init__(self, health_manager: HealthManager, device_id: str) -> None:
+        """Initialize the device health manager."""
+        self._health_manager = health_manager
+        self._device_id = device_id
+
+    async def on_success(self) -> None:
+        """Record a successful communication event."""
+        await self._health_manager.on_success(self._device_id)
+
+    async def on_timeout(self) -> None:
+        """Record a timeout event."""
+        await self._health_manager.on_timeout(self._device_id)
+
+    def diagnostic_data(self) -> dict[str, Any]:
+        """Return diagnostics about this device's MQTT requests."""
+        return self._health_manager.device_diagnostic_data(self._device_id)
diff --git a/roborock/mqtt/roborock_session.py b/roborock/mqtt/roborock_session.py
index a64992718..9abe27f49 100644
--- a/roborock/mqtt/roborock_session.py
+++ b/roborock/mqtt/roborock_session.py
@@ -79,6 +79,7 @@ def __init__(
         self._client_lock = asyncio.Lock()
         self._listeners: CallbackMap[str, bytes] = CallbackMap(_LOGGER)
         self._connection_task: asyncio.Task[None] | None = None
+        self._restart_requested = False
         self._topic_idle_timeout = topic_idle_timeout
         self._idle_timers: dict[str, asyncio.Task[None]] = {}
         self._diagnostics = params.diagnostics
@@ -145,12 +146,15 @@ async def close(self) -> None:
     async def restart(self) -> None:
         """Force the session to disconnect and reconnect.
 
-        The active connection task will be cancelled and restarted in the background, retried by
-        the reconnect loop. This is a no-op if there is no active connection.
+        The active connection task will be cancelled and restarted in the background by the
+        reconnect loop. The broker was reachable, so the first reconnect attempt is made right
+        away instead of after the backoff interval. This is a no-op if there is no active
+        connection.
         """
         _LOGGER.info("Forcing MQTT session restart")
         self._diagnostics.increment("restart")
-        if self._connection_task:
+        if self._connection_task and not self._connection_task.done():
+            self._restart_requested = True
             self._connection_task.cancel()
         else:
             _LOGGER.debug("No message loop task to cancel")
@@ -175,6 +179,7 @@ async def _run_reconnect_loop(self, start_future: asyncio.Future[None] | None) -
 
             self._healthy = False
             start_future = None
+            restart_requested, self._restart_requested = self._restart_requested, False
             if self._stop:
                 _LOGGER.debug("MQTT session closed, stopping retry loop")
                 return
@@ -188,6 +193,10 @@ async def _run_reconnect_loop(self, start_future: asyncio.Future[None] | None) -
                     return
                 self._backoff = MIN_BACKOFF_INTERVAL
                 continue
+            if restart_requested:
+                _LOGGER.info("MQTT session disconnected for a restart, reconnecting now")
+                self._diagnostics.increment("reconnect_restart")
+                continue
             _LOGGER.info("MQTT session disconnected, retrying in %s seconds", self._backoff.total_seconds())
             self._diagnostics.increment("reconnect_wait")
             await asyncio.sleep(self._backoff.total_seconds())
@@ -205,6 +214,7 @@ async def _run_connection(self, start_future: asyncio.Future[None] | None) -> No
                 async with self._mqtt_client(self._params) as client:
                     self._backoff = MIN_BACKOFF_INTERVAL
                     self._healthy = True
+                    self._health_manager.on_connected()
                     _LOGGER.info("MQTT Session connected.")
                     if start_future and not start_future.done():
                         start_future.set_result(None)
diff --git a/roborock/testing/channel.py b/roborock/testing/channel.py
index 65a36bc79..3e9f4f7d1 100644
--- a/roborock/testing/channel.py
+++ b/roborock/testing/channel.py
@@ -11,7 +11,7 @@
 from unittest.mock import AsyncMock, MagicMock
 
 from roborock.devices.transport.channel import Channel
-from roborock.mqtt.health_manager import HealthManager
+from roborock.mqtt.health_manager import DeviceHealthManager, HealthManager
 from roborock.mqtt.session import MqttQos
 from roborock.protocols.v1_protocol import LocalProtocolVersion
 from roborock.roborock_message import RoborockMessage
@@ -86,7 +86,7 @@ def __init__(self, is_local: bool = False):
         self.protocol_version = LocalProtocolVersion.V1
 
         self.restart = AsyncMock()
-        self.health_manager = HealthManager(self.restart)
+        self.health_manager = DeviceHealthManager(HealthManager(self.restart), "fake-device")
 
     async def _connect(self) -> None:
         self._is_connected = True
diff --git a/tests/devices/__snapshots__/test_device_manager.ambr b/tests/devices/__snapshots__/test_device_manager.ambr
index dadae5983..95963fe78 100644
--- a/tests/devices/__snapshots__/test_device_manager.ambr
+++ b/tests/devices/__snapshots__/test_device_manager.ambr
@@ -35,6 +35,15 @@
           'timeZoneId': 'America/Los_Angeles',
           'tuyaMigrated': False,
         }),
+        'mqtt_health': dict({
+          'events': list([
+          ]),
+          'failing_since': None,
+          'last_success': None,
+          'last_timeout': None,
+          'successes': 0,
+          'timeouts': 0,
+        }),
         'product': dict({
           'capability': 0,
           'category': 'robot.vacuum.cleaner',
diff --git a/tests/devices/test_device_manager.py b/tests/devices/test_device_manager.py
index 734c20657..0aaf87303 100644
--- a/tests/devices/test_device_manager.py
+++ b/tests/devices/test_device_manager.py
@@ -130,6 +130,9 @@ async def test_with_q10_device(cloud: FakeRoborockCloud, patch_device_manager: N
     assert q10_device.b01_q10_properties.status.status == YXDeviceState.CLEANING
     assert q10_sim.status[121] == 5
 
+    # B01 requests are not tracked by the MQTT health manager, so no empty history is reported.
+    assert "mqtt_health" not in q10_device.diagnostic_data()
+
     await device_manager.close()
 
 
diff --git a/tests/devices/transport/test_mqtt_channel.py b/tests/devices/transport/test_mqtt_channel.py
index 55e7b8bbb..dffde3fd1 100644
--- a/tests/devices/transport/test_mqtt_channel.py
+++ b/tests/devices/transport/test_mqtt_channel.py
@@ -10,6 +10,7 @@
 
 from roborock.data import HomeData, UserData
 from roborock.devices.transport.mqtt_channel import MqttChannel
+from roborock.mqtt.health_manager import HealthManager
 from roborock.mqtt.session import MqttParams
 from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder
 from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol
@@ -268,3 +269,22 @@ def failing_callback(message: RoborockMessage) -> None:
     # Unsubscribe all remaining subscribers
     unsub1()
     unsub2()
+
+
+async def test_health_is_tracked_per_device(mqtt_session: Mock) -> None:
+    """Test that a working device on the session does not hide a failing device's timeouts."""
+    restart = AsyncMock()
+    mqtt_session.health_manager = HealthManager(restart)
+    failing, working = (
+        MqttChannel(
+            mqtt_session, duid=duid, local_key=TEST_LOCAL_KEY, rriot=USER_DATA.rriot, mqtt_params=TEST_MQTT_PARAMS
+        )
+        for duid in ("failing", "working")
+    )
+
+    for _ in range(3):
+        restart.assert_not_called()
+        await working.health_manager.on_success()
+        await failing.health_manager.on_timeout()
+
+    restart.assert_called_once()
diff --git a/tests/mqtt/test_health_manager.py b/tests/mqtt/test_health_manager.py
index 94b253b4b..86a6ee5d2 100644
--- a/tests/mqtt/test_health_manager.py
+++ b/tests/mqtt/test_health_manager.py
@@ -1,15 +1,28 @@
 """Tests for the health manager."""
 
+import asyncio
 import datetime
+import logging
 from unittest.mock import AsyncMock, patch
 
-from roborock.mqtt.health_manager import HealthManager
+import pytest
+
+from roborock.mqtt.health_manager import (
+    MAX_DEVICE_EVENTS,
+    MAX_DEVICE_RESTART_COOLDOWN,
+    RECOVERY_THRESHOLD,
+    TIMEOUT_WINDOW,
+    DeviceHealthManager,
+    HealthManager,
+)
+
+DEVICE_ID = "device-1"
 
 
 async def test_health_manager_restart_called_after_timeouts() -> None:
     """Test that the health manager calls restart after consecutive timeouts."""
     restart = AsyncMock()
-    health_manager = HealthManager(restart=restart)
+    health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID)
 
     await health_manager.on_timeout()
     await health_manager.on_timeout()
@@ -19,29 +32,92 @@ async def test_health_manager_restart_called_after_timeouts() -> None:
     restart.assert_called_once()
 
 
-async def test_health_manager_success_resets_counter() -> None:
-    """Test that a successful message resets the timeout counter."""
+async def test_responses_in_between_do_not_hide_a_partial_failure() -> None:
+    """Test that timeouts interleaved with successful responses still restart the session."""
     restart = AsyncMock()
-    health_manager = HealthManager(restart=restart)
+    health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID)
 
     await health_manager.on_timeout()
+    await health_manager.on_success()
     await health_manager.on_timeout()
+    await health_manager.on_success()
     restart.assert_not_called()
 
+    await health_manager.on_timeout()
+    restart.assert_called_once()
+    # The device kept timing out, so the responses in between are not recorded as a recovery.
+    assert [event["event"] for event in health_manager.diagnostic_data()["events"]] == [
+        "session_restart",
+        "timeouts_started",
+    ]
+
+
+async def test_responses_in_a_row_mark_recovery() -> None:
+    """Test that a device answering several requests in a row recovers before its timeouts expire."""
+    restart = AsyncMock()
+    health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID)
+
+    await health_manager.on_timeout()
+    await health_manager.on_timeout()
+    for _ in range(RECOVERY_THRESHOLD - 1):
+        await health_manager.on_success()
+    assert health_manager.diagnostic_data()["failing_since"] is not None
+
     await health_manager.on_success()
+    assert health_manager.diagnostic_data()["failing_since"] is None
+    assert [event["event"] for event in health_manager.diagnostic_data()["events"]] == [
+        "recovered",
+        "timeouts_started",
+    ]
 
+    # The timeouts before the recovery no longer count towards a restart.
     await health_manager.on_timeout()
     await health_manager.on_timeout()
     restart.assert_not_called()
 
-    await health_manager.on_timeout()
+
+async def test_timeouts_before_a_reconnect_do_not_count() -> None:
+    """Test that a new connection is not restarted because of the old connection's timeouts."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    device = DeviceHealthManager(health_manager, DEVICE_ID)
+
+    await device.on_timeout()
+    await device.on_timeout()
+    health_manager.on_connected()
+
+    await device.on_timeout()
+    await device.on_timeout()
+    restart.assert_not_called()
+
+    await device.on_timeout()
     restart.assert_called_once()
 
 
+async def test_old_timeouts_do_not_count() -> None:
+    """Test that only the timeouts within the time window restart the session."""
+    restart = AsyncMock()
+    health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID)
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        mock_datetime.datetime.now.return_value = now
+        await health_manager.on_timeout()
+        await health_manager.on_timeout()
+
+        mock_datetime.datetime.now.return_value = now + TIMEOUT_WINDOW
+        await health_manager.on_timeout()
+        await health_manager.on_timeout()
+        restart.assert_not_called()
+
+        await health_manager.on_timeout()
+        restart.assert_called_once()
+
+
 async def test_cooldown() -> None:
     """Test that the health manager respects the restart cooldown."""
     restart = AsyncMock()
-    health_manager = HealthManager(restart=restart)
+    health_manager = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID)
 
     with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
         now = datetime.datetime(2023, 1, 1, 12, 0, 0)
@@ -71,3 +147,414 @@ async def test_cooldown() -> None:
         await health_manager.on_timeout()
         await health_manager.on_timeout()
         restart.assert_called_once()
+
+
+async def test_timeouts_across_devices_restart_session() -> None:
+    """Test that timeouts spread across devices still count towards the session."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+
+    await DeviceHealthManager(health_manager, "device-1").on_timeout()
+    await DeviceHealthManager(health_manager, "device-2").on_timeout()
+    restart.assert_not_called()
+
+    await DeviceHealthManager(health_manager, "device-3").on_timeout()
+    restart.assert_called_once()
+
+
+async def test_other_device_success_does_not_hide_failing_device() -> None:
+    """Test that one device's timeouts restart the session while another device works."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    failing = DeviceHealthManager(health_manager, "failing")
+    working = DeviceHealthManager(health_manager, "working")
+
+    await failing.on_timeout()
+    await failing.on_timeout()
+    await working.on_success()
+    restart.assert_not_called()
+
+    await failing.on_timeout()
+    restart.assert_called_once()
+
+
+async def test_device_needs_timeouts_of_its_own() -> None:
+    """Test that timeouts of other devices don't count towards a device's restart."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    device = DeviceHealthManager(health_manager, DEVICE_ID)
+    other = DeviceHealthManager(health_manager, "other")
+
+    await device.on_timeout()
+    await device.on_success()
+    await other.on_timeout()
+    await other.on_success()
+    await device.on_timeout()
+    restart.assert_not_called()
+
+
+async def test_restart_clears_counters_of_all_devices() -> None:
+    """Test that every device starts counting again after a session restart."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    first = DeviceHealthManager(health_manager, "first")
+    second = DeviceHealthManager(health_manager, "second")
+    working = DeviceHealthManager(health_manager, "working")
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        mock_datetime.datetime.now.return_value = now
+
+        await second.on_timeout()
+        await second.on_timeout()
+        await working.on_success()
+        await first.on_timeout()
+        await first.on_timeout()
+        await working.on_success()
+        await first.on_timeout()
+        restart.assert_called_once()
+        restart.reset_mock()
+
+        # The restart gave "second" a new subscription, so its earlier timeouts no longer count.
+        mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=31)
+        await working.on_success()
+        await second.on_timeout()
+        restart.assert_not_called()
+
+
+async def test_single_failing_device_cooldown_grows() -> None:
+    """Test that a device that stays unreachable restarts a working session less and less often."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    failing = DeviceHealthManager(health_manager, "failing")
+    working = DeviceHealthManager(health_manager, "working")
+
+    async def time_out_while_session_works() -> None:
+        for _ in range(3):
+            await working.on_success()
+            await failing.on_timeout()
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        mock_datetime.datetime.now.return_value = now
+        await time_out_while_session_works()
+        restart.assert_called_once()
+
+        # The interval between restarts doubles from the session cooldown up to the maximum.
+        hours = datetime.timedelta(hours=1)
+        for cooldown in (
+            hours / 2,
+            hours,
+            2 * hours,
+            4 * hours,
+            MAX_DEVICE_RESTART_COOLDOWN,
+            MAX_DEVICE_RESTART_COOLDOWN,
+        ):
+            restart.reset_mock()
+            mock_datetime.datetime.now.return_value = now + cooldown - datetime.timedelta(minutes=1)
+            await time_out_while_session_works()
+            restart.assert_not_called()
+
+            now += cooldown
+            mock_datetime.datetime.now.return_value = now
+            await time_out_while_session_works()
+            restart.assert_called_once()
+
+
+async def test_device_success_clears_device_cooldown() -> None:
+    """Test that a device that recovered can restart the session again if it fails later."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    failing = DeviceHealthManager(health_manager, "failing")
+    working = DeviceHealthManager(health_manager, "working")
+
+    async def time_out_while_session_works() -> None:
+        for _ in range(3):
+            await working.on_success()
+            await failing.on_timeout()
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        mock_datetime.datetime.now.return_value = now
+
+        await time_out_while_session_works()
+        restart.assert_called_once()
+        restart.reset_mock()
+
+        await failing.on_success()
+
+        # Only the session cooldown applies now.
+        mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=31)
+        await time_out_while_session_works()
+        restart.assert_called_once()
+
+
+async def test_device_diagnostics_are_separate_per_device() -> None:
+    """Test that diagnostics keep each device's history apart and survive counter resets."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    failing = DeviceHealthManager(health_manager, "failing")
+    working = DeviceHealthManager(health_manager, "working")
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC)
+        mock_datetime.datetime.now.return_value = now
+        await failing.on_success()
+
+        mock_datetime.datetime.now.return_value = now + datetime.timedelta(hours=1)
+        for _ in range(4):
+            await working.on_success()
+            await failing.on_timeout()
+
+    last_timeout = "2023-01-01T13:00:00+00:00"
+    assert failing.diagnostic_data() == {
+        "failing_since": last_timeout,
+        "last_success": "2023-01-01T12:00:00+00:00",
+        "last_timeout": last_timeout,
+        "successes": 1,
+        "timeouts": 4,
+        "events": [
+            {"time": last_timeout, "event": "session_restart", "cause": "this_device"},
+            {"time": last_timeout, "event": "timeouts_started"},
+        ],
+    }
+    assert working.diagnostic_data() == {
+        "failing_since": None,
+        "last_success": last_timeout,
+        "last_timeout": None,
+        "successes": 4,
+        "timeouts": 0,
+        "events": [
+            {"time": last_timeout, "event": "session_restart", "cause": "other_device"},
+        ],
+    }
+
+
+async def test_diagnostics_for_device_without_requests() -> None:
+    """Test the diagnostics of a device that has not used MQTT yet."""
+    device = DeviceHealthManager(HealthManager(restart=AsyncMock()), DEVICE_ID)
+
+    assert device.diagnostic_data() == {
+        "failing_since": None,
+        "last_success": None,
+        "last_timeout": None,
+        "successes": 0,
+        "timeouts": 0,
+        "events": [],
+    }
+
+
+async def test_diagnostics_record_when_timeouts_start_and_recover() -> None:
+    """Test that only changes of a device's health are recorded."""
+    device = DeviceHealthManager(HealthManager(restart=AsyncMock()), DEVICE_ID)
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        start = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC)
+        mock_datetime.datetime.now.return_value = start
+        await device.on_success()
+        await device.on_timeout()
+        await device.on_timeout()
+
+        # A response while the timeouts still count is not a recovery yet.
+        mock_datetime.datetime.now.return_value = start + datetime.timedelta(minutes=10)
+        await device.on_success()
+
+        recovered = start + TIMEOUT_WINDOW
+        mock_datetime.datetime.now.return_value = recovered
+        await device.on_success()
+        await device.on_success()
+
+    assert device.diagnostic_data()["events"] == [
+        {"time": recovered.isoformat(), "event": "recovered"},
+        {"time": start.isoformat(), "event": "timeouts_started"},
+    ]
+    assert device.diagnostic_data()["failing_since"] is None
+
+
+async def test_diagnostics_record_recovery_after_restart() -> None:
+    """Test that a device that recovers after a session restart is recorded as recovered."""
+    restart = AsyncMock()
+    device = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID)
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        start = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC)
+        mock_datetime.datetime.now.return_value = start
+        for _ in range(3):
+            await device.on_timeout()
+        restart.assert_called_once()
+
+        recovered = start + datetime.timedelta(minutes=10)
+        mock_datetime.datetime.now.return_value = recovered
+        await device.on_success()
+
+    assert device.diagnostic_data()["events"] == [
+        {"time": recovered.isoformat(), "event": "recovered"},
+        {"time": start.isoformat(), "event": "session_restart", "cause": "this_device"},
+        {"time": start.isoformat(), "event": "timeouts_started"},
+    ]
+    assert device.diagnostic_data()["failing_since"] is None
+
+
+async def test_health_changes_are_logged(caplog: pytest.LogCaptureFixture) -> None:
+    """Test that every decision about a failing device is visible in the info log."""
+    caplog.set_level(logging.INFO, logger="roborock.mqtt.health_manager")
+    health_manager = HealthManager(restart=AsyncMock())
+    failing = DeviceHealthManager(health_manager, "failing")
+    working = DeviceHealthManager(health_manager, "working")
+
+    for _ in range(3):
+        await working.on_success()
+        await failing.on_timeout()
+    await failing.on_success()
+
+    messages = [record.getMessage() for record in caplog.records]
+    assert [message.split(" ")[0] for message in messages] == ["Device", "Restarting", "Device"]
+    assert "started timing out over MQTT" in messages[0]
+    assert "may restart the session again in 0:30:00" in messages[1]
+    assert "(last restart: never)" in messages[1]
+    assert "responds over MQTT again" in messages[2]
+
+
+async def test_only_known_device_is_paced_like_any_failing_device() -> None:
+    """Test that the restarts triggered by the only device of a session become rarer."""
+    restart = AsyncMock()
+    device = DeviceHealthManager(HealthManager(restart=restart), DEVICE_ID)
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        mock_datetime.datetime.now.return_value = now
+        for _ in range(3):
+            await device.on_timeout()
+        restart.assert_called_once()
+        restart.reset_mock()
+
+        # The second restart is allowed after 30 minutes, the third only after another hour.
+        now += datetime.timedelta(minutes=31)
+        mock_datetime.datetime.now.return_value = now
+        for _ in range(3):
+            await device.on_timeout()
+        restart.assert_called_once()
+        restart.reset_mock()
+
+        mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=45)
+        for _ in range(3):
+            await device.on_timeout()
+        restart.assert_not_called()
+
+
+async def test_concurrent_timeouts_restart_once() -> None:
+    """Test that timeouts arriving while the session is restarting don't restart it again."""
+    restarted = asyncio.Event()
+    calls = 0
+
+    async def slow_restart() -> None:
+        nonlocal calls
+        calls += 1
+        await restarted.wait()
+
+    health_manager = HealthManager(restart=slow_restart)
+    failing = DeviceHealthManager(health_manager, "failing")
+    await DeviceHealthManager(health_manager, "working").on_success()
+
+    await failing.on_timeout()
+    await failing.on_timeout()
+    pending = [asyncio.create_task(failing.on_timeout()) for _ in range(5)]
+    await asyncio.sleep(0)
+    restarted.set()
+    await asyncio.gather(*pending)
+
+    assert calls == 1
+    assert [event["event"] for event in failing.diagnostic_data()["events"]] == [
+        "session_restart",
+        "timeouts_started",
+    ]
+
+
+async def test_session_failure_is_recorded_on_every_device() -> None:
+    """Test that a restart caused by several devices names them all as the cause."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    first = DeviceHealthManager(health_manager, "first")
+    second = DeviceHealthManager(health_manager, "second")
+
+    await first.on_timeout()
+    await second.on_timeout()
+    await first.on_timeout()
+
+    restart.assert_called_once()
+    for device in (first, second):
+        assert device.diagnostic_data()["events"][0]["cause"] == "several_devices"
+
+
+async def test_diagnostics_keep_only_recent_events() -> None:
+    """Test that the event history of a device is bounded."""
+    device = DeviceHealthManager(HealthManager(restart=AsyncMock()), DEVICE_ID)
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        for _ in range(MAX_DEVICE_EVENTS):
+            mock_datetime.datetime.now.return_value = now
+            await device.on_timeout()
+            now += TIMEOUT_WINDOW
+            mock_datetime.datetime.now.return_value = now
+            await device.on_success()
+
+    assert len(device.diagnostic_data()["events"]) == MAX_DEVICE_EVENTS
+
+
+async def test_device_cooldown_applies_without_successes_from_other_devices() -> None:
+    """Test that a failing device keeps its cooldown when other devices report nothing in between."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    failing = DeviceHealthManager(health_manager, "failing")
+    working = DeviceHealthManager(health_manager, "working")
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        mock_datetime.datetime.now.return_value = now
+        await working.on_success()
+
+        for _ in range(3):
+            await failing.on_timeout()
+        restart.assert_called_once()
+        restart.reset_mock()
+
+        now += datetime.timedelta(minutes=31)
+        mock_datetime.datetime.now.return_value = now
+        for _ in range(3):
+            await failing.on_timeout()
+        restart.assert_called_once()
+        restart.reset_mock()
+
+        # Past the session cooldown, but only this device is failing and its own cooldown has grown.
+        mock_datetime.datetime.now.return_value = now + datetime.timedelta(minutes=45)
+        for _ in range(3):
+            await failing.on_timeout()
+        restart.assert_not_called()
+
+
+async def test_timeouts_from_several_devices_bypass_device_cooldown() -> None:
+    """Test that a session failing for several devices restarts during one device's cooldown."""
+    restart = AsyncMock()
+    health_manager = HealthManager(restart=restart)
+    first = DeviceHealthManager(health_manager, "first")
+    second = DeviceHealthManager(health_manager, "second")
+
+    with patch("roborock.mqtt.health_manager.datetime") as mock_datetime:
+        now = datetime.datetime(2023, 1, 1, 12, 0, 0)
+        mock_datetime.datetime.now.return_value = now
+        await second.on_success()
+
+        for _ in range(3):
+            await first.on_timeout()
+        restart.assert_called_once()
+        restart.reset_mock()
+
+        # The second device is failing too, so the session is restarted again.
+        mock_datetime.datetime.now.return_value = now + datetime.timedelta(hours=1)
+        await first.on_timeout()
+        await second.on_timeout()
+        restart.assert_not_called()
+
+        await first.on_timeout()
+        restart.assert_called_once()
diff --git a/tests/mqtt/test_roborock_session.py b/tests/mqtt/test_roborock_session.py
index 65ccbbdf5..4206a1c67 100644
--- a/tests/mqtt/test_roborock_session.py
+++ b/tests/mqtt/test_roborock_session.py
@@ -539,6 +539,79 @@ def succeed_then_fail_unauthorized() -> Any:
         await session.close()
 
 
+async def test_restart_reconnects_without_backoff(
+    mock_aenter_client: AsyncMock,
+    mqtt_client_lite: AsyncMock,
+) -> None:
+    """Test that a forced restart reconnects right away instead of waiting for the backoff."""
+    params = copy.deepcopy(FAKE_PARAMS)
+    params.diagnostics = Diagnostics()
+    with patch("roborock.mqtt.roborock_session.MIN_BACKOFF_INTERVAL", datetime.timedelta(hours=1)):
+        session = await create_mqtt_session(params)
+        await session.subscribe("topic-1", Subscriber().append)
+
+        try:
+            await session.restart()
+            async with asyncio.timeout(1):
+                while mock_aenter_client.await_count < 2:
+                    await asyncio.sleep(0.01)
+            assert params.diagnostics.as_dict().get("reconnect_restart") == 1
+            assert "reconnect_wait" not in params.diagnostics.as_dict()
+        finally:
+            await session.close()
+
+
+async def test_disconnect_waits_for_backoff(
+    mock_aenter_client: AsyncMock,
+    message_iterator: FakeAsyncIterator,
+    mqtt_client_lite: AsyncMock,
+) -> None:
+    """Test that a connection lost without a forced restart still waits before reconnecting."""
+    params = copy.deepcopy(FAKE_PARAMS)
+    params.diagnostics = Diagnostics()
+    message_iterator.loop = False
+    with patch("roborock.mqtt.roborock_session.MIN_BACKOFF_INTERVAL", datetime.timedelta(hours=1)):
+        session = await create_mqtt_session(params)
+        await session.subscribe("topic-1", Subscriber().append)
+
+        try:
+            await asyncio.sleep(0.1)
+            assert mock_aenter_client.await_count == 1
+            assert params.diagnostics.as_dict().get("reconnect_wait", 0) >= 1
+            assert "reconnect_restart" not in params.diagnostics.as_dict()
+        finally:
+            await session.close()
+
+
+async def test_reconnect_resets_health_timeouts(
+    mock_aenter_client: AsyncMock,
+    message_iterator: FakeAsyncIterator,
+    mqtt_client_lite: AsyncMock,
+) -> None:
+    """Test that timeouts counted before a reconnect don't restart the new connection."""
+    params = copy.deepcopy(FAKE_PARAMS)
+    params.diagnostics = Diagnostics()
+    session = await create_mqtt_session(params)
+    await session.subscribe("topic-1", Subscriber().append)
+    health_manager = session.health_manager
+
+    try:
+        await health_manager.on_timeout("device-1")
+        await health_manager.on_timeout("device-1")
+
+        # The connection drops and the session reconnects on its own.
+        message_iterator.loop = False
+        async with asyncio.timeout(1):
+            while mock_aenter_client.await_count < 2:
+                await asyncio.sleep(0.01)
+        message_iterator.loop = True
+
+        await health_manager.on_timeout("device-1")
+        assert "restart" not in params.diagnostics.as_dict()
+    finally:
+        await session.close()
+
+
 async def test_session_defers_reconnect_when_idle(
     mock_aenter_client: AsyncMock,
     message_iterator: FakeAsyncIterator,