From 6c667d50b8ef3c3c6ce5d39a73443d7db67af5e8 Mon Sep 17 00:00:00 2001
From: Vincent <2070309+tubededentifrice@users.noreply.github.com>
Date: Sat, 25 Jul 2026 17:22:04 +0400
Subject: [PATCH 1/3] fix: request Q10 maps through dpMultiMap

---
 roborock/cli.py                             |  28 ++-
 roborock/devices/traits/b01/q10/__init__.py | 106 +++++++++-
 roborock/devices/traits/b01/q10/map.py      |  14 ++
 roborock/map/b01_q10_map_parser.py          |   4 +-
 tests/devices/traits/b01/q10/test_map.py    | 223 +++++++++++++++++++-
 tests/devices/traits/b01/q10/test_status.py |  11 +-
 tests/protocols/test_b01_q10_protocol.py    |  20 ++
 7 files changed, 373 insertions(+), 33 deletions(-)

diff --git a/roborock/cli.py b/roborock/cli.py
index 9def26af..c61b0d99 100644
--- a/roborock/cli.py
+++ b/roborock/cli.py
@@ -594,26 +594,28 @@ async def maps(ctx, device_id: str):
     await _display_v1_trait(context, device_id, lambda v1: v1.maps)
 
 
-# The Q10 pushes its map ~9s after a dpRequestDps; firmware throttles pushes to
-# ~once per 60-70s, so a single request is answered quickly but rapid re-requests
-# may not be. This bounds how long a one-shot CLI command waits for that push.
+# The Q10 publishes its map asynchronously after a dpMultiMap list/get request.
+# Firmware throttles pushes to ~once per 60-70s, so rapid re-requests may not be
+# answered immediately. This bounds how long a one-shot CLI command waits.
 _Q10_MAP_PUSH_TIMEOUT = 30.0
 
 
 async def _await_q10_map_push(
     properties: Q10PropertiesApi,
     predicate: Callable[[], bool],
+    add_source_listener: Callable[[Callable[[], None]], Callable[[], None]],
     *,
     timeout: float = _Q10_MAP_PUSH_TIMEOUT,
     allow_cached_on_timeout: bool = False,
 ) -> bool:
     """Nudge a Q10 to push its map/trace and wait for a fresh update.
 
-    The Q10 map API is entirely push-driven: there is no synchronous get-map
-    request. A ``dpRequestDps`` causes the device to publish a ``MAP_RESPONSE``,
-    which the device's subscribe loop feeds into the map trait. Here we register
-    an update listener, send the request, and wait for a newly pushed update to
-    satisfy ``predicate``. Returns whether it did within ``timeout``.
+    The Q10 map response remains asynchronous: ``refresh`` starts a
+    ``dpMultiMap`` list/get exchange, after which the device publishes a
+    ``MAP_RESPONSE`` that its subscribe loop feeds into the map trait. Here we
+    register a packet-specific listener, send the request, and wait for a newly
+    pushed update to satisfy ``predicate``. Returns whether it did within
+    ``timeout``.
     """
     loop = asyncio.get_running_loop()
     updated: asyncio.Future[None] = loop.create_future()
@@ -622,7 +624,7 @@ def on_update() -> None:
         if predicate() and not updated.done():
             updated.set_result(None)
 
-    unsub = properties.map.add_update_listener(on_update)
+    unsub = add_source_listener(on_update)
     try:
         await properties.refresh()
         await asyncio.wait_for(updated, timeout=timeout)
@@ -648,6 +650,7 @@ async def map_image(ctx, device_id: str, output_file: str):
         await _await_q10_map_push(
             properties,
             lambda: properties.map.image_content is not None,
+            properties.map._add_map_packet_listener,
             allow_cached_on_timeout=True,
         )
         image_content = properties.map.image_content
@@ -706,7 +709,11 @@ async def q10_position(ctx, device_id: str, include_path: bool):
         click.echo("Feature not supported by device")
         return
     properties = device.b01_q10_properties
-    got_trace = await _await_q10_map_push(properties, lambda: bool(properties.map.path))
+    got_trace = await _await_q10_map_push(
+        properties,
+        lambda: bool(properties.map.path),
+        properties.map._add_trace_packet_listener,
+    )
     if not got_trace:
         click.echo("No live trace available (the robot only reports position while cleaning).")
         return
@@ -871,6 +878,7 @@ async def rooms(ctx, device_id: str):
         await _await_q10_map_push(
             properties,
             lambda: properties.map.image_content is not None,
+            properties.map._add_map_packet_listener,
             allow_cached_on_timeout=True,
         )
         click.echo(dump_json({room.id: room.name for room in properties.map.rooms}))
diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py
index 3c8c73ff..a6205cd1 100644
--- a/roborock/devices/traits/b01/q10/__init__.py
+++ b/roborock/devices/traits/b01/q10/__init__.py
@@ -2,10 +2,12 @@
 
 import asyncio
 import logging
+from typing import Any
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.rpc.b01_q10_channel import B01Q10Channel
 from roborock.devices.traits import Trait
+from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket
 from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
@@ -38,6 +40,25 @@
 ]
 
 _LOGGER = logging.getLogger(__name__)
+_MAP_LIST_REQUEST_TIMEOUT = 30.0
+
+
+def _map_id_from_list_response(response: Any) -> int | str | None:
+    """Return the first usable map ID from a ``dpMultiMap`` list response."""
+    if not isinstance(response, dict) or response.get("op") != "list":
+        return None
+    data = response.get("data")
+    if not isinstance(data, list):
+        return None
+    for map_info in data:
+        if not isinstance(map_info, dict):
+            continue
+        map_id = map_info.get("id")
+        if isinstance(map_id, int) and not isinstance(map_id, bool):
+            return map_id
+        if isinstance(map_id, str) and map_id:
+            return map_id
+    return None
 
 
 class Q10PropertiesApi(Trait):
@@ -115,6 +136,9 @@ def __init__(self, channel: B01Q10Channel) -> None:
             self._map_dps,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
+        self._map_request_lock = asyncio.Lock()
+        self._map_list_request_token: object | None = None
+        self._map_list_requested_at: float | None = None
 
     async def start(self) -> None:
         """Start any necessary subscriptions for the trait."""
@@ -132,22 +156,55 @@ async def close(self) -> None:
 
     async def refresh(self) -> None:
         """Refresh all traits."""
-        # Sending the REQUEST_DPS will cause the device to send all DPS values
-        # to the device. Updates will be received by the subscribe loop below.
+        # Status and map retrieval use separate Q10 requests. A bare REQUEST_DPS
+        # reliably refreshes status but does not reliably make every firmware
+        # publish its map.
         await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
+        await self.request_map()
+
+    async def request_map(self) -> None:
+        """Request the current saved map through the Q10 multi-map protocol.
+
+        The list response arrives asynchronously on the subscribe stream.
+        ``_handle_message`` extracts its first map ID and follows up with the
+        matching ``get`` command; the resulting protocol-301 map packet is
+        routed to :attr:`map`.
+        """
+        async with self._map_request_lock:
+            now = asyncio.get_running_loop().time()
+            if (
+                self._map_list_request_token is not None
+                and self._map_list_requested_at is not None
+                and now - self._map_list_requested_at < _MAP_LIST_REQUEST_TIMEOUT
+            ):
+                return
+            token = object()
+            self._map_list_request_token = token
+            self._map_list_requested_at = now
+            sent = False
+            try:
+                await self.command.send(
+                    B01_Q10_DP.COMMON,
+                    {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
+                )
+                sent = True
+            finally:
+                if not sent and self._map_list_request_token is token:
+                    self._map_list_request_token = None
+                    self._map_list_requested_at = None
 
     async def _subscribe_loop(self) -> None:
         """Persistent loop dispatching decoded messages to the read-model traits."""
         async for message in self._channel.subscribe_stream():
-            self._handle_message(message)
+            await self._handle_message(message)
 
-    def _handle_message(self, message: Q10Message) -> None:
+    async def _handle_message(self, message: Q10Message) -> None:
         """Route a single decoded message to the trait responsible for it.
 
-        Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes (the
-        Q10 is entirely push-driven: there is no synchronous get-map request, a
-        ``dpRequestDps`` just nudges the device to publish its current map). DPS
-        updates feed the read-model traits. More traits can be dispatched here below.
+        Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes.
+        A ``dpMultiMap`` list response completes the asynchronous request flow
+        started by :meth:`request_map`; other DPS updates feed the read-model
+        traits.
         """
         if isinstance(message, Q10MapPacket):
             self.map.update_from_map_packet(message)
@@ -159,6 +216,39 @@ def _handle_message(self, message: Q10Message) -> None:
             # only updates the fields that it is responsible for.
             for trait in self._updatable_traits:
                 trait.update_from_dps(message.dps)
+            await self._request_map_from_list_response(message.dps)
+
+    async def _request_map_from_list_response(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
+        """Request map content after receiving our pending map-list response."""
+        response = decoded_dps.get(B01_Q10_DP.MULTI_MAP)
+        if self._map_list_request_token is None or not isinstance(response, dict) or response.get("op") != "list":
+            return
+
+        token = self._map_list_request_token
+        if (map_id := _map_id_from_list_response(response)) is None:
+            if self._map_list_request_token is token:
+                self._map_list_request_token = None
+                self._map_list_requested_at = None
+            _LOGGER.debug("Q10 map list response did not contain a usable map ID")
+            return
+
+        try:
+            await self.command.send(
+                B01_Q10_DP.COMMON,
+                {
+                    str(B01_Q10_DP.MULTI_MAP.code): {
+                        "op": "get",
+                        "id": map_id,
+                    }
+                },
+            )
+        except RoborockException as ex:
+            # A failed follow-up must not kill the persistent subscribe loop.
+            _LOGGER.debug("Failed to request Q10 map content: %s", ex)
+        finally:
+            if self._map_list_request_token is token:
+                self._map_list_request_token = None
+                self._map_list_requested_at = None
 
 
 def create(channel: B01Q10Channel) -> Q10PropertiesApi:
diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py
index ee51352a..c2a4e9b6 100644
--- a/roborock/devices/traits/b01/q10/map.py
+++ b/roborock/devices/traits/b01/q10/map.py
@@ -14,9 +14,11 @@
 """
 
 import logging
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import Any
 
+from roborock.callbacks import CallbackList
 from roborock.data import RoborockBase
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
@@ -89,6 +91,8 @@ def __init__(
         self._map_packet: Q10MapPacket | None = None
         self._trace_packet: Q10TracePacket | None = None
         self._image_content: bytes | None = None
+        self._map_packet_callbacks: CallbackList[None] = CallbackList(_LOGGER)
+        self._trace_packet_callbacks: CallbackList[None] = CallbackList(_LOGGER)
         self._map_dps.add_update_listener(self._map_dps_updated)
 
     @property
@@ -121,12 +125,22 @@ def update_from_map_packet(self, packet: Q10MapPacket) -> None:
         self._map_packet = packet
         self._render()
         self._notify_update()
+        self._map_packet_callbacks(None)
 
     def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
         """Store a trace-protocol update and render the latest sources."""
         self._trace_packet = packet
         self._render()
         self._notify_update()
+        self._trace_packet_callbacks(None)
+
+    def _add_map_packet_listener(self, callback: Callable[[], None]) -> Callable[[], None]:
+        """Register an internal callback for decoded map packets."""
+        return self._map_packet_callbacks.add_callback(lambda _: callback())
+
+    def _add_trace_packet_listener(self, callback: Callable[[], None]) -> Callable[[], None]:
+        """Register an internal callback for decoded trace packets."""
+        return self._trace_packet_callbacks.add_callback(lambda _: callback())
 
     def _map_dps_updated(self) -> None:
         """Render after the low-level DPS source changes."""
diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py
index 062f85f3..f7fd68f8 100644
--- a/roborock/map/b01_q10_map_parser.py
+++ b/roborock/map/b01_q10_map_parser.py
@@ -1,7 +1,7 @@
 """Parser for Roborock Q10 (B01/ss07) map packets.
 
-Q10 devices deliver map data as a protocol-301 ``MAP_RESPONSE`` message (pushed a
-few seconds after a ``dpRequestDps`` request). Unlike the Q7 ``SCMap`` protobuf
+Q10 devices deliver map data as a protocol-301 ``MAP_RESPONSE`` message after a
+``dpMultiMap`` list/get request. Unlike the Q7 ``SCMap`` protobuf
 format, the Q10 uses a custom, unencrypted binary packet:
 
 - ``01 01`` marker, then a ``u32be`` map id (bytes 2-5) and two consecutive
diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py
index 61120a7e..aa9e5194 100644
--- a/tests/devices/traits/b01/q10/test_map.py
+++ b/tests/devices/traits/b01/q10/test_map.py
@@ -1,10 +1,10 @@
 """Tests for the Q10 B01 map content trait.
 
-The Q10 map API is push-driven: the device publishes ``MAP_RESPONSE`` messages
-which the protocol layer decodes into typed map/trace packets; the trait updates
-its cached state from them via ``update_from_map_packet`` /
-``update_from_trace_packet`` (there is no synchronous get-map request). These
-tests cover that state management; the pixel/geometry work it drives is tested in
+The Q10 map API uses an asynchronous ``dpMultiMap`` list/get exchange. The
+device then publishes ``MAP_RESPONSE`` messages which the protocol layer
+decodes into typed map/trace packets; the trait updates its cached state from
+them via ``update_from_map_packet`` / ``update_from_trace_packet``. These tests
+cover that state management; the pixel/geometry work it drives is tested in
 ``tests/map/test_b01_q10_render.py``.
 """
 
@@ -13,7 +13,7 @@
 from collections.abc import AsyncGenerator
 from pathlib import Path
 from typing import cast
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 
@@ -29,7 +29,7 @@
     parse_trace_packet,
 )
 from roborock.map.b01_q10_render import Q10MapOverlays
-from roborock.protocols.b01_q10_protocol import Q10Message
+from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
 from .conftest import FakeB01Q10Channel
 
@@ -111,7 +111,10 @@ async def test_await_q10_map_push_waits_for_fresh_update() -> None:
     properties.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1, 2)]))
 
     got_trace = await _await_q10_map_push(
-        cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), timeout=0.01
+        cast(Q10PropertiesApi, properties),
+        lambda: bool(properties.map.path),
+        properties.map._add_trace_packet_listener,
+        timeout=0.01,
     )
 
     assert got_trace is False
@@ -122,7 +125,10 @@ async def test_await_q10_map_push_returns_true_after_update() -> None:
     properties = _FakeQ10PropertiesWithTrace()
 
     got_trace = await _await_q10_map_push(
-        cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), timeout=0.01
+        cast(Q10PropertiesApi, properties),
+        lambda: bool(properties.map.path),
+        properties.map._add_trace_packet_listener,
+        timeout=0.01,
     )
 
     assert got_trace is True
@@ -136,6 +142,7 @@ async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> No
     got_map = await _await_q10_map_push(
         cast(Q10PropertiesApi, properties),
         lambda: properties.map.image_content is not None,
+        properties.map._add_map_packet_listener,
         timeout=0.01,
         allow_cached_on_timeout=True,
     )
@@ -144,6 +151,30 @@ async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> No
     assert properties.refresh_count == 1
 
 
+async def test_await_q10_map_push_ignores_overlay_only_render() -> None:
+    """A DPS recomposition cannot masquerade as a fresh map packet."""
+    map_dps = MapDpsTrait()
+    properties = _FakeQ10Properties()
+    properties.map = MapContentTrait(map_dps)
+    properties.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes()))
+
+    async def refresh_overlay_only() -> None:
+        properties.refresh_count += 1
+        map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
+
+    properties.refresh = refresh_overlay_only  # type: ignore[method-assign]
+
+    got_map = await _await_q10_map_push(
+        cast(Q10PropertiesApi, properties),
+        lambda: properties.map.image_content is not None,
+        properties.map._add_map_packet_listener,
+        timeout=0.01,
+    )
+
+    assert got_map is False
+    assert properties.refresh_count == 1
+
+
 # --- Integration through the Q10PropertiesApi subscribe loop -----------------
 
 
@@ -204,6 +235,180 @@ async def test_subscribe_loop_routes_trace_push(
     assert q10_api.map.robot_position is not None
 
 
+async def test_map_list_response_requests_first_map(
+    q10_api: Q10PropertiesApi,
+    mock_channel: FakeB01Q10Channel,
+    message_queue: asyncio.Queue[Q10Message],
+) -> None:
+    """A requested map list is followed by a get for its first map ID."""
+    await q10_api.request_map()
+    assert mock_channel.published_commands == [
+        (
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
+        )
+    ]
+
+    message_queue.put_nowait(
+        Q10DpsUpdate(
+            dps={
+                B01_Q10_DP.MULTI_MAP: {
+                    "data": [
+                        {"id": "12345", "name": "Current", "timestamp": 1},
+                        {"id": "67890", "name": "Other", "timestamp": 2},
+                    ],
+                    "op": "list",
+                    "result": 1,
+                }
+            }
+        )
+    )
+
+    await _wait_for(lambda: len(mock_channel.published_commands) == 2)
+    assert mock_channel.published_commands[1] == (
+        B01_Q10_DP.COMMON,
+        {
+            str(B01_Q10_DP.MULTI_MAP.code): {
+                "op": "get",
+                "id": "12345",
+            }
+        },
+    )
+
+
+async def test_unsolicited_map_list_does_not_request_map(
+    q10_api: Q10PropertiesApi,
+    mock_channel: FakeB01Q10Channel,
+    message_queue: asyncio.Queue[Q10Message],
+) -> None:
+    """A list response not initiated by this API cannot trigger a map get."""
+    message_queue.put_nowait(
+        Q10DpsUpdate(
+            dps={
+                B01_Q10_DP.MULTI_MAP: {
+                    "data": [{"id": "12345"}],
+                    "op": "list",
+                    "result": 1,
+                }
+            }
+        )
+    )
+
+    await asyncio.sleep(0.01)
+    assert mock_channel.published_commands == []
+
+
+async def test_concurrent_map_requests_are_coalesced(q10_api: Q10PropertiesApi) -> None:
+    """Only one list request remains pending when callers refresh together."""
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def send_list(*_args, **_kwargs) -> None:
+        started.set()
+        await release.wait()
+
+    with patch.object(q10_api.command, "send", side_effect=send_list) as send:
+        first = asyncio.create_task(q10_api.request_map())
+        await started.wait()
+        second = asyncio.create_task(q10_api.request_map())
+        await asyncio.sleep(0)
+        release.set()
+        await asyncio.gather(first, second)
+
+    send.assert_awaited_once()
+
+
+async def test_cancelled_map_request_can_be_retried(q10_api: Q10PropertiesApi) -> None:
+    """Cancellation clears only its own pending list request."""
+    started = asyncio.Event()
+    blocked = asyncio.Event()
+
+    async def send_list(*_args, **_kwargs) -> None:
+        started.set()
+        await blocked.wait()
+
+    with patch.object(q10_api.command, "send", side_effect=send_list):
+        request = asyncio.create_task(q10_api.request_map())
+        await started.wait()
+        request.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await request
+
+    with patch.object(q10_api.command, "send", new=AsyncMock()) as retry:
+        await q10_api.request_map()
+
+    retry.assert_awaited_once()
+
+
+async def test_map_request_stays_coalesced_through_get(
+    q10_api: Q10PropertiesApi,
+    message_queue: asyncio.Queue[Q10Message],
+) -> None:
+    """The pending flow includes the follow-up get publish."""
+    get_started = asyncio.Event()
+    release_get = asyncio.Event()
+    get_finished = asyncio.Event()
+
+    async def send_command(_dp, params) -> None:
+        operation = params[str(B01_Q10_DP.MULTI_MAP.code)]["op"]
+        if operation == "get":
+            get_started.set()
+            await release_get.wait()
+            get_finished.set()
+
+    response = Q10DpsUpdate(
+        dps={
+            B01_Q10_DP.MULTI_MAP: {
+                "data": [{"id": "12345"}],
+                "op": "list",
+                "result": 1,
+            }
+        }
+    )
+    with patch.object(q10_api.command, "send", side_effect=send_command) as send:
+        await q10_api.request_map()
+        message_queue.put_nowait(response)
+        await get_started.wait()
+
+        await q10_api.request_map()
+        assert send.await_count == 2
+
+        release_get.set()
+        await get_finished.wait()
+        await q10_api.request_map()
+        assert send.await_count == 3
+
+
+async def test_failed_map_get_does_not_stop_subscription(
+    q10_api: Q10PropertiesApi,
+    message_queue: asyncio.Queue[Q10Message],
+) -> None:
+    """An expected command failure cannot stop later map-protocol updates."""
+
+    async def send_command(_dp, params) -> None:
+        if params[str(B01_Q10_DP.MULTI_MAP.code)]["op"] == "get":
+            raise RoborockException("device rejected map request")
+
+    with patch.object(q10_api.command, "send", side_effect=send_command):
+        await q10_api.request_map()
+        message_queue.put_nowait(
+            Q10DpsUpdate(
+                dps={
+                    B01_Q10_DP.MULTI_MAP: {
+                        "data": [{"id": "12345"}],
+                        "op": "list",
+                        "result": 1,
+                    }
+                }
+            )
+        )
+        message_queue.put_nowait(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()))
+
+        await _wait_for(lambda: bool(q10_api.map.path))
+
+    assert q10_api.map.robot_position is not None
+
+
 # --- Source composition + rendering ------------------------------------------
 
 
diff --git a/tests/devices/traits/b01/q10/test_status.py b/tests/devices/traits/b01/q10/test_status.py
index edd4bcb1..8f301483 100644
--- a/tests/devices/traits/b01/q10/test_status.py
+++ b/tests/devices/traits/b01/q10/test_status.py
@@ -128,10 +128,13 @@ async def test_status_trait_refresh(
 
     # Send a refresh command
     await q10_api.refresh()
-    assert len(mock_channel.published_commands) == 1
-    command, params = mock_channel.published_commands[0]
-    assert command == B01_Q10_DP.REQUEST_DPS
-    assert params == {}
+    assert mock_channel.published_commands == [
+        (B01_Q10_DP.REQUEST_DPS, {}),
+        (
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
+        ),
+    ]
 
     # Push the response message into the queue
     message_queue.put_nowait(message)
diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py
index f5d916d1..5fddc41a 100644
--- a/tests/protocols/test_b01_q10_protocol.py
+++ b/tests/protocols/test_b01_q10_protocol.py
@@ -41,6 +41,26 @@ def test_decode_message_dps_update() -> None:
     assert decoded == Q10DpsUpdate(dps={B01_Q10_DP.BATTERY: 100})
 
 
+def test_decode_message_common_multi_map_list() -> None:
+    """A nested dpCommon map-list response is flattened into dpMultiMap."""
+    message = _message(
+        b'{"dps":{"101":{"61":{"data":[{"id":"12345","name":"Map","timestamp":1}],"op":"list","result":1}}}}',
+        RoborockMessageProtocol.RPC_RESPONSE,
+    )
+
+    decoded = decode_message(message)
+
+    assert decoded == Q10DpsUpdate(
+        dps={
+            B01_Q10_DP.MULTI_MAP: {
+                "data": [{"id": "12345", "name": "Map", "timestamp": 1}],
+                "op": "list",
+                "result": 1,
+            }
+        }
+    )
+
+
 def test_decode_message_map_packet() -> None:
     """A MAP_RESPONSE 01 01 payload decodes into a Q10MapPacket."""
     message = _message(MAP_FIXTURE.read_bytes(), RoborockMessageProtocol.MAP_RESPONSE)

From c4415037de5db2f2eb2f85eb25fd8eb4bfd4577a Mon Sep 17 00:00:00 2001
From: Vincent <2070309+tubededentifrice@users.noreply.github.com>
Date: Thu, 30 Jul 2026 18:44:18 +0400
Subject: [PATCH 2/3] refactor: move Q10 refresh into map trait

---
 roborock/cli.py                             |   2 +-
 roborock/data/b01_q10/b01_q10_containers.py |  23 +++-
 roborock/devices/traits/b01/q10/__init__.py | 102 +----------------
 roborock/devices/traits/b01/q10/map.py      |  48 +++++++-
 tests/devices/traits/b01/q10/test_map.py    | 118 +++++---------------
 tests/devices/traits/b01/q10/test_status.py |   8 +-
 6 files changed, 104 insertions(+), 197 deletions(-)

diff --git a/roborock/cli.py b/roborock/cli.py
index c61b0d99..de528c3e 100644
--- a/roborock/cli.py
+++ b/roborock/cli.py
@@ -626,7 +626,7 @@ def on_update() -> None:
 
     unsub = add_source_listener(on_update)
     try:
-        await properties.refresh()
+        await properties.map.refresh()
         await asyncio.wait_for(updated, timeout=timeout)
         return True
     except TimeoutError:
diff --git a/roborock/data/b01_q10/b01_q10_containers.py b/roborock/data/b01_q10/b01_q10_containers.py
index 2223cc86..6ee43553 100644
--- a/roborock/data/b01_q10/b01_q10_containers.py
+++ b/roborock/data/b01_q10/b01_q10_containers.py
@@ -83,11 +83,32 @@ def start_datetime(self) -> datetime.datetime | None:
         return None
 
 
+@dataclass
+class Q10MapInfo(RoborockBase):
+    """A saved map reported by ``dpMultiMap``.
+
+    Q10 firmware represents the map identifier as a string on the wire. The
+    value is sent back unchanged in a subsequent ``{"op": "get"}`` request.
+    """
+
+    id: str
+    name: str | None = None
+    timestamp: int | None = None
+
+
 @dataclass
 class dpMultiMap(RoborockBase):
+    """Response envelope for the Q10 ``dpMultiMap`` data point."""
+
     op: str
     result: int
-    data: list
+    data: list[Q10MapInfo] = field(default_factory=list)
+
+    @property
+    def current_map_id(self) -> str | None:
+        """Return the first saved-map identifier, if one was reported."""
+        first = next((map_info for map_info in self.data if isinstance(map_info, Q10MapInfo) and map_info.id), None)
+        return first.id if first else None
 
 
 @dataclass
diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py
index a6205cd1..b13b8119 100644
--- a/roborock/devices/traits/b01/q10/__init__.py
+++ b/roborock/devices/traits/b01/q10/__init__.py
@@ -2,12 +2,10 @@
 
 import asyncio
 import logging
-from typing import Any
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.rpc.b01_q10_channel import B01Q10Channel
 from roborock.devices.traits import Trait
-from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket
 from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
@@ -40,25 +38,6 @@
 ]
 
 _LOGGER = logging.getLogger(__name__)
-_MAP_LIST_REQUEST_TIMEOUT = 30.0
-
-
-def _map_id_from_list_response(response: Any) -> int | str | None:
-    """Return the first usable map ID from a ``dpMultiMap`` list response."""
-    if not isinstance(response, dict) or response.get("op") != "list":
-        return None
-    data = response.get("data")
-    if not isinstance(data, list):
-        return None
-    for map_info in data:
-        if not isinstance(map_info, dict):
-            continue
-        map_id = map_info.get("id")
-        if isinstance(map_id, int) and not isinstance(map_id, bool):
-            return map_id
-        if isinstance(map_id, str) and map_id:
-            return map_id
-    return None
 
 
 class Q10PropertiesApi(Trait):
@@ -121,7 +100,7 @@ def __init__(self, channel: B01Q10Channel) -> None:
         self.network_info = NetworkInfoTrait()
         self.consumable = ConsumableTrait()
         self._map_dps = MapDpsTrait()
-        self.map = MapContentTrait(self._map_dps)
+        self.map = MapContentTrait(self._map_dps, self.command)
         self.clean_history = CleanHistoryTrait(self.command)
         # Read-model traits updated from the device's DPS push stream.
         self._updatable_traits = [
@@ -136,9 +115,6 @@ def __init__(self, channel: B01Q10Channel) -> None:
             self._map_dps,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
-        self._map_request_lock = asyncio.Lock()
-        self._map_list_request_token: object | None = None
-        self._map_list_requested_at: float | None = None
 
     async def start(self) -> None:
         """Start any necessary subscriptions for the trait."""
@@ -156,42 +132,9 @@ async def close(self) -> None:
 
     async def refresh(self) -> None:
         """Refresh all traits."""
-        # Status and map retrieval use separate Q10 requests. A bare REQUEST_DPS
-        # reliably refreshes status but does not reliably make every firmware
-        # publish its map.
+        # Sending REQUEST_DPS causes the device to publish its ordinary status
+        # values. Map refreshes have their own cadence through ``map.refresh()``.
         await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
-        await self.request_map()
-
-    async def request_map(self) -> None:
-        """Request the current saved map through the Q10 multi-map protocol.
-
-        The list response arrives asynchronously on the subscribe stream.
-        ``_handle_message`` extracts its first map ID and follows up with the
-        matching ``get`` command; the resulting protocol-301 map packet is
-        routed to :attr:`map`.
-        """
-        async with self._map_request_lock:
-            now = asyncio.get_running_loop().time()
-            if (
-                self._map_list_request_token is not None
-                and self._map_list_requested_at is not None
-                and now - self._map_list_requested_at < _MAP_LIST_REQUEST_TIMEOUT
-            ):
-                return
-            token = object()
-            self._map_list_request_token = token
-            self._map_list_requested_at = now
-            sent = False
-            try:
-                await self.command.send(
-                    B01_Q10_DP.COMMON,
-                    {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
-                )
-                sent = True
-            finally:
-                if not sent and self._map_list_request_token is token:
-                    self._map_list_request_token = None
-                    self._map_list_requested_at = None
 
     async def _subscribe_loop(self) -> None:
         """Persistent loop dispatching decoded messages to the read-model traits."""
@@ -202,9 +145,8 @@ async def _handle_message(self, message: Q10Message) -> None:
         """Route a single decoded message to the trait responsible for it.
 
         Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes.
-        A ``dpMultiMap`` list response completes the asynchronous request flow
-        started by :meth:`request_map`; other DPS updates feed the read-model
-        traits.
+        Map-list DPS responses are handed to the map trait; other DPS updates
+        feed the read-model traits.
         """
         if isinstance(message, Q10MapPacket):
             self.map.update_from_map_packet(message)
@@ -216,39 +158,7 @@ async def _handle_message(self, message: Q10Message) -> None:
             # only updates the fields that it is responsible for.
             for trait in self._updatable_traits:
                 trait.update_from_dps(message.dps)
-            await self._request_map_from_list_response(message.dps)
-
-    async def _request_map_from_list_response(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
-        """Request map content after receiving our pending map-list response."""
-        response = decoded_dps.get(B01_Q10_DP.MULTI_MAP)
-        if self._map_list_request_token is None or not isinstance(response, dict) or response.get("op") != "list":
-            return
-
-        token = self._map_list_request_token
-        if (map_id := _map_id_from_list_response(response)) is None:
-            if self._map_list_request_token is token:
-                self._map_list_request_token = None
-                self._map_list_requested_at = None
-            _LOGGER.debug("Q10 map list response did not contain a usable map ID")
-            return
-
-        try:
-            await self.command.send(
-                B01_Q10_DP.COMMON,
-                {
-                    str(B01_Q10_DP.MULTI_MAP.code): {
-                        "op": "get",
-                        "id": map_id,
-                    }
-                },
-            )
-        except RoborockException as ex:
-            # A failed follow-up must not kill the persistent subscribe loop.
-            _LOGGER.debug("Failed to request Q10 map content: %s", ex)
-        finally:
-            if self._map_list_request_token is token:
-                self._map_list_request_token = None
-                self._map_list_requested_at = None
+            await self.map.update_from_dps(message.dps)
 
 
 def create(channel: B01Q10Channel) -> Q10PropertiesApi:
diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py
index c2a4e9b6..157b5a12 100644
--- a/roborock/devices/traits/b01/q10/map.py
+++ b/roborock/devices/traits/b01/q10/map.py
@@ -21,6 +21,7 @@
 from roborock.callbacks import CallbackList
 from roborock.data import RoborockBase
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_containers import dpMultiMap
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
@@ -33,6 +34,7 @@
 from roborock.map.b01_q10_overlays import parse_virtual_wall_blob, parse_zone_blob
 from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map
 
+from .command import CommandTrait
 from .common import UpdatableTrait
 
 _LOGGER = logging.getLogger(__name__)
@@ -72,22 +74,34 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
         self._notify_update()
 
 
-class MapContentTrait(TraitUpdateListener):
+@dataclass
+class MapListDps(RoborockBase):
+    """Typed ``dpMultiMap`` state delivered through the Q10 DPS stream."""
+
+    multi_map: dpMultiMap | None = field(default=None, metadata={"dps": B01_Q10_DP.MULTI_MAP})
+
+
+class MapContentTrait(MapListDps, TraitUpdateListener):
     """High-level composed Q10 map view.
 
     The latest map and trace packets are combined with the injected
     :class:`MapDpsTrait` whenever any of those three sources changes.
     """
 
+    _CONVERTER = DpsDataConverter.from_dataclass(MapListDps)
+
     def __init__(
         self,
         map_dps: MapDpsTrait,
+        command: CommandTrait | None = None,
         *,
         map_parser_config: B01Q10MapParserConfig | None = None,
     ) -> None:
+        MapListDps.__init__(self)
         TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._config = map_parser_config or B01Q10MapParserConfig()
         self._map_dps = map_dps
+        self._command = command
         self._map_packet: Q10MapPacket | None = None
         self._trace_packet: Q10TracePacket | None = None
         self._image_content: bytes | None = None
@@ -95,6 +109,38 @@ def __init__(
         self._trace_packet_callbacks: CallbackList[None] = CallbackList(_LOGGER)
         self._map_dps.add_update_listener(self._map_dps_updated)
 
+    async def refresh(self) -> None:
+        """Request the current saved map independently of general status."""
+        if self._command is None:
+            raise ValueError("Trait is read-only; no command channel was provided")
+        await self._command.send(
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
+        )
+
+    async def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
+        """Request map content when a typed ``dpMultiMap`` list response arrives."""
+        if not self._CONVERTER.update_from_dps(self, decoded_dps):
+            return
+        if self._command is None or self.multi_map is None or self.multi_map.op != "list" or self.multi_map.result != 1:
+            return
+        if (map_id := self.multi_map.current_map_id) is None:
+            _LOGGER.debug("Q10 map list response did not contain a map ID")
+            return
+        try:
+            await self._command.send(
+                B01_Q10_DP.COMMON,
+                {
+                    str(B01_Q10_DP.MULTI_MAP.code): {
+                        "op": "get",
+                        "id": map_id,
+                    }
+                },
+            )
+        except RoborockException as ex:
+            # A failed follow-up must not kill the persistent subscribe loop.
+            _LOGGER.debug("Failed to request Q10 map content: %s", ex)
+
     @property
     def image_content(self) -> bytes | None:
         """The composed map PNG, if the latest map rendered successfully."""
diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py
index aa9e5194..2bac3104 100644
--- a/tests/devices/traits/b01/q10/test_map.py
+++ b/tests/devices/traits/b01/q10/test_map.py
@@ -13,7 +13,7 @@
 from collections.abc import AsyncGenerator
 from pathlib import Path
 from typing import cast
-from unittest.mock import AsyncMock, Mock, patch
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -95,14 +95,21 @@ def __init__(self) -> None:
         self.map = _map_trait()
         self.refresh_count = 0
 
-    async def refresh(self) -> None:
-        self.refresh_count += 1
+        async def refresh_map() -> None:
+            self.refresh_count += 1
+
+        self.map.refresh = refresh_map  # type: ignore[method-assign]
 
 
 class _FakeQ10PropertiesWithTrace(_FakeQ10Properties):
-    async def refresh(self) -> None:
-        await super().refresh()
-        self.map.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()))
+    def __init__(self) -> None:
+        super().__init__()
+
+        async def refresh_map() -> None:
+            self.refresh_count += 1
+            self.map.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()))
+
+        self.map.refresh = refresh_map  # type: ignore[method-assign]
 
 
 async def test_await_q10_map_push_waits_for_fresh_update() -> None:
@@ -162,7 +169,7 @@ async def refresh_overlay_only() -> None:
         properties.refresh_count += 1
         map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
 
-    properties.refresh = refresh_overlay_only  # type: ignore[method-assign]
+    properties.map.refresh = refresh_overlay_only  # type: ignore[method-assign]
 
     got_map = await _await_q10_map_push(
         cast(Q10PropertiesApi, properties),
@@ -240,8 +247,8 @@ async def test_map_list_response_requests_first_map(
     mock_channel: FakeB01Q10Channel,
     message_queue: asyncio.Queue[Q10Message],
 ) -> None:
-    """A requested map list is followed by a get for its first map ID."""
-    await q10_api.request_map()
+    """The map trait follows a typed map list with a get for its first ID."""
+    await q10_api.map.refresh()
     assert mock_channel.published_commands == [
         (
             B01_Q10_DP.COMMON,
@@ -274,19 +281,21 @@ async def test_map_list_response_requests_first_map(
             }
         },
     )
+    assert q10_api.map.multi_map is not None
+    assert q10_api.map.multi_map.current_map_id == "12345"
 
 
-async def test_unsolicited_map_list_does_not_request_map(
+async def test_map_list_response_without_map_id_is_ignored(
     q10_api: Q10PropertiesApi,
     mock_channel: FakeB01Q10Channel,
     message_queue: asyncio.Queue[Q10Message],
 ) -> None:
-    """A list response not initiated by this API cannot trigger a map get."""
+    """A typed list response with no entries cannot trigger a map get."""
     message_queue.put_nowait(
         Q10DpsUpdate(
             dps={
                 B01_Q10_DP.MULTI_MAP: {
-                    "data": [{"id": "12345"}],
+                    "data": [],
                     "op": "list",
                     "result": 1,
                 }
@@ -298,85 +307,13 @@ async def test_unsolicited_map_list_does_not_request_map(
     assert mock_channel.published_commands == []
 
 
-async def test_concurrent_map_requests_are_coalesced(q10_api: Q10PropertiesApi) -> None:
-    """Only one list request remains pending when callers refresh together."""
-    started = asyncio.Event()
-    release = asyncio.Event()
-
-    async def send_list(*_args, **_kwargs) -> None:
-        started.set()
-        await release.wait()
-
-    with patch.object(q10_api.command, "send", side_effect=send_list) as send:
-        first = asyncio.create_task(q10_api.request_map())
-        await started.wait()
-        second = asyncio.create_task(q10_api.request_map())
-        await asyncio.sleep(0)
-        release.set()
-        await asyncio.gather(first, second)
-
-    send.assert_awaited_once()
-
-
-async def test_cancelled_map_request_can_be_retried(q10_api: Q10PropertiesApi) -> None:
-    """Cancellation clears only its own pending list request."""
-    started = asyncio.Event()
-    blocked = asyncio.Event()
-
-    async def send_list(*_args, **_kwargs) -> None:
-        started.set()
-        await blocked.wait()
-
-    with patch.object(q10_api.command, "send", side_effect=send_list):
-        request = asyncio.create_task(q10_api.request_map())
-        await started.wait()
-        request.cancel()
-        with pytest.raises(asyncio.CancelledError):
-            await request
-
-    with patch.object(q10_api.command, "send", new=AsyncMock()) as retry:
-        await q10_api.request_map()
-
-    retry.assert_awaited_once()
-
-
-async def test_map_request_stays_coalesced_through_get(
-    q10_api: Q10PropertiesApi,
-    message_queue: asyncio.Queue[Q10Message],
-) -> None:
-    """The pending flow includes the follow-up get publish."""
-    get_started = asyncio.Event()
-    release_get = asyncio.Event()
-    get_finished = asyncio.Event()
-
-    async def send_command(_dp, params) -> None:
-        operation = params[str(B01_Q10_DP.MULTI_MAP.code)]["op"]
-        if operation == "get":
-            get_started.set()
-            await release_get.wait()
-            get_finished.set()
-
-    response = Q10DpsUpdate(
-        dps={
-            B01_Q10_DP.MULTI_MAP: {
-                "data": [{"id": "12345"}],
-                "op": "list",
-                "result": 1,
-            }
-        }
-    )
-    with patch.object(q10_api.command, "send", side_effect=send_command) as send:
-        await q10_api.request_map()
-        message_queue.put_nowait(response)
-        await get_started.wait()
-
-        await q10_api.request_map()
-        assert send.await_count == 2
+async def test_map_refresh_requests_are_not_rate_limited(q10_api: Q10PropertiesApi) -> None:
+    """The caller controls map cadence; each refresh emits its own list request."""
+    with patch.object(q10_api.command, "send") as send:
+        await q10_api.map.refresh()
+        await q10_api.map.refresh()
 
-        release_get.set()
-        await get_finished.wait()
-        await q10_api.request_map()
-        assert send.await_count == 3
+    assert send.await_count == 2
 
 
 async def test_failed_map_get_does_not_stop_subscription(
@@ -390,7 +327,6 @@ async def send_command(_dp, params) -> None:
             raise RoborockException("device rejected map request")
 
     with patch.object(q10_api.command, "send", side_effect=send_command):
-        await q10_api.request_map()
         message_queue.put_nowait(
             Q10DpsUpdate(
                 dps={
diff --git a/tests/devices/traits/b01/q10/test_status.py b/tests/devices/traits/b01/q10/test_status.py
index 8f301483..ed55c3c6 100644
--- a/tests/devices/traits/b01/q10/test_status.py
+++ b/tests/devices/traits/b01/q10/test_status.py
@@ -128,13 +128,7 @@ async def test_status_trait_refresh(
 
     # Send a refresh command
     await q10_api.refresh()
-    assert mock_channel.published_commands == [
-        (B01_Q10_DP.REQUEST_DPS, {}),
-        (
-            B01_Q10_DP.COMMON,
-            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
-        ),
-    ]
+    assert mock_channel.published_commands == [(B01_Q10_DP.REQUEST_DPS, {})]
 
     # Push the response message into the queue
     message_queue.put_nowait(message)

From 0fee0551e55d619fc6274b55e11df3007d9225c7 Mon Sep 17 00:00:00 2001
From: Vincent <2070309+tubededentifrice@users.noreply.github.com>
Date: Sat, 1 Aug 2026 11:19:52 +0400
Subject: [PATCH 3/3] refactor: split Q10 map list and content

---
 roborock/cli.py                             |  32 ++--
 roborock/devices/traits/b01/q10/__init__.py |  19 +-
 roborock/devices/traits/b01/q10/map.py      |  69 +++-----
 roborock/devices/traits/b01/q10/maps.py     |  58 +++++++
 tests/devices/traits/b01/q10/test_map.py    | 181 ++++++++++++++------
 5 files changed, 248 insertions(+), 111 deletions(-)
 create mode 100644 roborock/devices/traits/b01/q10/maps.py

diff --git a/roborock/cli.py b/roborock/cli.py
index de528c3e..3d8a6396 100644
--- a/roborock/cli.py
+++ b/roborock/cli.py
@@ -608,14 +608,11 @@ async def _await_q10_map_push(
     timeout: float = _Q10_MAP_PUSH_TIMEOUT,
     allow_cached_on_timeout: bool = False,
 ) -> bool:
-    """Nudge a Q10 to push its map/trace and wait for a fresh update.
-
-    The Q10 map response remains asynchronous: ``refresh`` starts a
-    ``dpMultiMap`` list/get exchange, after which the device publishes a
-    ``MAP_RESPONSE`` that its subscribe loop feeds into the map trait. Here we
-    register a packet-specific listener, send the request, and wait for a newly
-    pushed update to satisfy ``predicate``. Returns whether it did within
-    ``timeout``.
+    """Request Q10 map content and wait for a fresh map or trace packet.
+
+    A Q10 needs a saved-map ID before it can request content. The map list and
+    content have independent refresh schedules, so the list is requested only
+    when no ID is stored. The content then arrives as a later ``MAP_RESPONSE``.
     """
     loop = asyncio.get_running_loop()
     updated: asyncio.Future[None] = loop.create_future()
@@ -626,8 +623,23 @@ def on_update() -> None:
 
     unsub = add_source_listener(on_update)
     try:
-        await properties.map.refresh()
-        await asyncio.wait_for(updated, timeout=timeout)
+        async with asyncio.timeout(timeout):
+            if properties.maps.current_map_id is None:
+                map_list_updated: asyncio.Future[None] = loop.create_future()
+
+                def on_map_list_update() -> None:
+                    if properties.maps.current_map_id is not None and not map_list_updated.done():
+                        map_list_updated.set_result(None)
+
+                unsub_maps = properties.maps.add_update_listener(on_map_list_update)
+                try:
+                    await properties.maps.refresh()
+                    if properties.maps.current_map_id is None:
+                        await map_list_updated
+                finally:
+                    unsub_maps()
+            await properties.map.refresh()
+            await updated
         return True
     except TimeoutError:
         return allow_cached_on_timeout and predicate()
diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py
index b13b8119..6fe7c2cd 100644
--- a/roborock/devices/traits/b01/q10/__init__.py
+++ b/roborock/devices/traits/b01/q10/__init__.py
@@ -17,6 +17,7 @@
 from .do_not_disturb import DoNotDisturbTrait
 from .dust_collection import DustCollectionTrait
 from .map import MapContentTrait, MapDpsTrait
+from .maps import MapsTrait
 from .network_info import NetworkInfoTrait
 from .remote import RemoteTrait
 from .status import StatusTrait
@@ -32,6 +33,7 @@
     "DoNotDisturbTrait",
     "DustCollectionTrait",
     "MapContentTrait",
+    "MapsTrait",
     "NetworkInfoTrait",
     "SoundVolumeTrait",
     "StatusTrait",
@@ -79,6 +81,9 @@ class Q10PropertiesApi(Trait):
     map: MapContentTrait
     """Composed map image plus caller-facing map and trace data."""
 
+    maps: MapsTrait
+    """Saved-map list metadata."""
+
     _map_dps: MapDpsTrait
     """Private source of restricted zones and virtual walls received through DPS."""
 
@@ -100,7 +105,8 @@ def __init__(self, channel: B01Q10Channel) -> None:
         self.network_info = NetworkInfoTrait()
         self.consumable = ConsumableTrait()
         self._map_dps = MapDpsTrait()
-        self.map = MapContentTrait(self._map_dps, self.command)
+        self.maps = MapsTrait(self.command)
+        self.map = MapContentTrait(self._map_dps, self.maps, self.command)
         self.clean_history = CleanHistoryTrait(self.command)
         # Read-model traits updated from the device's DPS push stream.
         self._updatable_traits = [
@@ -113,6 +119,7 @@ def __init__(self, channel: B01Q10Channel) -> None:
             self.consumable,
             self.clean_history,
             self._map_dps,
+            self.maps,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
 
@@ -133,20 +140,19 @@ async def close(self) -> None:
     async def refresh(self) -> None:
         """Refresh all traits."""
         # Sending REQUEST_DPS causes the device to publish its ordinary status
-        # values. Map refreshes have their own cadence through ``map.refresh()``.
+        # values. Map-list and map-content refreshes have separate schedules.
         await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
 
     async def _subscribe_loop(self) -> None:
         """Persistent loop dispatching decoded messages to the read-model traits."""
         async for message in self._channel.subscribe_stream():
-            await self._handle_message(message)
+            self._handle_message(message)
 
-    async def _handle_message(self, message: Q10Message) -> None:
+    def _handle_message(self, message: Q10Message) -> None:
         """Route a single decoded message to the trait responsible for it.
 
         Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes.
-        Map-list DPS responses are handed to the map trait; other DPS updates
-        feed the read-model traits.
+        Map-list DPS responses and other DPS updates feed the read-model traits.
         """
         if isinstance(message, Q10MapPacket):
             self.map.update_from_map_packet(message)
@@ -158,7 +164,6 @@ async def _handle_message(self, message: Q10Message) -> None:
             # only updates the fields that it is responsible for.
             for trait in self._updatable_traits:
                 trait.update_from_dps(message.dps)
-            await self.map.update_from_dps(message.dps)
 
 
 def create(channel: B01Q10Channel) -> Q10PropertiesApi:
diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py
index 157b5a12..5aebca5d 100644
--- a/roborock/devices/traits/b01/q10/map.py
+++ b/roborock/devices/traits/b01/q10/map.py
@@ -6,11 +6,10 @@
 * trace packets are decoded from trace-protocol responses;
 * restricted zones and virtual walls arrive as ordinary DPS values.
 
-``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` depends
-on it and combines that state with the latest map/trace packets through the pure
-functions in :mod:`roborock.map.b01_q10_render`. The high-level trait keeps only
-the latest value from each source and one replace-whole rendered image;
-calibration, path placement and overlay placement remain inside the renderer.
+``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` uses a
+stored ID from ``MapsTrait`` only when it requests content. It combines map,
+trace, and overlay state through the pure functions in
+:mod:`roborock.map.b01_q10_render`. Map-list updates do not refresh content.
 """
 
 import logging
@@ -21,7 +20,6 @@
 from roborock.callbacks import CallbackList
 from roborock.data import RoborockBase
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
-from roborock.data.b01_q10.b01_q10_containers import dpMultiMap
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
@@ -36,6 +34,7 @@
 
 from .command import CommandTrait
 from .common import UpdatableTrait
+from .maps import MapsTrait
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -74,33 +73,27 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
         self._notify_update()
 
 
-@dataclass
-class MapListDps(RoborockBase):
-    """Typed ``dpMultiMap`` state delivered through the Q10 DPS stream."""
-
-    multi_map: dpMultiMap | None = field(default=None, metadata={"dps": B01_Q10_DP.MULTI_MAP})
-
-
-class MapContentTrait(MapListDps, TraitUpdateListener):
+class MapContentTrait(TraitUpdateListener):
     """High-level composed Q10 map view.
 
     The latest map and trace packets are combined with the injected
-    :class:`MapDpsTrait` whenever any of those three sources changes.
+    :class:`MapDpsTrait` whenever any of those three sources changes. The
+    :class:`MapsTrait` supplies a stored ID only when this trait requests
+    content.
     """
 
-    _CONVERTER = DpsDataConverter.from_dataclass(MapListDps)
-
     def __init__(
         self,
         map_dps: MapDpsTrait,
-        command: CommandTrait | None = None,
+        maps: MapsTrait,
+        command: CommandTrait,
         *,
         map_parser_config: B01Q10MapParserConfig | None = None,
     ) -> None:
-        MapListDps.__init__(self)
         TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._config = map_parser_config or B01Q10MapParserConfig()
         self._map_dps = map_dps
+        self._maps = maps
         self._command = command
         self._map_packet: Q10MapPacket | None = None
         self._trace_packet: Q10TracePacket | None = None
@@ -110,37 +103,21 @@ def __init__(
         self._map_dps.add_update_listener(self._map_dps_updated)
 
     async def refresh(self) -> None:
-        """Request the current saved map independently of general status."""
-        if self._command is None:
-            raise ValueError("Trait is read-only; no command channel was provided")
+        """Request content for the first map in the latest saved-map list."""
+        if (map_id := self._maps.current_map_id) is None:
+            raise RoborockException("Cannot request Q10 map content before the map list is available")
+        # Map lists and map content can change at different times. Reuse the
+        # stored ID so a content refresh does not also refresh the list.
         await self._command.send(
             B01_Q10_DP.COMMON,
-            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
+            {
+                str(B01_Q10_DP.MULTI_MAP.code): {
+                    "op": "get",
+                    "id": map_id,
+                }
+            },
         )
 
-    async def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
-        """Request map content when a typed ``dpMultiMap`` list response arrives."""
-        if not self._CONVERTER.update_from_dps(self, decoded_dps):
-            return
-        if self._command is None or self.multi_map is None or self.multi_map.op != "list" or self.multi_map.result != 1:
-            return
-        if (map_id := self.multi_map.current_map_id) is None:
-            _LOGGER.debug("Q10 map list response did not contain a map ID")
-            return
-        try:
-            await self._command.send(
-                B01_Q10_DP.COMMON,
-                {
-                    str(B01_Q10_DP.MULTI_MAP.code): {
-                        "op": "get",
-                        "id": map_id,
-                    }
-                },
-            )
-        except RoborockException as ex:
-            # A failed follow-up must not kill the persistent subscribe loop.
-            _LOGGER.debug("Failed to request Q10 map content: %s", ex)
-
     @property
     def image_content(self) -> bytes | None:
         """The composed map PNG, if the latest map rendered successfully."""
diff --git a/roborock/devices/traits/b01/q10/maps.py b/roborock/devices/traits/b01/q10/maps.py
new file mode 100644
index 00000000..d0945776
--- /dev/null
+++ b/roborock/devices/traits/b01/q10/maps.py
@@ -0,0 +1,58 @@
+"""Trait for Q10 saved-map list data."""
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+from roborock.data import RoborockBase
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_containers import dpMultiMap
+from roborock.devices.traits.common import DpsDataConverter
+
+from .command import CommandTrait
+from .common import UpdatableTrait
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class Maps(RoborockBase):
+    """Saved-map list data from the Q10 DPS stream."""
+
+    multi_map: dpMultiMap | None = field(default=None, metadata={"dps": B01_Q10_DP.MULTI_MAP})
+
+    @property
+    def current_map_id(self) -> str | None:
+        """Return the first saved-map ID for a content request, if available."""
+        if self.multi_map is None or self.multi_map.op != "list" or self.multi_map.result != 1:
+            return None
+        return self.multi_map.current_map_id
+
+
+class MapsTrait(Maps, UpdatableTrait):
+    """Request and store the Q10 saved-map list."""
+
+    _CONVERTER = DpsDataConverter.from_dataclass(Maps)
+    _command: CommandTrait
+
+    def __init__(self, command: CommandTrait) -> None:
+        """Initialize the saved-map list trait."""
+        Maps.__init__(self)
+        UpdatableTrait.__init__(self, command, _LOGGER)
+        self._command = command
+
+    async def refresh(self) -> None:
+        """Request a new saved-map list from the device."""
+        await self._command.send(
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
+        )
+
+    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
+        """Store a successful saved-map list response."""
+        response = decoded_dps.get(B01_Q10_DP.MULTI_MAP)
+        # DP 61 also carries map-content acknowledgements. Ignore them so they
+        # cannot replace a usable map list with an unrelated response.
+        if not isinstance(response, dict) or response.get("op") != "list" or response.get("result") != 1:
+            return
+        super().update_from_dps(decoded_dps)
diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py
index 2bac3104..6f3785c6 100644
--- a/tests/devices/traits/b01/q10/test_map.py
+++ b/tests/devices/traits/b01/q10/test_map.py
@@ -1,11 +1,9 @@
 """Tests for the Q10 B01 map content trait.
 
-The Q10 map API uses an asynchronous ``dpMultiMap`` list/get exchange. The
-device then publishes ``MAP_RESPONSE`` messages which the protocol layer
-decodes into typed map/trace packets; the trait updates its cached state from
-them via ``update_from_map_packet`` / ``update_from_trace_packet``. These tests
-cover that state management; the pixel/geometry work it drives is tested in
-``tests/map/test_b01_q10_render.py``.
+Map list data and map content have independent refresh schedules. Content
+requests use a stored map ID, and the device sends the data later in a
+``MAP_RESPONSE`` packet. These tests cover that state management. The render
+details are tested in ``tests/map/test_b01_q10_render.py``.
 """
 
 import asyncio
@@ -20,7 +18,9 @@
 from roborock.cli import _await_q10_map_push, cli
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.traits.b01.q10 import Q10PropertiesApi, create
+from roborock.devices.traits.b01.q10.command import CommandTrait
 from roborock.devices.traits.b01.q10.map import MapContentTrait, MapDpsTrait
+from roborock.devices.traits.b01.q10.maps import MapsTrait
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
     Q10Point,
@@ -37,9 +37,20 @@
 TRACE_SESSION_FIXTURE = Path("tests/map/testdata/b01_q10_trace_session.bin")
 
 
-def _map_trait() -> MapContentTrait:
-    """Create a high-level trait with its required low-level dependency."""
-    return MapContentTrait(MapDpsTrait())
+def _map_trait(map_dps: MapDpsTrait | None = None) -> MapContentTrait:
+    """Create map content with a stored map ID for tests that do not perform I/O."""
+    command = cast(CommandTrait, Mock(spec=CommandTrait))
+    maps = MapsTrait(command)
+    maps.update_from_dps(
+        {
+            B01_Q10_DP.MULTI_MAP: {
+                "data": [{"id": "12345"}],
+                "op": "list",
+                "result": 1,
+            }
+        }
+    )
+    return MapContentTrait(map_dps or MapDpsTrait(), maps, command)
 
 
 def _zone_blob() -> str:
@@ -92,7 +103,18 @@ def test_q10_position_is_available_as_top_level_cli_command() -> None:
 
 class _FakeQ10Properties:
     def __init__(self) -> None:
-        self.map = _map_trait()
+        command = cast(CommandTrait, Mock(spec=CommandTrait))
+        self.maps = MapsTrait(command)
+        self.maps.update_from_dps(
+            {
+                B01_Q10_DP.MULTI_MAP: {
+                    "data": [{"id": "12345"}],
+                    "op": "list",
+                    "result": 1,
+                }
+            }
+        )
+        self.map = MapContentTrait(MapDpsTrait(), self.maps, command)
         self.refresh_count = 0
 
         async def refresh_map() -> None:
@@ -112,6 +134,34 @@ async def refresh_map() -> None:
         self.map.refresh = refresh_map  # type: ignore[method-assign]
 
 
+class _FakeQ10PropertiesWithoutMapId:
+    def __init__(self) -> None:
+        command = cast(CommandTrait, Mock(spec=CommandTrait))
+        self.maps = MapsTrait(command)
+        self.map = MapContentTrait(MapDpsTrait(), self.maps, command)
+        self.maps_refresh_count = 0
+        self.map_refresh_count = 0
+
+        async def refresh_maps() -> None:
+            self.maps_refresh_count += 1
+            self.maps.update_from_dps(
+                {
+                    B01_Q10_DP.MULTI_MAP: {
+                        "data": [{"id": "12345"}],
+                        "op": "list",
+                        "result": 1,
+                    }
+                }
+            )
+
+        async def refresh_map() -> None:
+            self.map_refresh_count += 1
+            self.map.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()))
+
+        self.maps.refresh = refresh_maps  # type: ignore[method-assign]
+        self.map.refresh = refresh_map  # type: ignore[method-assign]
+
+
 async def test_await_q10_map_push_waits_for_fresh_update() -> None:
     """A cached trace alone is not treated as a successful new map push."""
     properties = _FakeQ10Properties()
@@ -142,6 +192,22 @@ async def test_await_q10_map_push_returns_true_after_update() -> None:
     assert len(properties.map.path) == 14
 
 
+async def test_await_q10_map_push_requests_map_list_only_on_first_use() -> None:
+    """Content gets the list first only when no stored map ID is available."""
+    properties = _FakeQ10PropertiesWithoutMapId()
+
+    got_trace = await _await_q10_map_push(
+        cast(Q10PropertiesApi, properties),
+        lambda: bool(properties.map.path),
+        properties.map._add_trace_packet_listener,
+        timeout=0.01,
+    )
+
+    assert got_trace is True
+    assert properties.maps_refresh_count == 1
+    assert properties.map_refresh_count == 1
+
+
 async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> None:
     properties = _FakeQ10Properties()
     properties.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes()))
@@ -162,7 +228,7 @@ async def test_await_q10_map_push_ignores_overlay_only_render() -> None:
     """A DPS recomposition cannot masquerade as a fresh map packet."""
     map_dps = MapDpsTrait()
     properties = _FakeQ10Properties()
-    properties.map = MapContentTrait(map_dps)
+    properties.map = _map_trait(map_dps)
     properties.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes()))
 
     async def refresh_overlay_only() -> None:
@@ -242,13 +308,13 @@ async def test_subscribe_loop_routes_trace_push(
     assert q10_api.map.robot_position is not None
 
 
-async def test_map_list_response_requests_first_map(
+async def test_map_list_and_content_refresh_are_independent(
     q10_api: Q10PropertiesApi,
     mock_channel: FakeB01Q10Channel,
     message_queue: asyncio.Queue[Q10Message],
 ) -> None:
-    """The map trait follows a typed map list with a get for its first ID."""
-    await q10_api.map.refresh()
+    """A list update stores an ID but does not request map content."""
+    await q10_api.maps.refresh()
     assert mock_channel.published_commands == [
         (
             B01_Q10_DP.COMMON,
@@ -271,7 +337,11 @@ async def test_map_list_response_requests_first_map(
         )
     )
 
-    await _wait_for(lambda: len(mock_channel.published_commands) == 2)
+    await _wait_for(lambda: q10_api.maps.current_map_id == "12345")
+    assert len(mock_channel.published_commands) == 1
+
+    await q10_api.map.refresh()
+
     assert mock_channel.published_commands[1] == (
         B01_Q10_DP.COMMON,
         {
@@ -281,16 +351,15 @@ async def test_map_list_response_requests_first_map(
             }
         },
     )
-    assert q10_api.map.multi_map is not None
-    assert q10_api.map.multi_map.current_map_id == "12345"
+    assert q10_api.maps.current_map_id == "12345"
 
 
-async def test_map_list_response_without_map_id_is_ignored(
+async def test_empty_map_list_does_not_request_content(
     q10_api: Q10PropertiesApi,
     mock_channel: FakeB01Q10Channel,
     message_queue: asyncio.Queue[Q10Message],
 ) -> None:
-    """A typed list response with no entries cannot trigger a map get."""
+    """An empty list leaves map content unavailable."""
     message_queue.put_nowait(
         Q10DpsUpdate(
             dps={
@@ -304,11 +373,27 @@ async def test_map_list_response_without_map_id_is_ignored(
     )
 
     await asyncio.sleep(0.01)
+    assert q10_api.maps.current_map_id is None
     assert mock_channel.published_commands == []
 
 
-async def test_map_refresh_requests_are_not_rate_limited(q10_api: Q10PropertiesApi) -> None:
-    """The caller controls map cadence; each refresh emits its own list request."""
+async def test_map_content_refresh_requires_stored_map_id(q10_api: Q10PropertiesApi) -> None:
+    """Content cannot be requested until the map list supplies an ID."""
+    with pytest.raises(RoborockException, match="map list is available"):
+        await q10_api.map.refresh()
+
+
+async def test_map_content_refresh_requests_are_not_rate_limited(q10_api: Q10PropertiesApi) -> None:
+    """The caller controls content cadence; each refresh sends a get request."""
+    q10_api.maps.update_from_dps(
+        {
+            B01_Q10_DP.MULTI_MAP: {
+                "data": [{"id": "12345"}],
+                "op": "list",
+                "result": 1,
+            }
+        }
+    )
     with patch.object(q10_api.command, "send") as send:
         await q10_api.map.refresh()
         await q10_api.map.refresh()
@@ -316,33 +401,33 @@ async def test_map_refresh_requests_are_not_rate_limited(q10_api: Q10PropertiesA
     assert send.await_count == 2
 
 
-async def test_failed_map_get_does_not_stop_subscription(
-    q10_api: Q10PropertiesApi,
-    message_queue: asyncio.Queue[Q10Message],
-) -> None:
-    """An expected command failure cannot stop later map-protocol updates."""
-
-    async def send_command(_dp, params) -> None:
-        if params[str(B01_Q10_DP.MULTI_MAP.code)]["op"] == "get":
-            raise RoborockException("device rejected map request")
-
-    with patch.object(q10_api.command, "send", side_effect=send_command):
-        message_queue.put_nowait(
-            Q10DpsUpdate(
-                dps={
-                    B01_Q10_DP.MULTI_MAP: {
-                        "data": [{"id": "12345"}],
-                        "op": "list",
-                        "result": 1,
-                    }
+def test_map_get_ack_does_not_replace_saved_map_list(q10_api: Q10PropertiesApi) -> None:
+    """A content acknowledgement cannot remove the stored map ID."""
+    q10_api._handle_message(
+        Q10DpsUpdate(
+            dps={
+                B01_Q10_DP.MULTI_MAP: {
+                    "data": [{"id": "12345"}],
+                    "op": "list",
+                    "result": 1,
                 }
-            )
+            }
         )
-        message_queue.put_nowait(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()))
+    )
 
-        await _wait_for(lambda: bool(q10_api.map.path))
+    q10_api._handle_message(
+        Q10DpsUpdate(
+            dps={
+                B01_Q10_DP.MULTI_MAP: {
+                    "data": [],
+                    "op": "get",
+                    "result": 1,
+                }
+            }
+        )
+    )
 
-    assert q10_api.map.robot_position is not None
+    assert q10_api.maps.current_map_id == "12345"
 
 
 # --- Source composition + rendering ------------------------------------------
@@ -379,7 +464,7 @@ def test_render_failure_clears_stale_image() -> None:
 def test_map_dps_update_renders_decoded_overlays() -> None:
     """A DPS update recomposes an existing map with decoded overlays."""
     map_dps = MapDpsTrait()
-    trait = MapContentTrait(map_dps)
+    trait = _map_trait(map_dps)
     packet = parse_map_packet(FIXTURE.read_bytes())
     notified: list[None] = []
     trait.add_update_listener(lambda: notified.append(None))
@@ -404,7 +489,7 @@ def test_map_dps_update_renders_decoded_overlays() -> None:
 def test_map_dps_blobs_are_decoded_only_when_dps_arrives() -> None:
     """Map and trace renders reuse the overlays decoded by the DPS trait."""
     map_dps = MapDpsTrait()
-    trait = MapContentTrait(map_dps)
+    trait = _map_trait(map_dps)
 
     with (
         patch("roborock.devices.traits.b01.q10.map.parse_zone_blob", return_value=[]) as parse_zones,
@@ -432,7 +517,7 @@ def test_load_overlays_partial_update_keeps_existing_zones() -> None:
 def test_map_dps_update_without_map_does_not_notify_map_content() -> None:
     """A DPS update cannot change high-level content before a map arrives."""
     map_dps = MapDpsTrait()
-    trait = MapContentTrait(map_dps)
+    trait = _map_trait(map_dps)
     notified = []
     trait.add_update_listener(lambda: notified.append(True))
 
@@ -445,7 +530,7 @@ def test_map_dps_update_without_map_does_not_notify_map_content() -> None:
 def test_map_dps_push_without_overlay_data_points_is_noop() -> None:
     """A DPS push carrying neither overlay DP leaves both traits untouched."""
     map_dps = MapDpsTrait()
-    trait = MapContentTrait(map_dps)
+    trait = _map_trait(map_dps)
     notified = []
     trait.add_update_listener(lambda: notified.append(True))