fix(discord): harden reconnect message recovery

Route recovered messages through the live Discord ingress policy, preserve dedup and completion invariants, bound and retain the recovery ledger, and expose the opt-in config with docs and backup coverage.

gateway/platforms/helpers.py

@@ -70,6 +70,18 @@ class MessageDeduplicator:
                 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 clear(self):
         """Clear all tracked messages."""
         self._seen.clear()

hermes_cli/backup.py

@@ -774,6 +774,7 @@ _QUICK_STATE_FILES = (
     "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 @@ DEFAULT_CONFIG = {
         "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/adapter.py

@@ -10,6 +10,7 @@ Uses discord.py library for:
 """
 
 import asyncio
+import datetime as dt
 import hashlib
 import inspect
 import json
@@ -52,6 +53,7 @@ _DISCORD_COMMAND_SYNC_STATE_SUBDIR = "gateway"
 _DISCORD_COMMAND_SYNC_STATE_FILENAME = "discord_command_sync_state.json"
 _DISCORD_NONCONVERSATIONAL_STATE_FILENAME = "discord_nonconversational_messages.json"
 _DISCORD_RECOVERY_DB_FILENAME = "discord_message_recovery.db"
+_DISCORD_RECOVERY_RETENTION_DAYS = 30
 _DISCORD_COMMAND_SYNC_MUTATION_INTERVAL_SECONDS = 4.5
 _DISCORD_COMMAND_SYNC_MAX_RATE_LIMIT_SLEEP_SECONDS = 30.0
 # Discord enforces a hard cap of 100 global application (slash) commands per
@@ -1176,113 +1178,7 @@ class DiscordAdapter(BasePlatformAdapter):
 
             @self._client.event
             async def on_message(message: DiscordMessage):
-                # Block until _resolve_allowed_usernames has swapped
-                # any raw usernames in DISCORD_ALLOWED_USERS for numeric
-                # IDs (otherwise on_message's author.id lookup can miss).
-                if not adapter_self._ready_event.is_set():
-                    try:
-                        await asyncio.wait_for(adapter_self._ready_event.wait(), timeout=30.0)
-                    except asyncio.TimeoutError:
-                        pass
-
-                # Dedup: Discord RESUME replays events after reconnects (#4777)
-                if adapter_self._dedup.is_duplicate(str(message.id)):
-                    return
-
-                # Always ignore our own messages
-                if message.author == self._client.user:
-                    return
-
-                # Ignore Discord system messages (thread renames, pins, member joins, etc.)
-                # Allow both default and reply types — replies have a distinct MessageType.
-                if message.type not in {discord.MessageType.default, discord.MessageType.reply}:
-                    return
-
-                # Bot message filtering (DISCORD_ALLOW_BOTS):
-                #   "none"     — ignore all other bots (default)
-                #   "mentions" — accept bot messages only when they @mention us
-                #   "all"      — accept all bot messages
-                # Must run BEFORE the user allowlist check so that bots
-                # permitted by DISCORD_ALLOW_BOTS are not rejected for
-                # not being in DISCORD_ALLOWED_USERS (fixes #4466).
-                _role_authorized = False
-                if getattr(message.author, "bot", False):
-                    allow_bots = os.getenv("DISCORD_ALLOW_BOTS", "none").lower().strip()
-                    if allow_bots == "none":
-                        return
-                    elif allow_bots == "mentions":
-                        if not self._self_is_explicitly_mentioned(message):
-                            return
-                    if (
-                        self._discord_bots_require_inline_mention()
-                        and not self._self_is_raw_mentioned(message)
-                    ):
-                        return
-                    # "all" falls through; bot is permitted — skip the
-                    # human-user allowlist below (bots aren't in it).
-                else:
-                    # Non-bot: enforce the configured user/role allowlists.
-                    # Pass guild + is_dm so role checks are scoped to the
-                    # originating guild (prevents cross-guild DM bypass, see
-                    # _is_allowed_user docstring).
-                    _msg_guild = getattr(message, "guild", None)
-                    _is_dm = isinstance(message.channel, discord.DMChannel) or _msg_guild is None
-                    _msg_channel_ids = None
-                    if not _is_dm:
-                        _msg_channel_ids = {str(message.channel.id)}
-                        _parent_id = adapter_self._get_parent_channel_id(message.channel)
-                        if _parent_id:
-                            _msg_channel_ids.add(_parent_id)
-                    if not self._is_allowed_user(
-                        str(message.author.id),
-                        message.author,
-                        guild=_msg_guild,
-                        is_dm=_is_dm,
-                        channel_ids=_msg_channel_ids,
-                    ):
-                        self._warn_if_fail_closed_default()
-                        return
-                    _role_authorized = bool(getattr(self, "_allowed_role_ids", set()))
-                
-                # Multi-agent filtering: if the message mentions specific bots
-                # but NOT this bot, the sender is talking to another agent —
-                # stay silent.  Messages with no bot mentions (general chat)
-                # still fall through to _handle_message for the existing
-                # DISCORD_REQUIRE_MENTION check.
-                #
-                # This replaces the older DISCORD_IGNORE_NO_MENTION logic
-                # with bot-aware filtering that works correctly when multiple
-                # agents share a channel.
-                _raw_self_mention = self._self_is_explicitly_mentioned(message)
-                if not isinstance(message.channel, discord.DMChannel) and (
-                    message.mentions or _raw_self_mention
-                ):
-                    _self_mentioned = _raw_self_mention
-                    _other_bots_mentioned = any(
-                        m.bot and m != self._client.user
-                        for m in message.mentions
-                    )
-                    # If other bots are mentioned but we're not → not for us
-                    if _other_bots_mentioned and not _self_mentioned:
-                        return
-                    # If humans are mentioned but we're not → not for us
-                    # (preserves old DISCORD_IGNORE_NO_MENTION=true behavior)
-                    # EXCEPT in free-response channels where the bot should
-                    # answer regardless of who is mentioned.
-                    _ignore_no_mention = os.getenv(
-                        "DISCORD_IGNORE_NO_MENTION", "true"
-                    ).lower() in {"true", "1", "yes"}
-                    if _ignore_no_mention and not _self_mentioned and not _other_bots_mentioned:
-                        _channel_id = str(message.channel.id)
-                        _parent_id = None
-                        if hasattr(message.channel, "parent_id") and message.channel.parent_id:
-                            _parent_id = str(message.channel.parent_id)
-                        _free_channels = adapter_self._discord_free_response_channels()
-                        _channel_keys = adapter_self._discord_channel_keys(message, _parent_id)
-                        if "*" not in _free_channels and not (_channel_keys & _free_channels):
-                            return
-
-                await self._handle_message(message, role_authorized=_role_authorized)
+                await adapter_self._dispatch_discord_message(message)
 
             @self._client.event
             async def on_voice_state_update(member, before, after):
@@ -1357,6 +1253,78 @@ class DiscordAdapter(BasePlatformAdapter):
             self._release_platform_lock()
             return False
 
+    async def _dispatch_discord_message(self, message: DiscordMessage) -> bool:
+        """Apply Discord ingress policy and dispatch one live or recovered event."""
+        if not self._ready_event.is_set():
+            try:
+                await asyncio.wait_for(self._ready_event.wait(), timeout=30.0)
+            except asyncio.TimeoutError:
+                pass
+
+        if self._dedup.is_duplicate(str(message.id)):
+            return False
+        if message.author == self._client.user:
+            return False
+        if message.type not in {discord.MessageType.default, discord.MessageType.reply}:
+            return False
+
+        role_authorized = False
+        if getattr(message.author, "bot", False):
+            allow_bots = os.getenv("DISCORD_ALLOW_BOTS", "none").lower().strip()
+            if allow_bots == "none":
+                return False
+            if allow_bots == "mentions" and not self._self_is_explicitly_mentioned(message):
+                return False
+            if (
+                self._discord_bots_require_inline_mention()
+                and not self._self_is_raw_mentioned(message)
+            ):
+                return False
+        else:
+            msg_guild = getattr(message, "guild", None)
+            is_dm = isinstance(message.channel, discord.DMChannel) or msg_guild is None
+            msg_channel_ids = None
+            if not is_dm:
+                msg_channel_ids = {str(message.channel.id)}
+                parent_id = self._get_parent_channel_id(message.channel)
+                if parent_id:
+                    msg_channel_ids.add(parent_id)
+            if not self._is_allowed_user(
+                str(message.author.id),
+                message.author,
+                guild=msg_guild,
+                is_dm=is_dm,
+                channel_ids=msg_channel_ids,
+            ):
+                self._warn_if_fail_closed_default()
+                return False
+            role_authorized = bool(getattr(self, "_allowed_role_ids", set()))
+
+        raw_self_mention = self._self_is_explicitly_mentioned(message)
+        if not isinstance(message.channel, discord.DMChannel) and (
+            message.mentions or raw_self_mention
+        ):
+            other_bots_mentioned = any(
+                mentioned.bot and mentioned != self._client.user
+                for mentioned in message.mentions
+            )
+            if other_bots_mentioned and not raw_self_mention:
+                return False
+            ignore_no_mention = os.getenv(
+                "DISCORD_IGNORE_NO_MENTION", "true"
+            ).lower() in {"true", "1", "yes"}
+            if ignore_no_mention and not raw_self_mention and not other_bots_mentioned:
+                parent_id = None
+                if hasattr(message.channel, "parent_id") and message.channel.parent_id:
+                    parent_id = str(message.channel.parent_id)
+                free_channels = self._discord_free_response_channels()
+                channel_keys = self._discord_channel_keys(message, parent_id)
+                if "*" not in free_channels and not (channel_keys & free_channels):
+                    return False
+
+        await self._handle_message(message, role_authorized=role_authorized)
+        return True
+
     async def _cancel_bot_task(self) -> None:
         """Cancel and await the background client.start() task, if running."""
         if self._bot_task and not self._bot_task.done():
@@ -1999,7 +1967,10 @@ class DiscordAdapter(BasePlatformAdapter):
                 scanned += 1
                 message_id = str(getattr(message, "id", ""))
                 self._record_discord_message_seen(message, status="discovered")
-                if self._dedup.is_duplicate(message_id):
+                # A live gateway event may race this REST scan. Check without
+                # claiming the ID; the shared ingress helper owns the dedup
+                # write immediately before normal auth/filter dispatch.
+                if self._dedup.contains(message_id):
                     continue
                 if not await self._should_backfill_discord_message(message):
                     continue
@@ -2012,8 +1983,8 @@ class DiscordAdapter(BasePlatformAdapter):
                 )
                 self._record_recovery_attempt(message, status="queued")
                 try:
-                    await self._handle_message(message)
-                    dispatched += 1
+                    if await self._dispatch_recovered_message(message):
+                        dispatched += 1
                 except Exception as exc:
                     self._record_recovery_attempt(message, status="failed", error=str(exc))
                     raise
@@ -2034,6 +2005,10 @@ class DiscordAdapter(BasePlatformAdapter):
             self._record_recovery_scan_complete(scan_id, status="failed", scanned=scanned, missed=missed, dispatched=dispatched, error=str(exc))
             logger.warning("[%s] Missed-message backfill failed: %s", self.name, exc, exc_info=True)
 
+    async def _dispatch_recovered_message(self, message: Any) -> bool:
+        """Run one recovered message through the live Discord ingress gates."""
+        return await self._dispatch_discord_message(message)
+
     async def _iter_missed_message_backfill_candidates(self, channel_ids: set[str]):
         if not self._client:
             return
@@ -2064,9 +2039,13 @@ class DiscordAdapter(BasePlatformAdapter):
                         continue
                 candidate_channels.append(channel)
 
+        yielded = 0
         for channel in candidate_channels:
             async for item in self._iter_channel_and_thread_messages(channel, limit=limit, after=after, seen_channels=seen):
                 yield item
+                yielded += 1
+                if yielded >= limit:
+                    return
 
     async def _iter_channel_and_thread_messages(self, channel: Any, *, limit: int, after: Any, seen_channels: set[str]):
         """Yield history from a channel plus active/recent archived child threads."""
@@ -2245,6 +2224,19 @@ class DiscordAdapter(BasePlatformAdapter):
                             error TEXT
                         )
                     """)
+                    cutoff = (
+                        dt.datetime.now(dt.timezone.utc)
+                        - dt.timedelta(days=_DISCORD_RECOVERY_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,),
+                    )
                     result = fn(conn)
                     conn.commit()
                     return result
@@ -2336,13 +2328,13 @@ class DiscordAdapter(BasePlatformAdapter):
         message_id = str(getattr(getattr(event, "raw_message", None), "id", "") or getattr(event, "message_id", "") or "")
         if not message_id:
             return
-        status = "responded" if outcome == ProcessingOutcome.SUCCESS else ("cancelled" if outcome == ProcessingOutcome.CANCELLED else "failed")
+        status = "processed" if outcome == ProcessingOutcome.SUCCESS else ("cancelled" if outcome == ProcessingOutcome.CANCELLED else "failed")
         now = self._utc_now_iso()
 
         def _op(conn):
             conn.execute(
-                "UPDATE discord_messages SET status=?, replied=CASE WHEN ? THEN 1 ELSE replied END, updated_at=? WHERE message_id=?",
-                (status, 1 if outcome == ProcessingOutcome.SUCCESS else 0, now, message_id),
+                "UPDATE discord_messages SET status=?, updated_at=? WHERE message_id=?",
+                (status, now, message_id),
             )
 
         self._with_discord_recovery_db(_op)

tests/gateway/test_discord_missed_message_backfill.py

@@ -1,5 +1,6 @@
 """Tests for Discord missed-message startup backfill."""
 
+import datetime as dt
 import os
 import sys
 from datetime import datetime, timezone
@@ -9,6 +10,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 from gateway.config import PlatformConfig
+from gateway.platforms.base import MessageEvent, MessageType, ProcessingOutcome
 
 
 def _ensure_discord_mock():
@@ -45,6 +47,7 @@ def _ensure_discord_mock():
 
 _ensure_discord_mock()
 
+import discord  # noqa: E402
 from plugins.platforms.discord.adapter import DiscordAdapter  # noqa: E402
 
 
@@ -81,13 +84,16 @@ def adapter(monkeypatch, tmp_path):
     monkeypatch.setenv("HERMES_HOME", str(tmp_path))
     config = PlatformConfig(enabled=True, token="fake-token")
     adapter = DiscordAdapter(config)
-    adapter._client = SimpleNamespace(user=SimpleNamespace(id=999), get_channel=lambda _id: None)
+    bot_user = SimpleNamespace(id=999, bot=True, display_name="Hermes", name="hermes")
+    adapter._client = SimpleNamespace(user=bot_user, get_channel=lambda _id: None)
+    adapter._ready_event.set()
     adapter._handle_message = AsyncMock()
     monkeypatch.setenv("DISCORD_MISSED_MESSAGE_BACKFILL", "true")
+    monkeypatch.setenv("DISCORD_ALLOW_ALL_USERS", "true")
     return adapter
 
 
-def make_message(*, message_id=1, author_id=42, content="please ingest", reactions=None, channel=None):
+def make_message(*, message_id=1, author_id=42, content="please ingest", reactions=None, channel=None, mentions=None):
     channel = channel or FakeChannel()
     return SimpleNamespace(
         id=message_id,
@@ -95,11 +101,12 @@ def make_message(*, message_id=1, author_id=42, content="please ingest", reactio
         reactions=list(reactions or []),
         author=SimpleNamespace(id=author_id, bot=False, display_name="Emo", name="emo"),
         channel=channel,
+        guild=getattr(channel, "guild", None),
         created_at=datetime.now(timezone.utc),
         attachments=[],
-        mentions=[],
+        mentions=list(mentions or []),
         reference=None,
-        type=None,
+        type=discord.MessageType.default,
     )
 
 
@@ -186,7 +193,58 @@ async def test_run_backfill_dispatches_unaddressed_messages(adapter, monkeypatch
 
     await adapter._run_missed_message_backfill()
 
-    adapter._handle_message.assert_awaited_once_with(message)
+    adapter._handle_message.assert_awaited_once_with(message, role_authorized=False)
+
+
+@pytest.mark.asyncio
+async def test_run_backfill_counts_only_messages_that_reach_dispatch(adapter, monkeypatch):
+    dropped = make_message(message_id=1)
+    accepted = make_message(message_id=2)
+
+    async def fake_candidates(_channels):
+        yield dropped
+        yield accepted
+
+    async def fake_dispatch(message):
+        return message is accepted
+
+    monkeypatch.setattr(adapter, "_iter_missed_message_backfill_candidates", fake_candidates)
+    monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True))
+    dispatch = AsyncMock(side_effect=fake_dispatch)
+    monkeypatch.setattr(adapter, "_dispatch_recovered_message", dispatch)
+    monkeypatch.setattr(adapter, "_missed_message_backfill_max_dispatches", lambda: 1)
+    monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"})
+
+    await adapter._run_missed_message_backfill()
+
+    assert dispatch.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_recovered_mention_reuses_live_auth_and_mention_gates(adapter, monkeypatch):
+    bot_user = adapter._client.user
+    monkeypatch.delenv("DISCORD_ALLOW_ALL_USERS", raising=False)
+    denied = make_message(
+        message_id=1,
+        author_id=41,
+        content=f"<@{bot_user.id}> denied",
+        mentions=[bot_user],
+    )
+    allowed = make_message(
+        message_id=2,
+        content=f"<@{bot_user.id}> allowed",
+        mentions=[bot_user],
+    )
+
+    monkeypatch.setattr(
+        adapter,
+        "_is_allowed_user",
+        lambda user_id, *_a, **_kw: user_id == str(allowed.author.id),
+    )
+
+    assert await adapter._dispatch_recovered_message(denied) is False
+    assert await adapter._dispatch_recovered_message(allowed) is True
+    adapter._handle_message.assert_awaited_once_with(allowed, role_authorized=False)
 
 
 def test_missed_message_backfill_config_bridge(monkeypatch, tmp_path):
@@ -224,6 +282,18 @@ def test_missed_message_backfill_config_bridge(monkeypatch, tmp_path):
     assert os.environ["DISCORD_MISSED_MESSAGE_BACKFILL_MAX_DISPATCHES"] == "3"
 
 
+def test_default_config_exposes_missed_message_backfill_settings():
+    from hermes_cli.config import DEFAULT_CONFIG
+
+    assert DEFAULT_CONFIG["discord"]["missed_message_backfill"] == {
+        "enabled": False,
+        "channels": "",
+        "window_seconds": 21600,
+        "limit": 100,
+        "max_dispatches": 10,
+    }
+
+
 @pytest.mark.asyncio
 async def test_persistent_responded_record_suppresses_backfill(adapter):
     message = make_message(message_id=77)
@@ -247,6 +317,51 @@ def test_down_notice_response_does_not_mark_message_complete(adapter):
     assert adapter._discord_message_is_persistently_complete("88") is False
 
 
+def test_recovery_ledger_prunes_expired_rows(adapter):
+    old = (datetime.now(timezone.utc) - dt.timedelta(days=31)).isoformat()
+
+    def insert_old_rows(conn):
+        conn.execute(
+            "INSERT INTO discord_messages "
+            "(message_id, status, updated_at) VALUES ('old-message', 'responded', ?)",
+            (old,),
+        )
+        conn.execute(
+            "INSERT INTO discord_recovery_scans "
+            "(scan_id, started_at, completed_at, status, channels, window_seconds, limit_count) "
+            "VALUES ('old-scan', ?, ?, 'success', '[]', 3600, 10)",
+            (old, old),
+        )
+
+    adapter._with_discord_recovery_db(insert_old_rows)
+    adapter._with_discord_recovery_db(lambda _conn: None)
+
+    def count_old(conn):
+        messages = conn.execute(
+            "SELECT COUNT(*) FROM discord_messages WHERE message_id='old-message'"
+        ).fetchone()[0]
+        scans = conn.execute(
+            "SELECT COUNT(*) FROM discord_recovery_scans WHERE scan_id='old-scan'"
+        ).fetchone()[0]
+        return messages, scans
+
+    assert adapter._with_discord_recovery_db(count_old) == (0, 0)
+
+
+def test_empty_successful_turn_is_not_persistently_complete(adapter):
+    message = make_message(message_id=89)
+    event = MessageEvent(
+        text=message.content,
+        message_type=MessageType.TEXT,
+        raw_message=message,
+        message_id=str(message.id),
+    )
+    adapter._record_discord_processing_start(event, emoji_ack=False)
+    adapter._record_discord_processing_complete(event, outcome=ProcessingOutcome.SUCCESS)
+
+    assert adapter._discord_message_is_persistently_complete("89") is False
+
+
 @pytest.mark.asyncio
 async def test_iter_candidates_includes_active_and_archived_threads(adapter):
     active_msg = make_message(message_id=201, channel=FakeChannel(channel_id=2010))
@@ -270,3 +385,23 @@ async def test_iter_candidates_includes_active_and_archived_threads(adapter):
         got.append(msg.id)
 
     assert got == [201, 202]
+
+
+@pytest.mark.asyncio
+async def test_iter_candidates_applies_one_global_scan_limit(adapter, monkeypatch):
+    first = FakeChannel(
+        channel_id=123,
+        history_messages=[make_message(message_id=1), make_message(message_id=2)],
+    )
+    second = FakeChannel(
+        channel_id=456,
+        history_messages=[make_message(message_id=3), make_message(message_id=4)],
+    )
+    adapter._client.get_channel = lambda channel_id: {123: first, 456: second}[channel_id]
+    monkeypatch.setattr(adapter, "_missed_message_backfill_limit", lambda: 3)
+
+    got = []
+    async for msg in adapter._iter_missed_message_backfill_candidates({"123", "456"}):
+        got.append(msg.id)
+
+    assert got == [1, 2, 3]

tests/gateway/test_message_deduplicator.py

@@ -60,6 +60,19 @@ class TestMessageDeduplicatorTTL:
         assert dedup.is_duplicate("") is False
         assert dedup.is_duplicate("") is False
 
+    def test_contains_does_not_claim_unseen_message(self):
+        dedup = MessageDeduplicator(ttl_seconds=60)
+
+        assert dedup.contains("msg-1") is False
+        assert dedup.is_duplicate("msg-1") is False
+
+    def test_contains_expires_stale_message_without_refreshing_it(self):
+        dedup = MessageDeduplicator(ttl_seconds=5)
+        dedup._seen["msg-1"] = time.time() - 10
+
+        assert dedup.contains("msg-1") is False
+        assert "msg-1" not in dedup._seen
+
     def test_max_size_eviction_prunes_expired(self):
         """Cache pruning on overflow removes expired entries."""
         dedup = MessageDeduplicator(max_size=5, ttl_seconds=60)

tests/hermes_cli/test_backup.py

@@ -1489,6 +1489,26 @@ class TestQuickSnapshot:
         snap_id = create_quick_snapshot(hermes_home=hermes_home)
         assert (hermes_home / "state-snapshots" / snap_id / "cron" / "jobs.json").exists()
 
+    def test_copies_discord_recovery_ledger(self, hermes_home):
+        from hermes_cli.backup import create_quick_snapshot
+
+        gateway_dir = hermes_home / "gateway"
+        gateway_dir.mkdir()
+        ledger = gateway_dir / "discord_message_recovery.db"
+        conn = sqlite3.connect(ledger)
+        conn.execute("CREATE TABLE handled (message_id TEXT PRIMARY KEY)")
+        conn.execute("INSERT INTO handled VALUES ('123')")
+        conn.commit()
+        conn.close()
+
+        snap_id = create_quick_snapshot(hermes_home=hermes_home)
+
+        copied = hermes_home / "state-snapshots" / snap_id / "gateway" / ledger.name
+        assert copied.exists()
+        conn = sqlite3.connect(copied)
+        assert conn.execute("SELECT message_id FROM handled").fetchall() == [("123",)]
+        conn.close()
+
     def test_copies_channel_aliases(self, hermes_home):
         from hermes_cli.backup import create_quick_snapshot
         snap_id = create_quick_snapshot(hermes_home=hermes_home)

website/docs/user-guide/messaging/discord.md

@@ -341,6 +341,12 @@ discord:
   no_thread_channels: []          # Channel IDs where bot responds without threading
   history_backfill: true          # Prepend recent channel scrollback on mention (default: true)
   history_backfill_limit: 50      # Max messages to scan backwards (default: 50)
+  missed_message_backfill:        # Replay messages missed while disconnected (opt-in)
+    enabled: false
+    channels: []                  # Empty uses free_response_channels
+    window_seconds: 21600         # Look back at most 6 hours
+    limit: 100                    # Global scan cap per reconnect
+    max_dispatches: 10            # Recovery dispatch cap per reconnect
   channel_prompts: {}             # Per-channel ephemeral system prompts
   allow_mentions:                 # What the bot is allowed to ping (safe defaults)
     everyone: false               # @everyone / @here pings (default: false)
@@ -510,6 +516,24 @@ discord:
   history_backfill_limit: 50
 ```
 
+#### `discord.missed_message_backfill`
+
+**Type:** object — **Default:** disabled
+
+Discord's WebSocket resume window can expire during a restart or network outage. Messages sent during that gap are not delivered as live gateway events. When this option is enabled, Hermes scans a bounded set of configured channel and thread histories after Discord reconnects, then sends still-unhandled messages through the same authorization, mention, channel, deduplication, and dispatch path as live events.
+
+```yaml
+discord:
+  missed_message_backfill:
+    enabled: true
+    channels: ["123456789012345678"]
+    window_seconds: 3600
+    limit: 100
+    max_dispatches: 10
+```
+
+If `channels` is empty, Hermes uses `discord.free_response_channels`. Set it to `"*"` only when the bot should inspect every reachable server text channel. The recovery ledger is stored per profile under `gateway/discord_message_recovery.db`, preventing a successfully answered message from being replayed again after a later restart.
+
 #### `group_sessions_per_user`
 
 **Type:** boolean — **Default:** `true`

website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md

@@ -332,6 +332,12 @@ discord:
   no_thread_channels: []          # 机器人不创建线程直接响应的频道 ID
   history_backfill: true          # 在提及时前置最近的频道滚动历史（默认：true）
   history_backfill_limit: 50      # 向后扫描的最大消息数（默认：50）
+  missed_message_backfill:        # 重新处理断线期间遗漏的消息（需主动启用）
+    enabled: false
+    channels: []                  # 留空时使用 free_response_channels
+    window_seconds: 21600         # 最多回溯 6 小时
+    limit: 100                    # 每次重连的全局扫描上限
+    max_dispatches: 10            # 每次重连的恢复分发上限
   channel_prompts: {}             # 每个频道的临时系统 prompt（提示词）
   allow_mentions:                 # 机器人允许 ping 的内容（安全默认值）
     everyone: false               # @everyone / @here ping（默认：false）
@@ -501,6 +507,24 @@ discord:
   history_backfill_limit: 50
 ```
 
+#### `discord.missed_message_backfill`
+
+**类型：** 对象 — **默认值：** 禁用
+
+Discord 的 WebSocket 恢复窗口可能在重启或网络中断时过期。在此期间发送的消息不会作为实时网关事件交付。启用后，Hermes 会在 Discord 重连后扫描一组有界的频道与线程历史记录，并将尚未处理的消息交给与实时事件相同的授权、提及、频道、去重和分发流程。
+
+```yaml
+discord:
+  missed_message_backfill:
+    enabled: true
+    channels: ["123456789012345678"]
+    window_seconds: 3600
+    limit: 100
+    max_dispatches: 10
+```
+
+如果 `channels` 留空，Hermes 会使用 `discord.free_response_channels`。只有当机器人确实需要检查所有可访问的服务器文字频道时才设置为 `"*"`。恢复账本按配置文件存储在 `gateway/discord_message_recovery.db`，避免已成功回复的消息在后续重启时再次执行。
+
 #### `group_sessions_per_user`
 
 **类型：** 布尔值 — **默认值：** `true`