fix(discord): persist per-channel recovery cursors

Resume each configured Discord channel/thread after its last scanned message so busy sources cannot permanently starve later history windows.

plugins/platforms/discord/adapter.py

@@ -2054,8 +2054,18 @@ class DiscordAdapter(BasePlatformAdapter):
                 )
                 self._record_recovery_attempt(message, status="queued")
                 try:
-                    if await self._dispatch_recovered_message(message):
+                    admitted = await self._dispatch_recovered_message(message)
+                    if admitted:
                         dispatched += 1
+                    channel_id = str(
+                        getattr(getattr(message, "channel", None), "id", "")
+                    )
+                    if channel_id and message_id:
+                        await asyncio.to_thread(
+                            self._advance_discord_recovery_cursor,
+                            channel_id,
+                            message_id,
+                        )
                 except asyncio.CancelledError:
                     self._dedup.discard(message_id)
                     self._record_recovery_attempt(message, status="cancelled")
@@ -2141,7 +2151,12 @@ class DiscordAdapter(BasePlatformAdapter):
 
         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):
+            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:
@@ -2154,6 +2169,10 @@ class DiscordAdapter(BasePlatformAdapter):
             return
         seen_channels.add(channel_key)
 
+        cursor = self._discord_recovery_cursor(channel_key)
+        if cursor:
+            with suppress(ValueError, TypeError):
+                after = discord.Object(id=int(cursor))
         history = getattr(channel, "history", None)
         if callable(history):
             try:
@@ -2189,6 +2208,38 @@ class DiscordAdapter(BasePlatformAdapter):
             async for message in self._iter_channel_and_thread_messages(thread, limit=limit, after=after, seen_channels=seen_channels):
                 yield message
 
+    def _discord_recovery_cursor(self, channel_id: str) -> Optional[str]:
+        if not channel_id:
+            return None
+
+        def _op(conn):
+            row = conn.execute(
+                "SELECT last_message_id FROM discord_recovery_cursors WHERE channel_id=?",
+                (channel_id,),
+            ).fetchone()
+            return str(row[0]) if row else None
+
+        return self._with_discord_recovery_db(_op)
+
+    def _advance_discord_recovery_cursor(self, channel_id: str, message_id: str) -> None:
+        if not channel_id or not message_id:
+            return
+        now = self._utc_now_iso()
+
+        def _op(conn):
+            conn.execute(
+                """
+                INSERT INTO discord_recovery_cursors (channel_id, last_message_id, updated_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(channel_id) DO UPDATE SET
+                    last_message_id=excluded.last_message_id,
+                    updated_at=excluded.updated_at
+                """,
+                (channel_id, message_id, now),
+            )
+
+        self._with_discord_recovery_db(_op)
+
     async def _should_backfill_discord_message(self, message: Any) -> bool:
         """Return True when a recent Discord message still needs Hermes work."""
         if not self._client or not getattr(self._client, "user", None):

plugins/platforms/discord/recovery.py

@@ -88,6 +88,13 @@ class DiscordRecoveryStore:
                 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()
@@ -97,3 +104,7 @@ class DiscordRecoveryStore:
             "WHERE COALESCE(completed_at, started_at) < ?",
             (cutoff,),
         )
+        conn.execute(
+            "DELETE FROM discord_recovery_cursors WHERE updated_at < ?",
+            (cutoff,),
+        )

tests/gateway/test_discord_missed_message_backfill.py

@@ -30,6 +30,7 @@ def _ensure_discord_mock():
     discord_mod.Color = SimpleNamespace(orange=lambda: 1, green=lambda: 2, blue=lambda: 3, red=lambda: 4, purple=lambda: 5)
     discord_mod.Interaction = object
     discord_mod.Embed = MagicMock
+    discord_mod.Object = lambda *, id: SimpleNamespace(id=id)
     discord_mod.app_commands = SimpleNamespace(
         describe=lambda **kwargs: (lambda fn: fn),
         choices=lambda **kwargs: (lambda fn: fn),
@@ -853,3 +854,35 @@ async def test_iter_candidates_keeps_latest_messages_when_window_exceeds_limit(a
         got.append(msg.id)
 
     assert got == [2, 3, 4]
+
+
+def test_recovery_cursor_round_trip_is_channel_scoped(adapter):
+    adapter._advance_discord_recovery_cursor("123", "1001")
+    adapter._advance_discord_recovery_cursor("456", "2002")
+
+    assert adapter._discord_recovery_cursor("123") == "1001"
+    assert adapter._discord_recovery_cursor("456") == "2002"
+
+
+@pytest.mark.asyncio
+async def test_iter_candidates_uses_persisted_channel_cursor(adapter, monkeypatch):
+    class CursorChannel(FakeChannel):
+        def history(self, **kwargs):
+            self.history_kwargs = kwargs
+
+            async def _gen():
+                yield make_message(message_id=11, channel=self)
+
+            return _gen()
+
+    channel = CursorChannel(channel_id=123)
+    adapter._client.get_channel = lambda _channel_id: channel
+    adapter._advance_discord_recovery_cursor("123", "10")
+    monkeypatch.setattr(discord, "Object", lambda *, id: SimpleNamespace(id=id))
+
+    got = []
+    async for message in adapter._iter_missed_message_backfill_candidates({"123"}):
+        got.append(message.id)
+
+    assert got == [11]
+    assert getattr(channel.history_kwargs["after"], "id", None) == 10