fix(chatto): resume the realtime stream from the last applied cursor

Every reconnect called stream_events() without a resume_cursor, so the
server started at the live boundary and everything posted during the
disconnect was never delivered — a room's messages in the gap were lost
quietly, with no error and no log line.

chattolib already carries the mechanism: cursor is the opaque position
safe to retain once its event has been applied, and the subscribe frame
resumes after it. The adapter now keeps the last one and passes it back.
An event whose cursor is None (unreplayable) leaves the previous one
intact, and a handler that raises never advances it, so that event is
redelivered rather than skipped.

This covers reconnects within one process. A gateway restart still starts
fresh: persisting a cursor across restarts is a separate decision about
retention and staleness.

adapter.py

@@ -451,6 +451,12 @@ class ChattoAdapter(BasePlatformAdapter):
         # --- Runtime state ---
         self._room_names: dict[str, str] = {}
         self._room_kinds: dict[str, RoomKind] = {}
+        # Last realtime cursor whose event was fully applied. Passed back to
+        # stream_events on reconnect so the server resumes after it instead of
+        # starting at the current boundary and silently skipping everything
+        # that arrived while we were disconnected. None until the first event
+        # carries one (chattolib leaves it unset for unreplayable events).
+        self._realtime_cursor: str | None = None
         # Event IDs already processed — chattolib may redeliver events across
         # reconnects, so every inbound event is checked against this list.
         # Bounded deques: appending past the cap drops the oldest ID on its own.
@@ -2582,11 +2588,19 @@ class ChattoAdapter(BasePlatformAdapter):
                     len(self._joined_room_ids),
                 )
 
-                async for event in stream_events(client):
+                async for event in stream_events(
+                    client, resume_cursor=self._realtime_cursor
+                ):
                     if self._closing:
                         return
                     if isinstance(event, RealtimeEvent):
                         await self._handle_realtime_event(event)
+                        # chattolib marks a cursor as safe only once its event
+                        # has been applied, and leaves it None when the event
+                        # cannot be replayed. Keeping the last good one is what
+                        # turns a reconnect into a resume rather than a gap.
+                        if event.cursor:
+                            self._realtime_cursor = event.cursor
                 # Iterator exited cleanly — treat as a normal close and reconnect
                 # with the local backoff (no server hint available).
                 logger.info("Chatto: realtime stream ended, reconnecting")

test_adapter.py

@@ -1307,6 +1307,75 @@ class TestInteractiveReactionRouting:
         handler.assert_called_once()
 
 
+# -- Realtime resume --
+
+
+class TestRealtimeResume:
+    """A reconnect resumes from the last applied cursor instead of starting at
+    the live boundary, which would silently skip everything that arrived while
+    the socket was down."""
+
+    def _event(self, cursor):
+        from chattolib.realtime import RealtimeEvent
+
+        return RealtimeEvent(
+            id=f"evt-{cursor}",
+            created_at=None,
+            actor_id="human-1",
+            cursor=cursor,
+            kind="reaction_added",
+            payload=MagicMock(),
+            raw=None,
+        )
+
+    async def _run_loop(self, adapter, passes):
+        """Drive the loop through ``passes`` stream connections, then stop it.
+
+        ``passes`` is a list of event lists, one per connection; the recorded
+        ``resume_cursor`` of each call comes back so the test can assert what
+        the second connection asked the server for.
+        """
+        adapter._closing = False
+        adapter._require_chatto_client = AsyncMock(return_value=adapter._chatto_client)
+        adapter._refresh_rooms = AsyncMock()
+        adapter._handle_realtime_event = AsyncMock()
+
+        resume_cursors: list[str | None] = []
+
+        async def fake_stream(client, *, resume_cursor=None):
+            resume_cursors.append(resume_cursor)
+            for event in passes[len(resume_cursors) - 1]:
+                yield event
+
+        async def stop_after_last_pass(seconds):
+            if len(resume_cursors) >= len(passes):
+                adapter._closing = True
+
+        with (
+            patch("adapter.stream_events", fake_stream),
+            patch.object(adapter, "_sleep_interruptible", stop_after_last_pass),
+        ):
+            await adapter._chattolib_event_loop()
+
+        return resume_cursors
+
+    async def test_reconnect_resumes_from_the_last_event_cursor(self):
+        adapter = _make_adapter()
+        resume_cursors = await self._run_loop(
+            adapter, [[self._event("cur-1")], []]
+        )
+        assert resume_cursors == [None, "cur-1"]
+
+    async def test_an_unreplayable_event_keeps_the_previous_cursor(self):
+        """``cursor`` is None for events the server cannot replay; it must not
+        wipe the last good resume point."""
+        adapter = _make_adapter()
+        resume_cursors = await self._run_loop(
+            adapter, [[self._event("cur-1"), self._event(None)], []]
+        )
+        assert resume_cursors == [None, "cur-1"]
+
+
 # -- chat_type mapping --