diff --git a/roborock/devices/device.py b/roborock/devices/device.py
index 33509c34..084145ef 100644
--- a/roborock/devices/device.py
+++ b/roborock/devices/device.py
@@ -202,6 +202,8 @@ async def connect(self) -> None:
                 await self.v1_properties.start()
             elif self.b01_q10_properties is not None:
                 await self.b01_q10_properties.start()
+            elif self.b01_q7_properties is not None:
+                await self.b01_q7_properties.start()
             elif self.zeo is not None:
                 await self.zeo.start()
         except RoborockException:
@@ -232,6 +234,8 @@ async def close(self) -> None:
             self.v1_properties.close()
         if self.b01_q10_properties is not None:
             await self.b01_q10_properties.close()
+        if self.b01_q7_properties is not None:
+            await self.b01_q7_properties.close()
         if self.zeo is not None:
             self.zeo.close()
         if self._unsub:
diff --git a/roborock/devices/rpc/b01_q7_channel.py b/roborock/devices/rpc/b01_q7_channel.py
index e2bc7b12..d01257b3 100644
--- a/roborock/devices/rpc/b01_q7_channel.py
+++ b/roborock/devices/rpc/b01_q7_channel.py
@@ -53,6 +53,10 @@ async def send_map_command(
         """Send a map command and get decoded bytes."""
         ...
 
+    async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]:
+        """Subscribe to unsolicited map pushes, invoking callback with decoded SCMap bytes."""
+        ...
+
 
 def _matches_map_response(response_message: RoborockMessage, *, version: bytes | None) -> bytes | None:
     """Return raw map payload bytes for matching MAP_RESPONSE messages."""
@@ -208,6 +212,25 @@ async def send_map_command(
 
         return decode_map_payload(raw_payload, map_key=self._map_key)
 
+    async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]:
+        """Subscribe to unsolicited ``MAP_RESPONSE`` pushes.
+
+        The device streams full SCMap frames on its own during cleaning; the
+        callback receives the decoded (inflated) SCMap bytes for each frame.
+        """
+
+        def on_message(message: RoborockMessage) -> None:
+            if (raw_payload := _matches_map_response(message, version=B01_VERSION)) is None:
+                return
+            try:
+                decoded = decode_map_payload(raw_payload, map_key=self._map_key)
+            except RoborockException as ex:
+                _LOGGER.debug("Failed to decode pushed B01 map payload: %s", ex)
+                return
+            callback(decoded)
+
+        return await self._mqtt_channel.subscribe(on_message)
+
 
 def create_b01_q7_channel(
     device: HomeDataDevice,
diff --git a/roborock/devices/traits/b01/q7/__init__.py b/roborock/devices/traits/b01/q7/__init__.py
index 35a29144..c19735b8 100644
--- a/roborock/devices/traits/b01/q7/__init__.py
+++ b/roborock/devices/traits/b01/q7/__init__.py
@@ -3,6 +3,7 @@
 Potentially other devices may fall into this category in the future.
 """
 
+from collections.abc import Callable
 from typing import Any
 
 from roborock import B01Props
@@ -71,6 +72,19 @@ def __init__(
             self._map_rpc_channel,
             self.map,
         )
+        self._unsub_map_pushes: Callable[[], None] | None = None
+
+    async def start(self) -> None:
+        """Start listening for unsolicited map pushes from the device."""
+        if self._unsub_map_pushes is not None:
+            return
+        self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push)
+
+    async def close(self) -> None:
+        """Stop listening for unsolicited map pushes."""
+        if self._unsub_map_pushes is not None:
+            self._unsub_map_pushes()
+            self._unsub_map_pushes = None
 
     async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
         """Query the device for the values of the given Q7 properties."""
diff --git a/roborock/devices/traits/b01/q7/map_content.py b/roborock/devices/traits/b01/q7/map_content.py
index 0becf91a..6c55aeed 100644
--- a/roborock/devices/traits/b01/q7/map_content.py
+++ b/roborock/devices/traits/b01/q7/map_content.py
@@ -9,6 +9,7 @@
 """
 
 import asyncio
+import logging
 from dataclasses import dataclass
 
 from vacuum_map_parser_base.map_data import MapData
@@ -16,13 +17,16 @@
 from roborock.data import RoborockBase
 from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel
 from roborock.devices.traits import Trait
+from roborock.devices.traits.common import TraitUpdateListener
 from roborock.exceptions import RoborockException
-from roborock.map.b01_map_parser import B01MapParser, B01MapParserConfig
+from roborock.map.b01_map_parser import B01MapParser, B01MapParserConfig, parse_map_type
 from roborock.roborock_typing import RoborockB01Q7Methods
 
 from .map import MapTrait
 
+_LOGGER = logging.getLogger(__name__)
 _TRUNCATE_LENGTH = 20
+_LIVE_MAP_TYPE = 0
 
 
 @dataclass
@@ -49,7 +53,7 @@ def __repr__(self) -> str:
         return f"MapContent(image_content={img!r}, map_data={self.map_data!r})"
 
 
-class MapContentTrait(MapContent, Trait):
+class MapContentTrait(MapContent, Trait, TraitUpdateListener):
     """Trait for fetching parsed map content for Q7 devices."""
 
     def __init__(
@@ -59,7 +63,8 @@ def __init__(
         *,
         map_parser_config: B01MapParserConfig | None = None,
     ) -> None:
-        super().__init__()
+        MapContent.__init__(self)
+        TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._map_rpc_channel = map_rpc_channel
         self._map_trait = map_trait
         self._map_parser = B01MapParser(map_parser_config)
@@ -82,6 +87,34 @@ async def refresh(self) -> None:
                 {"map_id": map_id},
             )
 
+        self._parse_and_store(raw_payload)
+
+    def update_from_push(self, raw_payload: bytes) -> None:
+        """Store an unsolicited SCMap frame pushed by the device during cleaning.
+
+        Pushed frames carry the live robot pose and cleaning path, so the
+        rendered image stays current without polling. Frames for other maps
+        than the live one are ignored to not overwrite the current map.
+        """
+        try:
+            map_type = parse_map_type(raw_payload)
+        except RoborockException as ex:
+            _LOGGER.debug("Failed to parse pushed B01 map frame: %s", ex)
+            return
+
+        if map_type != _LIVE_MAP_TYPE:
+            _LOGGER.debug("Ignoring pushed B01 map frame of type %s", map_type)
+            return
+
+        try:
+            self._parse_and_store(raw_payload)
+        except RoborockException as ex:
+            _LOGGER.debug("Failed to parse pushed B01 map frame: %s", ex)
+            return
+        self._notify_update()
+
+    def _parse_and_store(self, raw_payload: bytes) -> None:
+        """Parse decoded SCMap bytes and update the cached fields."""
         try:
             parsed_data = self._map_parser.parse(raw_payload)
         except RoborockException:
diff --git a/roborock/map/b01_map_parser.py b/roborock/map/b01_map_parser.py
index b57912e4..b974f631 100644
--- a/roborock/map/b01_map_parser.py
+++ b/roborock/map/b01_map_parser.py
@@ -65,6 +65,15 @@ def parse(self, payload: bytes) -> ParsedMapData:
         )
 
 
+def parse_map_type(payload: bytes) -> int:
+    """Return the SCMap frame's map type without rendering the image.
+
+    Type 0 is the live map of the current cleaning run, other types are
+    historic or saved maps.
+    """
+    return _parse_scmap_payload(payload).mapType
+
+
 def _parse_scmap_payload(payload: bytes) -> RobotMap:
     """Parse inflated SCMap bytes into a generated protobuf message."""
     parsed = RobotMap()
diff --git a/tests/devices/rpc/test_b01_q7_channel.py b/tests/devices/rpc/test_b01_q7_channel.py
index 21c6d31a..087291d9 100644
--- a/tests/devices/rpc/test_b01_q7_channel.py
+++ b/tests/devices/rpc/test_b01_q7_channel.py
@@ -293,3 +293,60 @@ async def test_send_command_general_exception(
 
     with pytest.raises(RuntimeError, match="Generic publish crash"):
         await channel.send_command("prop.get", {"property": ["status"]})
+
+
+async def test_subscribe_map_pushes_delivers_decoded_frames(
+    device: HomeDataDevice,
+    product: HomeDataProduct,
+    fake_channel: FakeChannel,
+    message_builder: B01MessageBuilder,
+) -> None:
+    """Unsolicited MAP_RESPONSE frames are decoded and delivered to the callback."""
+    channel = create_b01_q7_channel(device, product, fake_channel)  # type: ignore[arg-type]
+
+    received: list[bytes] = []
+    unsub = await channel.subscribe_map_pushes(received.append)
+
+    with patch(
+        "roborock.devices.rpc.b01_q7_channel.decode_map_payload",
+        return_value=b"inflated-payload",
+    ):
+        fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload"))
+
+    assert received == [b"inflated-payload"]
+
+    # Non-map messages are filtered out.
+    fake_channel.notify_subscribers(message_builder.build({"status": 1}))
+    assert received == [b"inflated-payload"]
+
+    # After unsubscribing, further frames are not delivered.
+    unsub()
+    with patch(
+        "roborock.devices.rpc.b01_q7_channel.decode_map_payload",
+        return_value=b"inflated-payload",
+    ):
+        fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload"))
+    assert received == [b"inflated-payload"]
+
+
+async def test_subscribe_map_pushes_skips_undecodable_frames(
+    device: HomeDataDevice,
+    product: HomeDataProduct,
+    fake_channel: FakeChannel,
+    message_builder: B01MessageBuilder,
+) -> None:
+    """Frames that fail map decoding are skipped without breaking the subscription."""
+    channel = create_b01_q7_channel(device, product, fake_channel)  # type: ignore[arg-type]
+
+    received: list[bytes] = []
+    await channel.subscribe_map_pushes(received.append)
+
+    fake_channel.notify_subscribers(message_builder.build_map_response(b"!!! not base64 !!!"))
+    assert received == []
+
+    with patch(
+        "roborock.devices.rpc.b01_q7_channel.decode_map_payload",
+        return_value=b"inflated-payload",
+    ):
+        fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload"))
+    assert received == [b"inflated-payload"]
diff --git a/tests/devices/traits/b01/q7/conftest.py b/tests/devices/traits/b01/q7/conftest.py
index efc9d166..03f36db6 100644
--- a/tests/devices/traits/b01/q7/conftest.py
+++ b/tests/devices/traits/b01/q7/conftest.py
@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from typing import Any
 
 import pytest
@@ -14,6 +15,8 @@ def __init__(self) -> None:
         self.published_commands: list[tuple[Any, Any]] = []
         self.response_queue: list[Any] = []
         self.side_effect: Exception | None = None
+        self.map_push_callback: Callable[[bytes], None] | None = None
+        self.map_push_subscribe_count = 0
 
     async def send_command(self, command: Any, params: Any = None) -> Any:
         if self.side_effect:
@@ -29,6 +32,15 @@ async def send_map_command(self, command: Any, params: Any = None) -> bytes:
             return self.response_queue.pop(0)
         return b""
 
+    async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]:
+        self.map_push_subscribe_count += 1
+        self.map_push_callback = callback
+
+        def unsub() -> None:
+            self.map_push_callback = None
+
+        return unsub
+
 
 @pytest.fixture(name="fake_channel")
 def fake_channel_fixture() -> FakeQ7Channel:
diff --git a/tests/devices/traits/b01/q7/test_map_content.py b/tests/devices/traits/b01/q7/test_map_content.py
index b8753d77..7582580d 100644
--- a/tests/devices/traits/b01/q7/test_map_content.py
+++ b/tests/devices/traits/b01/q7/test_map_content.py
@@ -6,11 +6,17 @@
 from roborock.devices.traits.b01.q7 import Q7PropertiesApi
 from roborock.exceptions import RoborockException
 from roborock.map.b01_map_parser import ParsedMapData
+from roborock.map.proto.b01_scmap_pb2 import RobotMap  # type: ignore[attr-defined]
 from roborock.roborock_typing import RoborockB01Q7Methods
 
 from .conftest import FakeQ7Channel
 
 
+def scmap_frame(map_type: int = 0) -> bytes:
+    """Serialize a minimal SCMap frame of the given map type."""
+    return RobotMap(mapType=map_type).SerializeToString()
+
+
 async def test_q7_map_content_refresh_populates_cached_values(
     q7_api: Q7PropertiesApi,
     fake_channel: FakeQ7Channel,
@@ -87,3 +93,114 @@ async def test_q7_map_content_refresh_errors_without_map_list(
 
     with pytest.raises(RoborockException, match="Unable to determine current map ID"):
         await q7_api.map_content.refresh()
+
+
+async def test_q7_map_content_updates_from_push(
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeQ7Channel,
+):
+    """Unsolicited map pushes update the cached map and notify listeners."""
+    await q7_api.start()
+    assert fake_channel.map_push_callback is not None
+
+    updates: list[bool] = []
+    q7_api.map_content.add_update_listener(lambda: updates.append(True))
+
+    dummy_map_data = MapData()
+    parsed_map_data = ParsedMapData(
+        image_content=b"pngbytes",
+        map_data=dummy_map_data,
+    )
+    pushed_payload = scmap_frame()
+    with patch(
+        "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse",
+        return_value=parsed_map_data,
+    ):
+        fake_channel.map_push_callback(pushed_payload)
+
+    assert q7_api.map_content.image_content == b"pngbytes"
+    assert q7_api.map_content.raw_api_response == pushed_payload
+    assert updates == [True]
+
+    await q7_api.close()
+    assert fake_channel.map_push_callback is None
+
+
+async def test_q7_map_content_push_parse_failure_keeps_previous_map(
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeQ7Channel,
+):
+    """A malformed pushed frame is dropped without clearing cached content."""
+    await q7_api.start()
+    assert fake_channel.map_push_callback is not None
+
+    dummy_map_data = MapData()
+    parsed_map_data = ParsedMapData(
+        image_content=b"pngbytes",
+        map_data=dummy_map_data,
+    )
+    good_payload = scmap_frame()
+    with patch(
+        "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse",
+        return_value=parsed_map_data,
+    ):
+        fake_channel.map_push_callback(good_payload)
+
+    updates: list[bool] = []
+    q7_api.map_content.add_update_listener(lambda: updates.append(True))
+
+    fake_channel.map_push_callback(b"not a map")
+
+    assert q7_api.map_content.image_content == b"pngbytes"
+    assert q7_api.map_content.raw_api_response == good_payload
+    assert updates == []
+    await q7_api.close()
+
+
+async def test_q7_map_content_push_ignores_non_live_map(
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeQ7Channel,
+):
+    """Only frames of the live map may update the cached map."""
+    await q7_api.start()
+    assert fake_channel.map_push_callback is not None
+
+    dummy_map_data = MapData()
+    parsed_map_data = ParsedMapData(
+        image_content=b"pngbytes",
+        map_data=dummy_map_data,
+    )
+    good_payload = scmap_frame()
+    with patch(
+        "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse",
+        return_value=parsed_map_data,
+    ):
+        fake_channel.map_push_callback(good_payload)
+
+    updates: list[bool] = []
+    q7_api.map_content.add_update_listener(lambda: updates.append(True))
+
+    with patch(
+        "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse",
+        return_value=ParsedMapData(image_content=b"othermap", map_data=MapData()),
+    ) as parse:
+        fake_channel.map_push_callback(scmap_frame(map_type=1))
+
+    parse.assert_not_called()
+    assert q7_api.map_content.image_content == b"pngbytes"
+    assert q7_api.map_content.raw_api_response == good_payload
+    assert updates == []
+    await q7_api.close()
+
+
+async def test_q7_start_is_idempotent(
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeQ7Channel,
+):
+    """Repeated start() calls must not leak additional subscriptions."""
+    await q7_api.start()
+    await q7_api.start()
+
+    assert fake_channel.map_push_subscribe_count == 1
+    await q7_api.close()
+    assert fake_channel.map_push_callback is None