feat(discord): recover messages missed during reconnect

contributors/emails/emodoteth@gmail.com

@@ -0,0 +1 @@
+emo-eth

contributors/emails/wenzel.james.r@gmail.com

@@ -0,0 +1 @@
+emo-eth

gateway/platforms/helpers.py

@@ -70,6 +70,22 @@ def is_duplicate(self, msg_id: str) -> bool:
                 self._seen = dict(newest)
         return False
 
+    def contains(self, msg_id: str) -> bool:
+        """Return whether *msg_id* is live in the cache without inserting it."""
+        if not msg_id:
+            return False
+        seen_at = self._seen.get(msg_id)
+        if seen_at is None:
+            return False
+        if time.time() - seen_at < self._ttl:
+            return True
+        del self._seen[msg_id]
+        return False
+
+    def discard(self, msg_id: str) -> None:
+        """Release a claimed message ID after cancelled/failed handoff."""
+        self._seen.pop(msg_id, None)
+
     def clear(self):
         """Clear all tracked messages."""
         self._seen.clear()

gateway/stream_consumer.py

@@ -240,6 +240,8 @@ def _metadata_for_send(
         final-message delivery.
         """
         meta = dict(self.metadata) if self.metadata else {}
+        if self._initial_reply_to_id:
+            meta["reply_to_message_id"] = self._initial_reply_to_id
         if expect_edits:
             meta["expect_edits"] = True
         if final:

hermes_cli/backup.py

@@ -774,6 +774,7 @@ def run_import(args) -> None:
     "channel_directory.json",
     "channel_aliases.json",
     "processes.json",
+    "gateway/discord_message_recovery.db",  # Discord reconnect replay ledger
     # Per-profile user-created stores that live outside the git checkout and
     # are therefore destroyed if the update flow removes/replaces the file and
     # the post-update schema-init re-creates an empty one (issue #52889). All

hermes_cli/config.py

@@ -2532,6 +2532,13 @@ def _ensure_hermes_home_managed(home: Path):
         "bots_require_inline_mention": False,  # Multi-bot rooms: if True, another bot must type @thisbot in its message to trigger a reply; a Discord reply/quote alone won't. Prevents two bots auto-replying to each other forever. Does not affect humans.
         "history_backfill": True,         # If True, prepend recent channel scrollback when bot is triggered (recovers messages missed while require_mention gated them out)
         "history_backfill_limit": 50,     # Max number of recent messages to scan when assembling the backfill block
+        "missed_message_backfill": {
+            "enabled": False,             # Replay missed Discord messages after reconnect/startup
+            "channels": "",               # Comma-separated channel IDs; empty uses free_response_channels
+            "window_seconds": 21600,      # Only inspect messages from the last 6 hours
+            "limit": 100,                 # Global cap on messages scanned per reconnect
+            "max_dispatches": 10,         # Cap on recovered messages dispatched per reconnect
+        },
         "reactions": True,             # Add 👀/✅/❌ reactions to messages during processing
         # Discord Gateway transport health. These settings inspect the active
         # WebSocket's ready/open/heartbeat state; they never use Discord REST as

plugins/platforms/discord/recovery.py

@@ -0,0 +1,110 @@
+"""Durable state for Discord reconnect message recovery."""
+
+from __future__ import annotations
+
+import datetime as dt
+import logging
+import os
+import sqlite3
+import threading
+from contextlib import suppress
+from pathlib import Path
+from typing import Any, Callable
+
+from hermes_constants import get_hermes_home
+
+logger = logging.getLogger(__name__)
+
+_DB_FILENAME = "discord_message_recovery.db"
+_RETENTION_DAYS = 30
+
+
+class DiscordRecoveryStore:
+    """Small profile-scoped SQLite ledger for completed Discord messages."""
+
+    def __init__(self, hermes_home: Path | None = None) -> None:
+        self._lock = threading.Lock()
+        self._initialized = False
+        self._hermes_home = Path(hermes_home or get_hermes_home())
+
+    def path(self) -> Path:
+        directory = self._hermes_home / "gateway"
+        directory.mkdir(parents=True, exist_ok=True)
+        return directory / _DB_FILENAME
+
+    def call(self, fn: Callable[[sqlite3.Connection], Any], default: Any = None) -> Any:
+        try:
+            with self._lock:
+                path = self.path()
+                conn = sqlite3.connect(path, timeout=0.1)
+                try:
+                    if not self._initialized:
+                        self._initialize(conn)
+                        self._initialized = True
+                        with suppress(OSError):
+                            os.chmod(path, 0o600)
+                    result = fn(conn)
+                    conn.commit()
+                    return result
+                finally:
+                    conn.close()
+        except Exception as exc:
+            logger.warning("Discord recovery ledger unavailable: %s", exc)
+            return default
+
+    def _initialize(self, conn: sqlite3.Connection) -> None:
+        conn.execute("PRAGMA journal_mode=WAL")
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS discord_messages (
+                message_id TEXT PRIMARY KEY,
+                channel_id TEXT,
+                thread_id TEXT,
+                parent_channel_id TEXT,
+                author_id TEXT,
+                created_at TEXT,
+                status TEXT NOT NULL,
+                replied INTEGER NOT NULL DEFAULT 0,
+                emoji_ack INTEGER NOT NULL DEFAULT 0,
+                outage_response INTEGER NOT NULL DEFAULT 0,
+                response_message_id TEXT,
+                attempts INTEGER NOT NULL DEFAULT 0,
+                last_attempt_at TEXT,
+                last_error TEXT,
+                updated_at TEXT NOT NULL
+            )
+        """)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS discord_recovery_scans (
+                scan_id TEXT PRIMARY KEY,
+                started_at TEXT NOT NULL,
+                completed_at TEXT,
+                status TEXT NOT NULL,
+                channels TEXT NOT NULL,
+                window_seconds REAL NOT NULL,
+                limit_count INTEGER NOT NULL,
+                scanned INTEGER NOT NULL DEFAULT 0,
+                missed INTEGER NOT NULL DEFAULT 0,
+                dispatched INTEGER NOT NULL DEFAULT 0,
+                error TEXT
+            )
+        """)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS discord_recovery_cursors (
+                channel_id TEXT PRIMARY KEY,
+                last_message_id TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            )
+        """)
+        cutoff = (
+            dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=_RETENTION_DAYS)
+        ).isoformat()
+        conn.execute("DELETE FROM discord_messages WHERE updated_at < ?", (cutoff,))
+        conn.execute(
+            "DELETE FROM discord_recovery_scans "
+            "WHERE COALESCE(completed_at, started_at) < ?",
+            (cutoff,),
+        )
+        conn.execute(
+            "DELETE FROM discord_recovery_cursors WHERE updated_at < ?",
+            (cutoff,),
+        )