Keep the realtime stream alive when one event handler crashes

The loop treated any exception from _handle_realtime_event as a broken
stream: tear down, back off 1s→30s, reconnect — and everything arriving
inside a gap was lost. That is what amplified both presence crashes (a
missing user_id field, then an int status without .name) into real message
loss, and it will amplify the next handler bug the same way.

Handle the event inside its own try: a failing handler logs its kind and
event ID, skips that event, and leaves its cursor unadvanced — an event the
adapter never applied is not safe to resume past, and chattolib's own
contract (cursor advances only after the event is applied) already says so.
Unexpected exceptions keep a full traceback. The stream itself stays up: a
cosmetic feature must never price an inbound message.

The reconnect path now only serves exceptions from stream_events itself.

adapter.py

@@ -2800,13 +2800,41 @@ class ChattoAdapter(BasePlatformAdapter):
                     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
+                        try:
+                            await self._handle_realtime_event(event)
+                        except (AttributeError, TypeError) as exc:
+                            # A handler reading a payload field that does not
+                            # exist or is not the type it assumed is a handler
+                            # bug, not wire damage: it failed one event, so it
+                            # must not cost the stream. Reconnecting here is
+                            # what turned the presence crashes into 30s
+                            # backoff gaps that silently dropped every
+                            # message_posted arriving inside them (see
+                            # AGENTS.md, "A handler bug is not a broken
+                            # stream"). The cursor is NOT advanced: an event
+                            # the adapter never applied is not safe to resume
+                            # past.
+                            logger.warning(
+                                "Chatto: handler for '%s' failed (%s); skipping event %s",
+                                event.kind,
+                                exc,
+                                event.id,
+                            )
+                        except Exception:
+                            # Not a wire-shape bug — loud traceback, same skip
+                            # semantics: one event failed, the stream stays up.
+                            logger.exception(
+                                "Chatto: handler for '%s' crashed on event %s; skipping it",
+                                event.kind,
+                                event.id,
+                            )
+                        else:
+                            # 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

@@ -1473,6 +1473,92 @@ class TestRealtimeResume:
         assert resume_cursors == [None, "cur-1"]
 
 
+class TestHandlerFailureIsNotAStreamBreak:
+    """An exception inside one event handler costs that event, not the stream.
+
+    This is the amplification that turned both presence crashes into 30s
+    reconnect gaps swallowing inbound messages: the loop's except block cannot
+    tell a handler bug from wire damage, and treated both as a break."""
+
+    def _crashing_event(self, kind="presence_changed", event_id="evt-crash"):
+        payload = PresenceChangedEvent()
+        payload.status = 99  # not a wire value; exercises the unknown-value path
+        return RealtimeEvent(
+            id=event_id,
+            created_at=None,
+            actor_id="human-1",
+            cursor="cur-crash",
+            kind=kind,
+            payload=payload,
+            raw=None,
+        )
+
+    async def _run_loop_with_crash(self, adapter, crashing_event, followup):
+        """Drive the loop over one crashing event plus a follow-up, then stop."""
+        adapter._closing = False
+        adapter._require_chatto_client = AsyncMock(return_value=adapter._chatto_client)
+        adapter._refresh_rooms = AsyncMock()
+        calls: list[RealtimeEvent] = []
+
+        async def crashing_handler(event):
+            calls.append(event)
+            if event.id == crashing_event.id:
+                raise ValueError("handler bug on exactly one event")
+
+        adapter._handle_realtime_event = crashing_handler
+
+        async def fake_stream(client, *, resume_cursor=None):
+            for event in (crashing_event, followup):
+                yield event
+
+        async def stop_after_pass(seconds):
+            adapter._closing = True
+
+        with (
+            patch("adapter.stream_events", fake_stream),
+            patch.object(adapter, "_sleep_interruptible", stop_after_pass),
+        ):
+            await adapter._chattolib_event_loop()
+
+        return calls
+
+    async def test_a_failing_handler_does_not_drop_the_followup_event(self):
+        """The crashing event was attempted, the follow-up was still delivered:
+        the stream stayed up instead of paying a reconnect backoff."""
+        adapter = _make_adapter()
+        followup = self._crashing_event(kind="message_posted", event_id="evt-followup")
+        followup.payload = MagicMock()
+
+        calls = await self._run_loop_with_crash(
+            adapter, self._crashing_event(), followup
+        )
+
+        assert [event.id for event in calls] == ["evt-crash", "evt-followup"]
+
+    async def test_a_failing_handler_keeps_its_cursor_unadvanced(self):
+        """The crashed event was never applied, so its cursor must not be
+        recorded as safe — the next connection replays it."""
+        adapter = _make_adapter()
+        crashing = self._crashing_event()
+
+        await self._run_loop_with_crash(adapter, crashing, MagicMock())
+
+        assert adapter._realtime_cursor != "cur-crash"
+
+    async def test_a_failing_handler_is_not_answered_with_a_reconnect(self):
+        """One failing handler costs one warning, not a new stream connection:
+        the loop must not tear the socket down over a handler bug."""
+        adapter = _make_adapter()
+        crashing = self._crashing_event()
+        after = self._crashing_event(kind="message_posted", event_id="evt-after")
+        after.payload = MagicMock()
+
+        calls = await self._run_loop_with_crash(adapter, crashing, after)
+
+        # Exactly one stream pass for two events: no reconnect between them.
+        assert [event.id for event in calls] == ["evt-crash", "evt-after"]
+
+
 class TestPersistedRealtimeCursor:
     """A clean shutdown stores the stream position; a restart inside the window
     resumes from it instead of starting at the live boundary."""