feat(chatto): resume the realtime stream across a clean restart

A gateway restart forgot its stream position, so everything posted while
it was down was never delivered. The position is now stored by
disconnect() and restored by connect(), which closes the gap deploys and
manual restarts leave behind.

Three bounds keep it from becoming an unbounded replay:

- Only a clean stop writes it. A crash stores nothing, and the file is
  written in one place on purpose — per-event writes would buy crash
  coverage at the cost of I/O on every presence change, which is a
  decision to make explicitly, not a side effect.
- CHATTO_RESUME_WINDOW (default 900 s, 0 disables) decides how much of an
  outage may be replayed; an older position is discarded, which is what
  the server does anyway when LIVE_ONLY cannot resume the cursor.
- The payload carries _profile_scope_key(): under multiplexing a position
  written by one profile must never be spent on another's stream.

adapter.py

@@ -39,6 +39,7 @@ from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from difflib import SequenceMatcher
 from enum import StrEnum
+from pathlib import Path
 from typing import Any, Final, NamedTuple, TypeVar
 from urllib.parse import unquote, urlsplit
 
@@ -650,6 +651,12 @@ class ChattoAdapter(BasePlatformAdapter):
         await self._announce_online()
 
         self._closing = False
+        # A cursor persisted by a clean shutdown lets this process pick up where
+        # the previous one stopped, so a restart inside the window answers what
+        # arrived while it was down instead of starting at the live boundary.
+        self._realtime_cursor = _restore_realtime_cursor(self.chatto_config)
+        if self._realtime_cursor:
+            logger.info("Chatto: resuming the realtime stream from the stored cursor")
         # Start background realtime WS event stream loop.
         self._ws_task = asyncio.create_task(
             self._chattolib_event_loop(),
@@ -714,6 +721,11 @@ class ChattoAdapter(BasePlatformAdapter):
         # below is what actually takes the bot offline.
         self._closing = True
 
+        # Store the stream position before anything is torn down: this is the
+        # one moment where a restart can resume instead of starting at the live
+        # boundary, and it costs one small write.
+        _persist_realtime_cursor(self.chatto_config, self._realtime_cursor)
+
         # Cancel all typing tasks
         for chat_id in list(self._typing_tasks.keys()):
             await self.stop_typing(chat_id)
@@ -4125,11 +4137,87 @@ def _profile_scope_key() -> str:
         return ""
 
 
+def _realtime_cursor_file() -> Path | None:
+    """Where a restart-resumable realtime cursor lives, or None without Hermes.
+
+    Under the profile's own home so two profiles never read each other's
+    position; ``None`` (a bare process with no ``hermes_constants``) simply
+    means nothing is persisted, which is the same behaviour as no config.
+    """
+    try:
+        from hermes_constants import get_hermes_home
+    except ImportError:
+        return None
+    return Path(get_hermes_home()) / "state" / "chatto_realtime_cursor.json"
+
+
+def _restore_realtime_cursor(cfg: ChattoConfiguration) -> str | None:
+    """The stored realtime cursor, if it is this profile's and still fresh.
+
+    Three ways a stored value is refused instead of trusted: a different
+    profile wrote it, it carries a shape this build does not understand, or it
+    is older than ``resume_window``. In every case the stream starts at the
+    live boundary, which is what the plugin did before the cursor was stored at
+    all — so a bad file costs the gap it was meant to close, never a crash.
+    """
+    window = cfg.resume_window.value
+    if window <= 0:
+        return None
+    path = _realtime_cursor_file()
+    if path is None:
+        return None
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+    if payload.get("scope") != _profile_scope_key():
+        return None
+    cursor = payload.get("cursor")
+    saved_at = payload.get("saved_at")
+    if not isinstance(cursor, str) or not cursor or not isinstance(saved_at, str):
+        return None
+    try:
+        saved = datetime.fromisoformat(saved_at)
+    except ValueError:
+        return None
+    if saved.tzinfo is None:
+        saved = saved.replace(tzinfo=UTC)
+    if (datetime.now(UTC) - saved).total_seconds() > window:
+        return None
+    return cursor
+
+
+def _persist_realtime_cursor(cfg: ChattoConfiguration, cursor: str | None) -> None:
+    """Store the cursor of the last applied event, best-effort.
+
+    Written only on a clean stop, not per event: a deploy or restart should not
+    lose the gap, while a crash is allowed to — the alternative is a file write
+    on every presence change. A failed write is logged and otherwise ignored,
+    because losing the resume point is not a reason to fail a shutdown.
+    """
+    if not cursor or cfg.resume_window.value <= 0:
+        return
+    path = _realtime_cursor_file()
+    if path is None:
+        return
+    payload = {
+        "scope": _profile_scope_key(),
+        "cursor": cursor,
+        "saved_at": datetime.now(UTC).isoformat(),
+    }
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(payload), encoding="utf-8")
+    except OSError as exc:
+        logger.debug("Chatto: realtime cursor not persisted: %s", exc)
+
+
 # How similar an ``extra`` key must be to a declared config_key before the
 # unknown-key check reads it as a likely typo or renamed field.
 _NEAR_MISS_RATIO = 0.75
 
-
 def _warn_on_unknown_extra_keys(config: PlatformConfig) -> None:
     """Warn about ``extra`` keys that look like misspelled config fields.
 
@@ -4185,6 +4273,14 @@ def hermes_validate_config(config: PlatformConfig) -> bool:
         )
         return False
 
+    if resolved.resume_window.value < 0:
+        logger.error(
+            "Chatto: 'resume_window' must be 0 (disabled) or a positive number "
+            "of seconds, not %s.",
+            resolved.resume_window.value,
+        )
+        return False
+
     _by_room, malformed_toolsets = _parse_room_toolsets(
         resolved.room_toolsets.value
     )

platform_config.py

@@ -435,6 +435,13 @@ class ChattoConfiguration:
     # to name everything the room keeps. A malformed entry is rejected at
     # startup rather than guessed at.
     room_toolsets = ConfigField("list")
+    # How long a cursor persisted by a clean shutdown may still be used, in
+    # seconds. A restart within this window resumes the realtime stream and
+    # receives what arrived while the gateway was down; an older cursor is
+    # discarded so a long outage starts at the live boundary instead of
+    # replaying hours of history. 0 disables the persisted cursor (reconnects
+    # inside one process still resume from memory).
+    resume_window = ConfigField("int", default=900)
     # Auto-thread: by default, Chatto creates a thread for replies to room
     # messages (not DMs, not already in a thread). This keeps conversations
     # organized in the room. Can be disabled via extra.auto_thread=false.

plugin.yaml

@@ -76,4 +76,8 @@ optional_env:
   - name: CHATTO_EDIT_WINDOW
     description: "How long after posting an edit may still land, in seconds; older edits are ignored (default: 300)"
     prompt: "Edit window in seconds (or empty for 300)"
+    password: false
+  - name: CHATTO_RESUME_WINDOW
+    description: "How long a stream position saved by a clean shutdown may still be used after a gateway restart, in seconds. Within the window the bot receives what arrived while it was down; an older position is discarded so a long outage does not replay hours of history. 0 disables it, and a crash (no clean shutdown) leaves no position to resume from. (default: 900)"
+    prompt: "Restart resume window in seconds (or empty for 900, 0 disables)"
     password: false

test_adapter.py

@@ -1436,6 +1436,97 @@ class TestRealtimeResume:
         assert resume_cursors == [None, "cur-1"]
 
 
+class TestPersistedRealtimeCursor:
+    """A clean shutdown stores the stream position; a restart inside the window
+    resumes from it instead of starting at the live boundary."""
+
+    def _cfg(self, **extra):
+        return ChattoConfiguration(pconfig=_make_config(token="test-token", **extra))
+
+    def _store(self, monkeypatch, tmp_path, payload):
+        path = tmp_path / "chatto_realtime_cursor.json"
+        monkeypatch.setattr("adapter._realtime_cursor_file", lambda: path)
+        path.write_text(json.dumps(payload), encoding="utf-8")
+        return path
+
+    def test_a_stored_cursor_survives_a_clean_restart(self, monkeypatch, tmp_path):
+        from adapter import _persist_realtime_cursor, _restore_realtime_cursor
+
+        path = tmp_path / "chatto_realtime_cursor.json"
+        monkeypatch.setattr("adapter._realtime_cursor_file", lambda: path)
+        _persist_realtime_cursor(self._cfg(), "cur-42")
+        assert _restore_realtime_cursor(self._cfg()) == "cur-42"
+
+    def test_a_cursor_older_than_the_window_is_discarded(
+        self, monkeypatch, tmp_path
+    ):
+        """Replaying an outage of hours is worse than losing it: the window is
+        what keeps a restart bounded."""
+        from adapter import _profile_scope_key, _restore_realtime_cursor
+
+        stale = datetime.now(UTC) - timedelta(seconds=1000)
+        self._store(
+            monkeypatch,
+            tmp_path,
+            {
+                "scope": _profile_scope_key(),
+                "cursor": "cur-old",
+                "saved_at": stale.isoformat(),
+            },
+        )
+        assert _restore_realtime_cursor(self._cfg(resume_window=900)) is None
+
+    def test_a_cursor_written_by_another_profile_is_ignored(
+        self, monkeypatch, tmp_path
+    ):
+        from adapter import _restore_realtime_cursor
+
+        self._store(
+            monkeypatch,
+            tmp_path,
+            {
+                "scope": "some-other-profile",
+                "cursor": "cur-other",
+                "saved_at": datetime.now(UTC).isoformat(),
+            },
+        )
+        assert _restore_realtime_cursor(self._cfg()) is None
+
+    def test_an_unreadable_file_means_no_resume(self, monkeypatch, tmp_path):
+        from adapter import _restore_realtime_cursor
+
+        path = tmp_path / "chatto_realtime_cursor.json"
+        monkeypatch.setattr("adapter._realtime_cursor_file", lambda: path)
+        path.write_text("{not json", encoding="utf-8")
+        assert _restore_realtime_cursor(self._cfg()) is None
+
+    def test_window_zero_disables_the_stored_cursor(self, monkeypatch, tmp_path):
+        from adapter import _persist_realtime_cursor, _restore_realtime_cursor
+
+        path = tmp_path / "chatto_realtime_cursor.json"
+        monkeypatch.setattr("adapter._realtime_cursor_file", lambda: path)
+        cfg = self._cfg(resume_window=0)
+        _persist_realtime_cursor(cfg, "cur-1")
+        assert not path.exists()
+        assert _restore_realtime_cursor(cfg) is None
+
+    async def test_disconnect_stores_the_last_applied_cursor(
+        self, monkeypatch, tmp_path
+    ):
+        """The shutdown path is the only writer — one small file instead of a
+        write on every presence change."""
+        path = tmp_path / "chatto_realtime_cursor.json"
+        monkeypatch.setattr("adapter._realtime_cursor_file", lambda: path)
+        adapter = _make_adapter()
+        adapter._chatto_client = None
+        adapter._realtime_cursor = "cur-final"
+
+        await adapter.disconnect()
+
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        assert payload["cursor"] == "cur-final"
+
+
 # -- chat_type mapping --