[AIT-316] feat: enhance annotation handling and protocol integration

- Added support for updating annotation fields (`id`, `connectionId`, `timestamp`) from protocol messages.
- Introduced validation for required annotation fields in `RestAnnotations`.
- Enabled idempotent annotation publishing with auto-generated IDs.
- Improved error handling for annotation processing in `RealtimeChannel`.
- Allowed unsubscribing all annotation listeners when no arguments are provided.

Author: ttypic

ably/realtime/annotations.py

@@ -193,16 +193,17 @@ def unsubscribe(self, *args):
                 Unsubscribe from all annotations on the channel
 
             When no type is provided, arg1 is used as the listener.
+            When no arguments are provided, unsubscribes all annotation listeners (RTAN5).
 
         Raises
         ------
         ValueError
-            If no valid unsubscribe arguments are passed
+            If invalid unsubscribe arguments are passed
         """
+        # RTAN5: Support no arguments to unsubscribe all annotation listeners
         if len(args) == 0:
-            raise ValueError("annotations.unsubscribe called without arguments")
-
-        if len(args) >= 2 and isinstance(args[0], str):
+            self.__subscriptions.off()
+        elif len(args) >= 2 and isinstance(args[0], str):
             annotation_type = args[0]
             listener = args[1]
             self.__subscriptions.off(annotation_type, listener)

ably/realtime/channel.py

@@ -758,11 +758,15 @@ def _on_message(self, proto_msg: dict) -> None:
             self.__presence.set_presence(decoded_presence, is_sync=True, sync_channel_serial=sync_channel_serial)
         elif action == ProtocolMessageAction.ANNOTATION:
             # Handle ANNOTATION messages
+            # RTAN4b: Populate annotation fields from protocol message
+            Annotation.update_inner_annotation_fields(proto_msg)
             annotation_data = proto_msg.get('annotations', [])
             try:
                 annotations = Annotation.from_encoded_array(annotation_data, cipher=self.cipher)
                 # Process annotations through the annotations handler
                 self.annotations._process_incoming(annotations)
+                # RTL15b: Update channel serial for ANNOTATION messages
+                self.__channel_serial = channel_serial
             except Exception as e:
                 log.error(f"Annotation processing error {e}. Skip annotations {annotation_data}")
         elif action == ProtocolMessageAction.ERROR:

ably/rest/annotations.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import uuid
 from urllib import parse
 
 import msgpack
@@ -13,6 +14,7 @@
     make_annotation_response_handler,
 )
 from ably.types.message import Message
+from ably.types.options import Options
 from ably.util.exceptions import AblyException
 
 log = logging.getLogger(__name__)
@@ -73,6 +75,14 @@ def construct_validate_annotation(msg_or_serial, annotation: Annotation) -> Anno
             code=40003,
         )
 
+    # RSAN1a3: Validate that annotation type is specified
+    if not annotation.type:
+        raise AblyException(
+            message='Annotation type must be specified',
+            status_code=400,
+            code=40000,
+        )
+
     return annotation._copy_with(
         message_serial=message_serial,
     )
@@ -83,6 +93,8 @@ class RestAnnotations:
     Provides REST API methods for managing annotations on messages.
     """
 
+    __client_options: Options
+
     def __init__(self, channel):
         """
         Initialize RestAnnotations.
@@ -91,6 +103,7 @@ def __init__(self, channel):
             channel: The REST Channel this annotations instance belongs to
         """
         self.__channel = channel
+        self.__client_options = channel.ably.options
 
     def __base_path_for_serial(self, serial):
         """
@@ -127,6 +140,10 @@ async def publish(
         """
         annotation = construct_validate_annotation(msg_or_serial, annotation)
 
+        # RSAN1c4: Generate random ID if not provided (for idempotent publishing)
+        if not annotation.id and self.__client_options.idempotent_rest_publishing:
+            annotation = annotation._copy_with(id=str(uuid.uuid4()))
+
         # Convert to wire format
         request_body = annotation.as_dict(binary=self.__channel.ably.options.use_binary_protocol)
 
@@ -142,7 +159,7 @@ async def publish(
         # Build path
         path = self.__base_path_for_serial(annotation.message_serial)
         if params:
-            params = {k: str(v).lower() if type(v) is bool else v for k, v in params.items()}
+            params = {k: str(v).lower() if isinstance(v, bool) else v for k, v in params.items()}
             path += '?' + parse.urlencode(params)
 
         # Send request

ably/types/annotation.py

@@ -34,7 +34,9 @@ def __init__(self,
                  count=None,
                  data=None,
                  encoding='',
+                 id=None,
                  client_id=None,
+                 connection_id=None,
                  timestamp=None,
                  extras=None):
         """
@@ -47,7 +49,9 @@ def __init__(self,
             count: Count associated with the annotation
             data: Optional data payload for the annotation
             encoding: Encoding format for the data
+            id: (TAN2a) A unique identifier for this annotation
             client_id: The client ID that created this annotation
+            connection_id: The connection ID that created this annotation
             timestamp: Timestamp of the annotation
             extras: Additional metadata
         """
@@ -60,18 +64,25 @@ def __init__(self,
         self.__action = action if action is not None else AnnotationAction.ANNOTATION_CREATE
         self.__count = count
         self.__data = data
+        self.__id = to_text(id) if id is not None else None
         self.__client_id = to_text(client_id) if client_id is not None else None
+        self.__connection_id = to_text(connection_id) if connection_id is not None else None
         self.__timestamp = timestamp
         self.__extras = extras
         self.__encoding = encoding
 
     def __eq__(self, other):
         if isinstance(other, Annotation):
-            return (self.serial == other.serial
-                    and self.message_serial == other.message_serial
+            # TAN2i: serial is the unique identifier for the annotation
+            # If both have serials, use serial for comparison
+            if self.serial is not None and other.serial is not None:
+                return self.serial == other.serial
+            # Otherwise fall back to comparing multiple fields
+            return (self.message_serial == other.message_serial
                     and self.type == other.type
                     and self.name == other.name
-                    and self.action == other.action)
+                    and self.action == other.action
+                    and self.client_id == other.client_id)
         return NotImplemented
 
     def __ne__(self, other):
@@ -121,6 +132,14 @@ def timestamp(self):
     def extras(self):
         return self.__extras
 
+    @property
+    def id(self):
+        return self.__id
+
+    @property
+    def connection_id(self):
+        return self.__connection_id
+
     def as_dict(self, binary=False):
         """
         Convert annotation to dictionary format for API communication.
@@ -134,7 +153,9 @@ def as_dict(self, binary=False):
             'type': self.type,  # Annotation type (not data type)
             'name': self.name,
             'count': self.count,
+            'id': self.id or None,
             'clientId': self.client_id or None,
+            'connectionId': self.connection_id or None,
             'timestamp': self.timestamp or None,
             'extras': self.extras,
             **encode_data(self.data, self._encoding_array, binary)
@@ -160,7 +181,9 @@ def from_encoded(obj, cipher=None, context=None):
         count = obj.get('count')
         data = obj.get('data')
         encoding = obj.get('encoding', '')
+        id = obj.get('id')
         client_id = obj.get('clientId')
+        connection_id = obj.get('connectionId')
         timestamp = obj.get('timestamp')
         extras = obj.get('extras', None)
 
@@ -184,7 +207,9 @@ def from_encoded(obj, cipher=None, context=None):
             type=type_val,
             name=name,
             count=count,
+            id=id,
             client_id=client_id,
+            connection_id=connection_id,
             timestamp=timestamp,
             extras=extras,
             **decoded_data
@@ -200,6 +225,31 @@ def from_values(values):
         """Create an Annotation from a dict of values"""
         return Annotation(**values)
 
+    @staticmethod
+    def __update_empty_fields(proto_msg: dict, annotation: dict, annotation_index: int):
+        """Update empty annotation fields with values from protocol message"""
+        if annotation.get("id") is None or annotation.get("id") == '':
+            annotation['id'] = f"{proto_msg.get('id')}:{annotation_index}"
+        if annotation.get("connectionId") is None or annotation.get("connectionId") == '':
+            annotation['connectionId'] = proto_msg.get('connectionId')
+        if annotation.get("timestamp") is None or annotation.get("timestamp") == 0:
+            annotation['timestamp'] = proto_msg.get('timestamp')
+
+    @staticmethod
+    def update_inner_annotation_fields(proto_msg: dict):
+        """
+        Update inner annotation fields with protocol message data (RTAN4b).
+
+        Populates empty id, connectionId, and timestamp fields in annotations
+        from the protocol message values.
+        """
+        annotations: list[dict] = proto_msg.get('annotations')
+        if annotations is not None:
+            annotation_index = 0
+            for annotation in annotations:
+                Annotation.__update_empty_fields(proto_msg, annotation, annotation_index)
+                annotation_index = annotation_index + 1
+
     def __str__(self):
         return (
             f"Annotation(action={self.action}, messageSerial={self.message_serial}, "
@@ -218,7 +268,9 @@ def _copy_with(self,
                   count=_UNSET,
                   data=_UNSET,
                   encoding=_UNSET,
+                  id=_UNSET,
                   client_id=_UNSET,
+                  connection_id=_UNSET,
                   timestamp=_UNSET,
                   extras=_UNSET):
         """
@@ -236,7 +288,9 @@ def _copy_with(self,
             count: Override the count (or None to clear it)
             data: Override the data payload (or None to clear it)
             encoding: Override the encoding format (or None to clear it)
+            id: Override the ID (or None to clear it)
             client_id: Override the client ID (or None to clear it)
+            connection_id: Override the connection ID (or None to clear it)
             timestamp: Override the timestamp (or None to clear it)
             extras: Override the extras metadata (or None to clear it)
 
@@ -260,7 +314,9 @@ def _copy_with(self,
             count=self.__count if count is _UNSET else count,
             data=self.__data if data is _UNSET else data,
             encoding=self.__encoding if encoding is _UNSET else encoding,
+            id=self.__id if id is _UNSET else id,
             client_id=self.__client_id if client_id is _UNSET else client_id,
+            connection_id=self.__connection_id if connection_id is _UNSET else connection_id,
             timestamp=self.__timestamp if timestamp is _UNSET else timestamp,
             extras=self.__extras if extras is _UNSET else extras,
         )

ably/util/encoding.py

@@ -11,8 +11,7 @@ def encode_data(data: Any, encoding_array: list, binary: bool = False):
 
     if isinstance(data, (dict, list)):
         encoding.append('json')
-        data = json.dumps(data)
-        data = str(data)
+        data = json.dumps(data)  # json.dumps already returns str
     elif isinstance(data, str) and not binary:
         pass
     elif not binary and isinstance(data, (bytearray, bytes)):