From bc0e5adb1de7f80b81f2c1eacc1acb46da05a18c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 17 Jul 2026 03:43:42 -0700 Subject: [PATCH] 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 | 55 ++++++++++++++++++- plugins/platforms/discord/recovery.py | 11 ++++ .../test_discord_missed_message_backfill.py | 33 +++++++++++ 3 files changed, 97 insertions(+), 2 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 6daf2c5e8..30c4fd2f4 100644 --- a/plugins/platforms/discord/adapter.py +++ b/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): diff --git a/plugins/platforms/discord/recovery.py b/plugins/platforms/discord/recovery.py index cac4072b6..060197e48 100644 --- a/plugins/platforms/discord/recovery.py +++ b/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,), + ) diff --git a/tests/gateway/test_discord_missed_message_backfill.py b/tests/gateway/test_discord_missed_message_backfill.py index 2d54cdeda..63592270b 100644 --- a/tests/gateway/test_discord_missed_message_backfill.py +++ b/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