@@ -45,7 +45,7 @@
 
 if TYPE_CHECKING:
     from ably.pubsub.objects.realtimeobject import RealtimeObject
-    from ably.pubsub.objects.valuetypes import LiveMapValue, Primitive, T
+    from ably.pubsub.objects.valuetypes import JsonValue, LiveMapValue, Primitive, T
 
     # RTBC16b: builds the ObjectMessages of one queued write when the batch is flushed
     MessageConstructor = Callable[[], Awaitable[list[ObjectMessage]]]
@@ -200,13 +200,7 @@ def get(self, key: str) -> BatchContext | None:
             return None
         return self._root_context.wrap_instance(instance)  # RTBC4e
 
-    def compact(self) -> Any:
-        """RTBC10: `Instance.compact` on the wrapped instance."""
-        self._realtime_object._check_access_preconditions()  # RTBC10a
-        self._throw_if_closed()  # RTBC10b
-        return self._instance.compact()  # RTBC10c
-
-    def compact_json(self) -> Any:
+    def compact_json(self) -> JsonValue:
         """RTBC11: `Instance.compact_json` on the wrapped instance."""
         self._realtime_object._check_access_preconditions()  # RTBC11a
         self._throw_if_closed()  # RTBC11b
 
@@ -29,7 +29,7 @@
     from ably.pubsub.objects.liveobject import LiveObjectUpdate
     from ably.pubsub.objects.realtimeobject import RealtimeObject
     from ably.pubsub.objects.subscription import Subscription
-    from ably.pubsub.objects.valuetypes import LiveMapValue, Primitive, T
+    from ably.pubsub.objects.valuetypes import JsonObject, JsonValue, LiveMapValue, Primitive, T
 
 # The types `value(expected)` accepts as `expected`, and the category each one selects
 _EXPECTED_VALUE_TYPES: tuple[tuple[type, ValueType], ...] = (
@@ -103,36 +103,37 @@ def primitive_value(value: Any, value_type: ValueType | None = None) -> Any:
     return value
 
 
-def compact_value(value: Any, *, for_json: bool = False) -> Any:
-    """RTPO13, RTPO14: a plain snapshot of a resolved value.
+def compact_json_value(value: Any) -> JsonValue:
+    """RTPO14, RTINS11: a JSON-serializable snapshot of a resolved value, a map as
+    `compact_json_map` builds it."""
+    if isinstance(value, InternalLiveMap):
+        return compact_json_map(value)
+    return _compact_json_leaf(value)
 
-    A map becomes a dict of its entries that are not tombstoned (RTPO13c1), nested maps
-    becoming nested dicts (RTPO13c2) and counters their values (RTPO13c3). Maps are visited
-    depth first in entry order, and a map reached again in the same snapshot, through a
-    cycle or a second reference, is the dict already built for it (RTPO13c5). With
-    `for_json` it is `{'objectId': id}` instead (RTPO14b2), and binary is a base64 string
-    (RTPO14b1), so that the result is JSON-serializable.
-    """
-    if not isinstance(value, InternalLiveMap):
-        return _compact_leaf(value, for_json)
 
-    result: dict[str, Any] = {}
-    built: dict[str, dict[str, Any]] = {value.object_id: result}
+def compact_json_map(live_map: InternalLiveMap) -> JsonObject:
+    """RTPO13c, RTPO14b: a JSON object of a map's entries that are not tombstoned (RTPO13c1).
+
+    Nested maps become nested objects (RTPO13c2) and counters their values (RTPO13c3). Maps
+    are visited depth first in entry order, and a map reached again in the same snapshot,
+    through a cycle or a second reference, is `{'objectId': id}` (RTPO13c5, RTPO14b2).
+    """
+    result: JsonObject = {}
+    visited = {live_map.object_id}
     # A stack in place of recursion, so that the depth of the tree is not limited by the
-    # interpreter's. Each item is a dict being built and the entries of its map still to add.
-    stack = [(result, iter(value.entries()))]
+    # interpreter's. Each item is an object being built and the entries of its map still to add.
+    stack = [(result, iter(live_map.entries()))]
     while stack:
         target, entries = stack[-1]
         for key, entry_value in entries:
             if not isinstance(entry_value, InternalLiveMap):
-                target[key] = _compact_leaf(entry_value, for_json)
+                target[key] = _compact_json_leaf(entry_value)
                 continue
-            seen = built.get(entry_value.object_id)
-            if seen is not None:
-                target[key] = {'objectId': entry_value.object_id} if for_json else seen
+            if entry_value.object_id in visited:
+                target[key] = {'objectId': entry_value.object_id}
                 continue
-            child: dict[str, Any] = {}
-            built[entry_value.object_id] = child
+            child: JsonObject = {}
+            visited.add(entry_value.object_id)
             target[key] = child
             stack.append((child, iter(entry_value.entries())))
             break
@@ -141,11 +142,14 @@ def compact_value(value: Any, *, for_json: bool = False) -> Any:
     return result
 
 
-def _compact_leaf(value: Any, for_json: bool) -> Any:
-    """RTPO13c3, RTPO13c4, RTPO13d, RTPO13e, RTPO14b1: the snapshot of a value that is not a map."""
+def _compact_json_leaf(value: Any) -> JsonValue:
+    """RTPO13c3, RTPO13c4, RTPO13d, RTPO13e, RTPO14b1: the snapshot of a value that is not a map.
+
+    A counter is its value, binary a base64 string, and a JSON object or array a copy.
+    """
     if isinstance(value, InternalLiveCounter):
         return value.value()
-    if for_json and isinstance(value, (bytes, bytearray)):
+    if isinstance(value, (bytes, bytearray)):
         return base64.b64encode(value).decode('ascii')
     if isinstance(value, (dict, list)):
         return copy.deepcopy(value)
@@ -208,15 +212,11 @@ def get(self, key: str) -> Instance | None:
             return None
         return Instance._wrap(self._realtime_object, value)  # RTINS5c
 
-    def compact(self) -> Any:
-        """RTINS10: a plain snapshot of the wrapped value."""
-        self._realtime_object._check_access_preconditions()  # RTINS10a
-        return compact_value(self._value)  # RTINS10b
-
-    def compact_json(self) -> Any:
-        """RTINS11: `compact`, with binary as base64 and cycles as `{'objectId': ...}`. Never None."""
+    def compact_json(self) -> JsonValue:
+        """RTINS11: a JSON-serializable snapshot of the wrapped value, as
+        `PathObject.compact_json` builds it. Never None (RTINS11c)."""
         self._realtime_object._check_access_preconditions()  # RTINS11a
-        return compact_value(self._value, for_json=True)  # RTINS11b
+        return compact_json_value(self._value)  # RTINS11b
 
     def as_live_map(self) -> LiveMapInstance:
         """RTTS9a: this instance as a map. Raises AblyException 92007 if it wraps something else."""
@@ -266,6 +266,11 @@ def batch(self) -> Batch[LiveMapBatchContext]:
 
         return Batch(self._realtime_object, lambda: self._value, LiveMapBatchContext, f'object {self.id!r}')
 
+    def compact_json(self) -> JsonObject:
+        """RTINS11, RTTS7a1: a JSON-serializable snapshot of the wrapped map."""
+        self._realtime_object._check_access_preconditions()  # RTINS11a
+        return compact_json_map(self._value)  # RTINS11b
+
     def entries(self) -> list[tuple[str, Instance]]:
         """RTINS6: `(key, instance)` for each entry of the map.
 
@@ -332,6 +337,11 @@ def value(self) -> float:
         self._realtime_object._check_access_preconditions()  # RTINS4a
         return self._value.value()
 
+    def compact_json(self) -> float:
+        """RTINS11, RTTS7a3: the wrapped counter's value."""
+        self._realtime_object._check_access_preconditions()  # RTINS11a
+        return self._value.value()  # RTINS11b
+
     async def increment(self, amount: float = 1) -> None:
         """RTINS14: increments the wrapped counter by `amount`."""
         self._realtime_object._check_write_preconditions()  # RTINS14b
 
@@ -22,7 +22,7 @@
 
 from ably.pubsub.objects.instance import (
     Instance,
-    compact_value,
+    compact_json_value,
     expected_value_type,
     primitive_value,
     value_type_of,
@@ -37,7 +37,7 @@
     from ably.pubsub.objects.enums import ValueType
     from ably.pubsub.objects.realtimeobject import RealtimeObject
     from ably.pubsub.objects.subscription import Subscription
-    from ably.pubsub.objects.valuetypes import LiveMapValue, Primitive, T
+    from ably.pubsub.objects.valuetypes import JsonValue, LiveMapValue, Primitive, T
 
 
 @dataclass
@@ -112,21 +112,19 @@ def instance(self) -> Instance | None:
             return None  # RTPO8e
         return Instance._wrap(self._realtime_object, resolved)  # RTPO8c, RTPO8f
 
-    def compact(self) -> Any:
-        """RTPO13: a plain snapshot of what the path resolves to, or None."""
-        self._realtime_object._check_access_preconditions()  # RTPO13a
-        resolved = self._resolve()  # RTPO13b
-        if resolved is None:
-            return None  # RTPO13f
-        return compact_value(resolved)  # RTPO13c, RTPO13d, RTPO13e
+    def compact_json(self) -> JsonValue:
+        """RTPO14: a JSON-serializable snapshot of what the path resolves to, or None.
 
-    def compact_json(self) -> Any:
-        """RTPO14: `compact`, with binary as base64 and cycles as `{'objectId': ...}`."""
+        A map is a JSON object of its entries that are not tombstoned, nested maps becoming
+        nested objects and counters their values. Binary is a base64 string, and a map
+        reached again in the same snapshot, through a cycle or a second reference, is
+        `{'objectId': id}`.
+        """
         self._realtime_object._check_access_preconditions()  # RTPO14a
-        resolved = self._resolve()
+        resolved = self._resolve()  # RTPO13b
         if resolved is None:
-            return None
-        return compact_value(resolved, for_json=True)  # RTPO14b
+            return None  # RTPO13f
+        return compact_json_value(resolved)  # RTPO14b
 
     def subscribe(self, listener: Callable[[PathObjectSubscriptionEvent], None], *,
                   depth: int | None = None) -> Subscription:
 
@@ -13,7 +13,7 @@
 import json
 import math
 from dataclasses import dataclass
-from typing import Any, TypeVar, Union
+from typing import Any, Dict, List, TypeVar, Union
 
 from ably.pubsub.objects.objectid import generate_nonce, generate_object_id
 from ably.pubsub.objects.objectmessage import (
@@ -37,6 +37,10 @@
 
 T = TypeVar('T', bound=Primitive)
 
+# A JSON value and a JSON object, as `compact_json` returns them (RTPO14, RTINS11)
+JsonValue = Union[str, float, bool, None, List['JsonValue'], Dict[str, 'JsonValue']]
+JsonObject = Dict[str, JsonValue]
+
 
 @dataclass(frozen=True)
 class LiveCounter:
 
@@ -48,7 +48,14 @@
 from ably.pubsub.objects.publicmessage import ObjectMessage, ObjectOperation
 from ably.pubsub.objects.realtimeobject import RealtimeObject
 from ably.pubsub.objects.subscription import StatusSubscription, Subscription
-from ably.pubsub.objects.valuetypes import LiveCounter, LiveMap, LiveMapValue, Primitive
+from ably.pubsub.objects.valuetypes import (
+    JsonObject,
+    JsonValue,
+    LiveCounter,
+    LiveMap,
+    LiveMapValue,
+    Primitive,
+)
 from ably.pubsub.prototypes import PubSubHttpClient, PubSubRealtimeClient
 from ably.pubsub.realtime.channel import Channels as RealtimeChannels
 from ably.pubsub.realtime.connection import Connection
@@ -291,6 +298,8 @@ def create_realtime_client(**kwargs) -> PubSubRealtimeClient:
     'IncompatibleClientIdException',
     'Instance',
     'InstanceSubscriptionEvent',
+    'JsonObject',
+    'JsonValue',
     'LiveCounter',
     'LiveCounterBatchContext',
     'LiveCounterInstance',
 
@@ -574,14 +574,13 @@ async def test_rtbc6_rtbc9_map_enumeration():
         assert entries[2][1].get('theme').as_primitive().value() == 'dark'
 
 
-async def test_rtbc10_rtbc11_compact():
-    """RTBC10, RTBC11: `compact` and `compact_json` snapshot the wrapped value as `Instance` does."""
+async def test_rtbc11_compact_json():
+    """RTBC11: `compact_json` snapshots the wrapped value as `Instance` does."""
     (client, channel, root, mock_ws), published = await _synced()
 
     async with root.batch() as ctx:
-        assert ctx.get('profile').compact() == {
+        assert ctx.get('profile').compact_json() == {
             'email': 'alice@example.com', 'nested_counter': 5, 'prefs': {'theme': 'dark'}}
-        assert ctx.compact()['avatar'] == b'\x01\x02\x03'
         assert ctx.compact_json()['avatar'] == 'AQID'
         assert ctx.get('score').compact_json() == 100
 
@@ -635,7 +634,6 @@ class _Contexts(NamedTuple):
 # A call of each context method that reads the objects, and of each that writes
 _READS = {
     'get': lambda c: c.root.get('name'),
-    'compact': lambda c: c.root.compact(),
     'compact_json': lambda c: c.root.compact_json(),
     'entries': lambda c: c.root.entries(),
     'keys': lambda c: c.root.keys(),
 
@@ -1,4 +1,5 @@
-"""The typed views: `type()` and `exists()`, the unchecked path views and the checked instance views.
+"""The typed views: `type()` and `exists()`, the unchecked path views, the checked instance views
+and the snapshots `compact_json` takes through them.
 
 Spec points: RTTS2-RTTS10 (ably/specification#491), as LODR-061 binds them: one
 `PrimitivePathObject` and `PrimitiveInstance` whose `value(expected)` stands for the six
@@ -9,6 +10,7 @@
 Each test drives the standard synced pool over the mock websocket.
 """
 
+import json
 import typing
 
 import pytest
@@ -352,11 +354,45 @@ async def test_rtts10_instance_id_is_a_property():
     assert root.get('name').instance().id is None
 
 
+async def test_rtts7a_instance_compact_json_is_of_the_wrapped_type():
+    """RTINS11, RTTS7a1, RTTS7a3: a map instance's snapshot is a JSON object, a counter
+    instance's is its value as a float, and a primitive instance's is the primitive."""
+    root = await _synced_root()
+
+    profile = root.get('profile').instance().compact_json()
+    score = root.get('score').instance().compact_json()
+
+    assert profile == {'email': 'alice@example.com', 'nested_counter': 5, 'prefs': {'theme': 'dark'}}
+    assert type(score) is float
+    assert score == 100
+    assert root.get('name').instance().compact_json() == 'Alice'
+    assert root.get('avatar').instance().compact_json() == 'AQID'
+
+
+async def test_rtpo14b_compact_json_is_json_serializable():
+    """RTPO14b: binary is a base64 string, and a map reached a second time in one snapshot is
+    its object id, through a second reference as through a cycle, so `json.dumps` takes the
+    snapshot."""
+    client, channel, root, mock_ws = await setup_synced_channel('test')
+    # root.alias references the same map as root.profile
+    mock_ws.send_to_client(build_object_message('test', [
+        build_map_set('root', 'alias', {'objectId': 'map:profile@1000'}, remote_serial(0), 'remote')]))
+    await poll_until(lambda: 'alias' in root.keys(), description='the reference to be applied')
+
+    result = root.compact_json()
+
+    assert result['avatar'] == 'AQID'
+    assert result['profile'] == {'email': 'alice@example.com', 'nested_counter': 5, 'prefs': {'theme': 'dark'}}
+    assert result['alias'] == {'objectId': 'map:profile@1000'}
+    assert json.loads(json.dumps(result)) == result
+
+
 @pytest.mark.parametrize('typed_class, present, absent', [
     (PathObject,
-     {'path', 'type', 'exists', 'get', 'at', 'instance', 'compact', 'compact_json', 'subscribe',
+     {'path', 'type', 'exists', 'get', 'at', 'instance', 'compact_json', 'subscribe',
       'as_live_map', 'as_live_counter', 'as_primitive'},
-     {'value', 'entries', 'keys', 'values', 'size', 'set', 'remove', 'increment', 'decrement', 'batch'}),
+     {'compact', 'value', 'entries', 'keys', 'values', 'size', 'set', 'remove', 'increment', 'decrement',
+      'batch'}),
     (LiveMapPathObject,
      {'entries', 'keys', 'values', 'size', 'set', 'remove', 'batch'},
      {'value', 'increment', 'decrement'}),
@@ -367,8 +403,8 @@ async def test_rtts10_instance_id_is_a_property():
      {'value'},
      {'entries', 'keys', 'values', 'size', 'set', 'remove', 'increment', 'decrement', 'batch'}),
     (Instance,
-     {'id', 'type', 'get', 'compact', 'compact_json', 'as_live_map', 'as_live_counter', 'as_primitive'},
-     {'value', 'entries', 'size', 'set', 'remove', 'increment', 'decrement', 'subscribe', 'batch'}),
+     {'id', 'type', 'get', 'compact_json', 'as_live_map', 'as_live_counter', 'as_primitive'},
+     {'compact', 'value', 'entries', 'size', 'set', 'remove', 'increment', 'decrement', 'subscribe', 'batch'}),
     (LiveMapInstance,
      {'entries', 'keys', 'values', 'size', 'set', 'remove', 'subscribe', 'batch'},
      {'value', 'increment', 'decrement'}),