Properly return prev_batch tokens for threads extension

synapse/handlers/sliding_sync/extensions.py

@@ -996,8 +996,8 @@ class SlidingSyncExtensionHandler:
         """Handle Threads extension (MSC4360)
 
         Args:
-            sync_config: Sync configuration
-            threads_request: The threads extension from the request
+            sync_config: Sync configuration.
+            threads_request: The threads extension from the request.
             to_token: The point in the stream to sync up to.
             from_token: The point in the stream to sync from.
 
@@ -1007,24 +1007,30 @@ class SlidingSyncExtensionHandler:
         if not threads_request.enabled:
             return None
 
-        limit = threads_request.limit
-
-        # TODO: is the `room_key` the right thing to use here?
-        # ie. does it translate into /relations
-
-        updates, prev_batch = await self.store.get_thread_updates_for_user(
+        # Fetch thread updates globally across all joined rooms.
+        # The database layer returns a StreamToken (exclusive) for prev_batch if there
+        # are more results.
+        (
+            all_thread_updates,
+            prev_batch_token,
+        ) = await self.store.get_thread_updates_for_user(
             user_id=sync_config.user.to_string(),
-            from_token=from_token.stream_token if from_token else None,
-            to_token=to_token,
-            limit=limit,
+            from_token=from_token.stream_token.room_key if from_token else None,
+            to_token=to_token.room_key,
+            limit=threads_request.limit,
             include_thread_roots=threads_request.include_roots,
         )
 
-        if len(updates) == 0:
+        if len(all_thread_updates) == 0:
             return None
 
-        # Collect thread root events and get bundled aggregations
-        thread_root_events = [event for _, _, event in updates if event]
+        # Collect thread root events and get bundled aggregations.
+        # Only fetch bundled aggregations if we have thread root events to attach them to.
+        thread_root_events = [
+            update.thread_root_event
+            for update in all_thread_updates
+            if update.thread_root_event
+        ]
         aggregations_map = {}
         if thread_root_events:
             aggregations_map = await self.relations_handler.get_bundled_aggregations(
@@ -1033,17 +1039,23 @@ class SlidingSyncExtensionHandler:
             )
 
         thread_updates: Dict[str, Dict[str, _ThreadUpdate]] = {}
-        for thread_root_id, room_id, thread_root_event in updates:
+        for update in all_thread_updates:
+            # Only look up bundled aggregations if we have a thread root event
             bundled_aggs = (
-                aggregations_map.get(thread_root_id) if thread_root_event else None
+                aggregations_map.get(update.thread_id)
+                if update.thread_root_event
+                else None
             )
-            thread_updates.setdefault(room_id, {})[thread_root_id] = _ThreadUpdate(
-                thread_root=thread_root_event,
-                prev_batch=None,
-                bundled_aggregations=bundled_aggs,
+
+            thread_updates.setdefault(update.room_id, {})[update.thread_id] = (
+                _ThreadUpdate(
+                    thread_root=update.thread_root_event,
+                    prev_batch=update.prev_batch,
+                    bundled_aggregations=bundled_aggs,
+                )
             )
 
         return SlidingSyncResult.Extensions.ThreadsExtension(
             updates=thread_updates,
-            prev_batch=prev_batch,
+            prev_batch=prev_batch_token,
         )

synapse/rest/client/sync.py

@@ -57,6 +57,7 @@ from synapse.http.servlet import (
 from synapse.http.site import SynapseRequest
 from synapse.logging.opentracing import log_kv, set_tag, trace_with_opname
 from synapse.rest.admin.experimental_features import ExperimentalFeature
+from synapse.storage.databases.main import DataStore
 from synapse.types import JsonDict, Requester, SlidingSyncStreamToken, StreamToken
 from synapse.types.rest.client import SlidingSyncBody
 from synapse.util.caches.lrucache import LruCache
@@ -1107,6 +1108,7 @@ class SlidingSyncRestServlet(RestServlet):
                 self.event_serializer,
                 time_now,
                 extensions.threads,
+                self.store,
             )
 
         return serialized_extensions
@@ -1149,6 +1151,7 @@ async def _serialise_threads(
     event_serializer: EventClientSerializer,
     time_now: int,
     threads: SlidingSyncResult.Extensions.ThreadsExtension,
+    store: "DataStore",
 ) -> JsonDict:
     """
     Serialize the threads extension response for sliding sync.
@@ -1157,6 +1160,7 @@ async def _serialise_threads(
         event_serializer: The event serializer to use for serializing thread root events.
         time_now: The current time in milliseconds, used for event serialization.
         threads: The threads extension data containing thread updates and pagination tokens.
+        store: The datastore, needed for serializing stream tokens.
 
     Returns:
         A JSON-serializable dict containing:
@@ -1195,7 +1199,7 @@ async def _serialise_threads(
 
                 # Add prev_batch if present
                 if update.prev_batch is not None:
-                    update_dict["prev_batch"] = str(update.prev_batch)
+                    update_dict["prev_batch"] = await update.prev_batch.to_string(store)
 
                 room_updates[thread_root_id] = update_dict
 
@@ -1204,7 +1208,7 @@ async def _serialise_threads(
         out["updates"] = updates_dict
 
     if threads.prev_batch:
-        out["prev_batch"] = str(threads.prev_batch)
+        out["prev_batch"] = await threads.prev_batch.to_string(store)
 
     return out
 

synapse/storage/databases/main/relations.py

@@ -54,7 +54,12 @@ from synapse.storage.databases.main.stream import (
     generate_pagination_where_clause,
 )
 from synapse.storage.engines import PostgresEngine
-from synapse.types import JsonDict, StreamKeyType, StreamToken
+from synapse.types import (
+    JsonDict,
+    RoomStreamToken,
+    StreamKeyType,
+    StreamToken,
+)
 from synapse.util.caches.descriptors import cached, cachedList
 
 if TYPE_CHECKING:
@@ -96,6 +101,28 @@ class _RelatedEvent:
     sender: str
 
 
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class ThreadUpdateInfo:
+    """
+    Information about a thread update for the sliding sync threads extension.
+
+    Attributes:
+        thread_id: The event ID of the thread root event (the event that started the thread).
+        room_id: The room ID where this thread exists.
+        thread_root_event: The actual EventBase object for the thread root event,
+            if include_thread_roots was True in the request. Otherwise None.
+        prev_batch: A pagination token (exclusive) for fetching older events in this thread.
+            Only present if update_count > 1. This token can be used with the /relations
+            endpoint with dir=b to paginate backwards through the thread's history without
+            re-receiving the latest event that was already included in the sliding sync response.
+    """
+
+    thread_id: str
+    room_id: str
+    thread_root_event: Optional[EventBase]
+    prev_batch: Optional[StreamToken]
+
+
 class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore):
     def __init__(
         self,
@@ -1127,31 +1154,35 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore):
         self,
         *,
         user_id: str,
-        from_token: Optional[StreamToken] = None,
-        to_token: Optional[StreamToken] = None,
+        from_token: Optional[RoomStreamToken] = None,
+        to_token: Optional[RoomStreamToken] = None,
         limit: int = 5,
         include_thread_roots: bool = False,
-    ) -> Tuple[Sequence[Tuple[str, str, Optional[EventBase]]], Optional[int]]:
+    ) -> Tuple[Sequence[ThreadUpdateInfo], Optional[StreamToken]]:
         """Get a list of updated threads, ordered by stream ordering of their
         latest reply, filtered to only include threads in rooms where the user
         is currently joined.
 
         Args:
-            user_id: Only fetch threads for rooms that the user is currently joined to.
-            from_token: Fetch rows from a previous next_batch, or from the start if None.
-            to_token: Fetch rows from a previous prev_batch, or from the stream end if None.
-            limit: Only fetch the most recent `limit` threads.
+            user_id: The user ID to fetch thread updates for. Only threads in rooms
+                where this user is currently joined will be returned.
+            from_token: The lower bound (exclusive) for thread updates. If None,
+                fetch from the start of the room timeline.
+            to_token: The upper bound (inclusive) for thread updates. If None,
+                fetch up to the current position in the room timeline.
+            limit: Maximum number of thread updates to return.
             include_thread_roots: If True, fetch and return the thread root EventBase
-                objects. If False, return None for the event.
+                objects. If False, return None for the thread_root_event field.
 
         Returns:
             A tuple of:
-                A list of (thread_id, room_id, thread_root_event) tuples.
-                    thread_root_event will be None if include_thread_roots=False.
-                The next_batch, if one exists.
+                A list of ThreadUpdateInfo objects containing thread update information,
+                ordered by stream_ordering descending (most recent first).
+                A prev_batch StreamToken (exclusive) if there are more results available,
+                None otherwise.
         """
         # Ensure bad limits aren't being passed in.
-        assert limit >= 0
+        assert limit > 0
 
         # Generate the pagination clause, if necessary.
         #
@@ -1159,18 +1190,35 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore):
         pagination_clause = ""
         pagination_args: List[str] = []
         if from_token:
-            from_bound = from_token.room_key.stream
+            from_bound = from_token.stream
             pagination_clause += " AND stream_ordering > ?"
             pagination_args.append(str(from_bound))
 
         if to_token:
-            to_bound = to_token.room_key.stream
+            to_bound = to_token.stream
             pagination_clause += " AND stream_ordering <= ?"
             pagination_args.append(str(to_bound))
 
-        # Filter threads to only those in rooms that the user is currently joined to.
+        # Build the update count clause - count events in the thread within the sync window
+        update_count_clause = ""
+        update_count_args: List[str] = []
+        update_count_clause = f"""
+            (SELECT COUNT(*)
+             FROM event_relations AS er
+             INNER JOIN events AS e ON er.event_id = e.event_id
+             WHERE er.relates_to_id = threads.thread_id
+               AND er.relation_type = '{RelationTypes.THREAD}'"""
+        if from_token:
+            update_count_clause += " AND e.stream_ordering > ?"
+            update_count_args.append(str(from_token.stream))
+        if to_token:
+            update_count_clause += " AND e.stream_ordering <= ?"
+            update_count_args.append(str(to_token.stream))
+        update_count_clause += ")"
+
+        # Filter threads to only those in rooms where the user is currently joined.
         sql = f"""
-            SELECT thread_id, room_id, latest_event_id, stream_ordering
+            SELECT thread_id, room_id, stream_ordering, {update_count_clause} AS update_count
               FROM threads
               WHERE EXISTS (
                   SELECT 1
@@ -1186,44 +1234,86 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore):
 
         def _get_thread_updates_for_user_txn(
             txn: LoggingTransaction,
-        ) -> Tuple[List[Tuple[str, str]], Optional[int]]:
-            txn.execute(sql, (user_id, Membership.JOIN, *pagination_args, limit + 1))
+        ) -> Tuple[List[Tuple[str, str, int, int]], Optional[int]]:
+            # Add 1 to the limit as a free way of determining if there are more results
+            # than the limit amount. If `limit + 1` results are returned, then there are
+            # more results. Otherwise we would need to do a separate query to determine
+            # if this was true when exactly `limit` results are returned.
+            txn.execute(
+                sql,
+                (
+                    *update_count_args,
+                    user_id,
+                    Membership.JOIN,
+                    *pagination_args,
+                    limit + 1,
+                ),
+            )
 
-            rows = cast(List[Tuple[str, str, str, int]], txn.fetchall())
-            thread_ids = [(r[0], r[1]) for r in rows]
+            # SQL returns: thread_id, room_id, stream_ordering, update_count
+            rows = cast(List[Tuple[str, str, int, int]], txn.fetchall())
 
             # If there are more events, generate the next pagination key from the
             # last thread which will be returned.
             next_token = None
-            if len(thread_ids) > limit:
-                # TODO: why -2?
-                next_token = rows[-2][3]
+            if len(rows) > limit:
+                # Set the next_token to be the second last row in the result set since
+                # that will be the last row we return from this function.
+                # This works as an exclusive bound that can be backpaginated from.
+                # Use the stream_ordering field (index 2 in original rows)
+                next_token = rows[-2][2]
 
-            return thread_ids[:limit], next_token
+            return rows[:limit], next_token
 
-        thread_ids, next_token = await self.db_pool.runInteraction(
+        thread_infos, next_token_int = await self.db_pool.runInteraction(
             "get_thread_updates_for_user", _get_thread_updates_for_user_txn
         )
 
+        # Convert the next_token int (stream ordering) to a StreamToken.
+        # Use StreamToken.START as base (all other streams at 0) since only room
+        # position matters.
+        # Subtract 1 to make it exclusive - the client can paginate from this point without
+        # receiving the last thread update that was already returned.
+        next_token = None
+        if next_token_int is not None:
+            next_token = StreamToken.START.copy_and_replace(
+                StreamKeyType.ROOM, RoomStreamToken(stream=next_token_int - 1)
+            )
+
         # Optionally fetch thread root events
-        if include_thread_roots and thread_ids:
-            thread_root_ids = [thread_id for thread_id, _ in thread_ids]
+        event_map = {}
+        if include_thread_roots and thread_infos:
+            thread_root_ids = [thread_id for thread_id, _, _, _ in thread_infos]
             thread_root_events = await self.get_events_as_list(thread_root_ids)
             event_map = {e.event_id: e for e in thread_root_events}
 
-            return (
-                [
-                    (thread_id, room_id, event_map.get(thread_id))
-                    for thread_id, room_id in thread_ids
-                ],
-                next_token,
-            )
-        else:
-            return (
-                [(thread_id, room_id, None) for thread_id, room_id in thread_ids],
-                next_token,
+        # Build ThreadUpdateInfo objects with per-thread prev_batch tokens.
+        thread_update_infos = []
+        for thread_id, room_id, stream_ordering, update_count in thread_infos:
+            # Generate prev_batch token if this thread has more than one update.
+            per_thread_prev_batch = None
+            if update_count > 1:
+                # Create a token pointing to one position before the latest event's
+                # stream position.
+                # This makes it exclusive - /relations with dir=b won't return the
+                # latest event again.
+                # Use StreamToken.START as base (all other streams at 0) since only room
+                # position matters.
+                per_thread_prev_batch = StreamToken.START.copy_and_replace(
+                    StreamKeyType.ROOM, RoomStreamToken(stream=stream_ordering - 1)
+                )
+
+            thread_update_infos.append(
+                ThreadUpdateInfo(
+                    thread_id=thread_id,
+                    room_id=room_id,
+                    thread_root_event=event_map.get(thread_id),
+                    prev_batch=per_thread_prev_batch,
+                )
             )
 
+        return (thread_update_infos, next_token)
+
 
 class RelationsStore(RelationsWorkerStore):
     pass

synapse/types/handlers/sliding_sync.py

@@ -401,28 +401,47 @@ class SlidingSyncResult:
 
         @attr.s(slots=True, frozen=True, auto_attribs=True)
         class ThreadsExtension:
-            # TODO: comment
             """The Threads extension (MSC4360)
 
+            Provides thread updates for threads that have new activity across all of the
+            user's joined rooms within the sync window.
+
             Attributes:
+                updates: A nested mapping of room_id -> thread_root_id -> ThreadUpdate.
+                    Each ThreadUpdate contains information about a thread that has new activity,
+                    including the thread root event (if requested) and a pagination token
+                    for fetching older events in that specific thread.
+                prev_batch: A pagination token for fetching more thread updates across all rooms.
+                    If present, indicates there are more thread updates available beyond what
+                    was returned in this response. This token can be used with a future request
+                    to paginate through older thread updates.
             """
 
             @attr.s(slots=True, frozen=True, auto_attribs=True)
             class ThreadUpdate:
-                # The thread root event, if requested via include_roots
+                """Information about a single thread that has new activity.
+
+                Attributes:
+                    thread_root: The thread root event, if requested via include_roots in the
+                        request. This is the event that started the thread.
+                    prev_batch: A pagination token (exclusive) for fetching older events in this
+                        specific thread. Only present if the thread has multiple updates in the
+                        sync window. This token can be used with the /relations endpoint with
+                        dir=b to paginate backwards through the thread's history.
+                    bundled_aggregations: Bundled aggregations for the thread root event,
+                        including the latest_event in the thread (found in
+                        unsigned.m.relations.m.thread). Only present if thread_root is included.
+                """
+
                 thread_root: Optional[EventBase]
-
-                # Pagination token for this thread
                 prev_batch: Optional[StreamToken]
-
-                # Bundled aggregations for the thread root event (includes latest_event)
                 bundled_aggregations: Optional["BundledAggregations"] = None
 
                 def __bool__(self) -> bool:
                     return bool(self.thread_root) or bool(self.prev_batch)
 
             updates: Optional[Mapping[str, Mapping[str, ThreadUpdate]]]
-            prev_batch: Optional[int]
+            prev_batch: Optional[StreamToken]
 
             def __bool__(self) -> bool:
                 return bool(self.updates) or bool(self.prev_batch)
@@ -893,6 +912,7 @@ class PerConnectionState:
     Attributes:
         rooms: The status of each room for the events stream.
         receipts: The status of each room for the receipts stream.
+        account_data: The status of each room for the account data stream.
         room_configs: Map from room_id to the `RoomSyncConfig` of all
             rooms that we have previously sent down.
     """

synapse/types/rest/client/__init__.py

@@ -380,9 +380,9 @@ class SlidingSyncBody(RequestBodyModel):
             """The Threads extension (MSC4360)
 
             Attributes:
-                enabled
+                enabled: Whether the threads extension is enabled.
                 include_roots: whether to include thread root events in the extension response.
-                limit: maximum number of thread updates to return.
+                limit: maximum number of thread updates to return across all joined rooms.
             """
 
             enabled: Optional[StrictBool] = False

tests/rest/client/sliding_sync/test_extension_threads.py

@@ -17,7 +17,7 @@ from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
 from synapse.api.constants import RelationTypes
-from synapse.rest.client import login, room, sync
+from synapse.rest.client import login, relations, room, sync
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util.clock import Clock
@@ -43,8 +43,7 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
         login.register_servlets,
         room.register_servlets,
         sync.register_servlets,
-        # TODO:
-        # threads.register_servlets,
+        relations.register_servlets,
     ]
 
     def default_config(self) -> JsonDict:
@@ -64,7 +63,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
         user1_id = self.register_user("user1", "pass")
         user1_tok = self.login(user1_id, "pass")
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -84,16 +82,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
         """
         user1_id = self.register_user("user1", "pass")
         user1_tok = self.login(user1_id, "pass")
-        initial_sync_body: JsonDict = {
-            "lists": {},
-        }
+        initial_sync_body: JsonDict = {}
 
         # Initial sync
         response_body, sync_pos = self.do_sync(initial_sync_body, tok=user1_tok)
 
         # Incremental sync with extension enabled
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -139,7 +134,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
         # base = self.store.get_max_thread_subscriptions_stream_id()
 
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -164,7 +158,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
         user1_tok = self.login(user1_id, "pass")
         room_id = self.helper.create_room_as(user1_id, tok=user1_tok)
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -263,7 +256,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
 
         # User2 syncs with threads extension enabled
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -304,7 +296,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
 
         # Initial sync for user2
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -389,7 +380,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
 
         # Sync with include_roots=True
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -444,7 +434,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
 
         # Sync with include_roots=False (explicitly)
         sync_body = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -462,7 +451,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
 
         # Also test with include_roots omitted (should behave the same)
         sync_body_no_param = {
-            "lists": {},
             "extensions": {
                 EXT_NAME: {
                     "enabled": True,
@@ -475,3 +463,261 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
             "updates"
         ][room_id][thread_root_id]
         self.assertNotIn("thread_root", thread_update_no_param)
+
+    def test_per_thread_prev_batch_single_update(self) -> None:
+        """
+        Test that threads with only a single update do NOT get a prev_batch token.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        # Create thread root
+        thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok)
+        thread_root_id = thread_root_resp["event_id"]
+
+        # Initial sync to establish baseline
+        sync_body = {
+            "extensions": {
+                EXT_NAME: {
+                    "enabled": True,
+                }
+            },
+        }
+        _, sync_pos = self.do_sync(sync_body, tok=user1_tok)
+
+        # Add ONE reply to thread
+        self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "Single reply",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+
+        # Incremental sync
+        response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos)
+
+        # Assert: Thread update should NOT have prev_batch (only 1 update)
+        thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][
+            thread_root_id
+        ]
+        self.assertNotIn(
+            "prev_batch",
+            thread_update,
+            "Threads with single update should not have prev_batch",
+        )
+
+    def test_per_thread_prev_batch_multiple_updates(self) -> None:
+        """
+        Test that threads with multiple updates get a prev_batch token that can be
+        used with /relations endpoint to paginate backwards.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        # Create thread root
+        thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok)
+        thread_root_id = thread_root_resp["event_id"]
+
+        # Initial sync to establish baseline
+        sync_body = {
+            "extensions": {
+                EXT_NAME: {
+                    "enabled": True,
+                }
+            },
+        }
+        _, sync_pos = self.do_sync(sync_body, tok=user1_tok)
+
+        # Add MULTIPLE replies to thread
+        reply1_resp = self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "First reply",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+        reply1_id = reply1_resp["event_id"]
+
+        reply2_resp = self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "Second reply",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+        reply2_id = reply2_resp["event_id"]
+
+        reply3_resp = self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "Third reply",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+        reply3_id = reply3_resp["event_id"]
+
+        # Incremental sync
+        response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos)
+
+        # Assert: Thread update SHOULD have prev_batch (3 updates)
+        thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][
+            thread_root_id
+        ]
+        self.assertIn(
+            "prev_batch",
+            thread_update,
+            "Threads with multiple updates should have prev_batch",
+        )
+
+        prev_batch = thread_update["prev_batch"]
+        self.assertIsNotNone(prev_batch, "prev_batch should not be None")
+
+        # Now use the prev_batch token with /relations endpoint to paginate backwards
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/v1/rooms/{room_id}/relations/{thread_root_id}?from={prev_batch}&dir=b",
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        relations_response = channel.json_body
+        returned_event_ids = [
+            event["event_id"] for event in relations_response["chunk"]
+        ]
+
+        # Assert: Only the older replies should be returned (not the latest one we already saw)
+        # The prev_batch token should be exclusive, pointing just before the latest event
+        self.assertIn(
+            reply1_id,
+            returned_event_ids,
+            "First reply should be in relations response",
+        )
+        self.assertIn(
+            reply2_id,
+            returned_event_ids,
+            "Second reply should be in relations response",
+        )
+        self.assertNotIn(
+            reply3_id,
+            returned_event_ids,
+            "Third reply (latest) should NOT be in relations response - already returned in sliding sync",
+        )
+
+    def test_per_thread_prev_batch_on_initial_sync(self) -> None:
+        """
+        Test that threads with multiple updates get prev_batch tokens on initial sync
+        so clients can paginate through the full thread history.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        # Create thread with multiple replies BEFORE any sync
+        thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok)
+        thread_root_id = thread_root_resp["event_id"]
+
+        reply1_resp = self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "Reply 1",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+        reply1_id = reply1_resp["event_id"]
+
+        reply2_resp = self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "Reply 2",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+        reply2_id = reply2_resp["event_id"]
+
+        # Initial sync (no from_token)
+        sync_body = {
+            "extensions": {
+                EXT_NAME: {
+                    "enabled": True,
+                }
+            },
+        }
+        response_body, _ = self.do_sync(sync_body, tok=user1_tok)
+
+        # Assert: Thread update SHOULD have prev_batch on initial sync (2+ updates exist)
+        thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][
+            thread_root_id
+        ]
+        self.assertIn(
+            "prev_batch",
+            thread_update,
+            "Threads with multiple updates should have prev_batch even on initial sync",
+        )
+
+        prev_batch = thread_update["prev_batch"]
+        self.assertIsNotNone(prev_batch)
+
+        # Use prev_batch with /relations to fetch the thread history
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/v1/rooms/{room_id}/relations/{thread_root_id}?from={prev_batch}&dir=b",
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        relations_response = channel.json_body
+        returned_event_ids = [
+            event["event_id"] for event in relations_response["chunk"]
+        ]
+
+        # Assert: Only the older reply should be returned (not the latest one we already saw)
+        # The prev_batch token should be exclusive, pointing just before the latest event
+        self.assertIn(
+            reply1_id,
+            returned_event_ids,
+            "First reply should be in relations response",
+        )
+        self.assertNotIn(
+            reply2_id,
+            returned_event_ids,
+            "Second reply (latest) should NOT be in relations response - already returned in sliding sync",
+        )