From 80744bc2bc4a8ff736f467c4510ef80d1961a3a9 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 17 Jul 2026 00:26:29 -0700 Subject: [PATCH] fix(discord): close reconnect recovery edge cases Include allowed mention-gated channels in default recovery scope, keep the newest bounded history window, narrow outage-message detection, avoid disabled-path ledger I/O, and persist forum replies. --- plugins/platforms/discord/adapter.py | 58 +++++++++----- .../test_discord_missed_message_backfill.py | 77 +++++++++++++++++++ 2 files changed, 115 insertions(+), 20 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 868b522c6147..32fdf84908f8 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -1912,7 +1912,12 @@ class DiscordAdapter(BasePlatformAdapter): """ raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", "") if not raw.strip(): - return self._discord_free_response_channels() + allowed = { + item.strip() + for item in os.getenv("DISCORD_ALLOWED_CHANNELS", "").split(",") + if item.strip() + } + return allowed | self._discord_free_response_channels() return {item.strip() for item in raw.split(",") if item.strip()} def _missed_message_backfill_window_seconds(self) -> float: @@ -2012,9 +2017,7 @@ class DiscordAdapter(BasePlatformAdapter): async def _iter_missed_message_backfill_candidates(self, channel_ids: set[str]): if not self._client: return - import datetime as _dt - - after = _dt.datetime.now(_dt.timezone.utc) - _dt.timedelta( + after = dt.datetime.now(dt.timezone.utc) - dt.timedelta( seconds=self._missed_message_backfill_window_seconds() ) limit = self._missed_message_backfill_limit() @@ -2057,10 +2060,18 @@ class DiscordAdapter(BasePlatformAdapter): history = getattr(channel, "history", None) if callable(history): try: + # Fetch the latest N messages in the window, then restore + # chronological dispatch order. With oldest_first=True the API + # returns the earliest N and can permanently starve newer work. + history_iter = history( + limit=limit, + after=after, + oldest_first=False, + ) messages = [] - async for message in history(limit=limit, after=after, oldest_first=True): + async for message in history_iter: # type: ignore[attr-defined] messages.append(message) - for message in messages: + for message in reversed(messages): yield message except Exception as exc: logger.debug("[%s] Cannot read history for %s: %s", self.name, channel_key, exc) @@ -2117,19 +2128,12 @@ class DiscordAdapter(BasePlatformAdapter): return False def _is_down_notice_content(self, content: str) -> bool: + """Recognize only explicit Hermes/gateway outage notices.""" text = (content or "").lower() - down_markers = ( - "agent is down", - "agent was down", - "gateway is down", - "gateway was down", - "bmo is down", - "currently down", - "offline", - "unavailable", - "not running", - ) - return any(marker in text for marker in down_markers) + subject = r"(?:hermes|the agent|agent|the gateway|gateway|bmo)" + state = r"(?:is|was|appears to be|is currently|was currently)" + condition = r"(?:down|offline|unavailable|not running)" + return re.search(rf"\b{subject}\s+{state}\s+{condition}\b", text) is not None async def _message_has_non_down_bot_response(self, message: Any) -> bool: """Detect an already-addressed message without trusting down notices.""" @@ -2259,6 +2263,8 @@ class DiscordAdapter(BasePlatformAdapter): return channel_id, thread_id, parent_id def _record_discord_message_seen(self, message: Any, *, status: str) -> None: + if not self._missed_message_backfill_enabled(): + return message_id = str(getattr(message, "id", "") or "") if not message_id: return @@ -2290,6 +2296,8 @@ class DiscordAdapter(BasePlatformAdapter): self._with_discord_recovery_db(_op) def _record_recovery_attempt(self, message: Any, *, status: str, error: Optional[str] = None) -> None: + if not self._missed_message_backfill_enabled(): + return self._record_discord_message_seen(message, status=status) message_id = str(getattr(message, "id", "") or "") if not message_id: @@ -2309,6 +2317,8 @@ class DiscordAdapter(BasePlatformAdapter): self._with_discord_recovery_db(_op) def _record_discord_processing_start(self, event: MessageEvent, *, emoji_ack: bool) -> None: + if not self._missed_message_backfill_enabled(): + return message = event.raw_message self._record_discord_message_seen(message, status="processing") message_id = str(getattr(message, "id", "") or getattr(event, "message_id", "") or "") @@ -2325,6 +2335,8 @@ class DiscordAdapter(BasePlatformAdapter): self._with_discord_recovery_db(_op) def _record_discord_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None: + if not self._missed_message_backfill_enabled(): + return message_id = str(getattr(getattr(event, "raw_message", None), "id", "") or getattr(event, "message_id", "") or "") if not message_id: return @@ -2340,7 +2352,7 @@ class DiscordAdapter(BasePlatformAdapter): self._with_discord_recovery_db(_op) def _record_discord_response(self, *, reply_to: Optional[str], result: SendResult, content: str) -> None: - if not reply_to: + if not self._missed_message_backfill_enabled() or not reply_to: return now = self._utc_now_iso() outage = self._is_down_notice_content(content) @@ -2684,7 +2696,13 @@ class DiscordAdapter(BasePlatformAdapter): # Forum channels reject channel.send() — create a thread post instead. if self._is_forum_parent(channel): - return await self._send_to_forum(channel, content) + result = await self._send_to_forum(channel, content) + self._record_discord_response( + reply_to=reply_to, + result=result, + content=content, + ) + return result # Format and split message if needed formatted = self.format_message(content) diff --git a/tests/gateway/test_discord_missed_message_backfill.py b/tests/gateway/test_discord_missed_message_backfill.py index b8ad52fd2ab4..31088c3bc79c 100644 --- a/tests/gateway/test_discord_missed_message_backfill.py +++ b/tests/gateway/test_discord_missed_message_backfill.py @@ -177,6 +177,21 @@ async def test_backfills_when_only_down_notice_exists(adapter): assert await adapter._should_backfill_discord_message(message) is True +@pytest.mark.asyncio +async def test_generic_unavailable_response_counts_as_completed(adapter): + bot_reply = SimpleNamespace( + id=2, + content="That package is unavailable on this platform.", + author=SimpleNamespace(id=999, bot=True), + reference=SimpleNamespace(message_id=1), + created_at=datetime.now(timezone.utc), + ) + channel = FakeChannel(history_messages=[bot_reply]) + message = make_message(message_id=1, channel=channel) + + assert await adapter._should_backfill_discord_message(message) is False + + @pytest.mark.asyncio async def test_run_backfill_dispatches_unaddressed_messages(adapter, monkeypatch): message = make_message(message_id=1) @@ -294,6 +309,14 @@ def test_default_config_exposes_missed_message_backfill_settings(): } +def test_default_recovery_scope_includes_allowed_and_free_response_channels(adapter, monkeypatch): + monkeypatch.delenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", raising=False) + monkeypatch.setenv("DISCORD_ALLOWED_CHANNELS", "100,200") + monkeypatch.setenv("DISCORD_FREE_RESPONSE_CHANNELS", "200,300") + + assert adapter._missed_message_backfill_channels() == {"100", "200", "300"} + + @pytest.mark.asyncio async def test_persistent_responded_record_suppresses_backfill(adapter): message = make_message(message_id=77) @@ -362,6 +385,28 @@ def test_empty_successful_turn_is_not_persistently_complete(adapter): assert adapter._discord_message_is_persistently_complete("89") is False +def test_disabled_recovery_does_not_create_hot_path_ledger(adapter, monkeypatch): + monkeypatch.setenv("DISCORD_MISSED_MESSAGE_BACKFILL", "false") + message = make_message(message_id=90) + 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, ProcessingOutcome.SUCCESS) + adapter._record_discord_response( + reply_to="90", + result=SimpleNamespace(success=True, message_id="9003"), + content="Done", + ) + + db_path = adapter._discord_recovery_db_path() + assert not db_path.exists() + + @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)) @@ -405,3 +450,35 @@ async def test_iter_candidates_applies_one_global_scan_limit(adapter, monkeypatc got.append(msg.id) assert got == [1, 2, 3] + + +@pytest.mark.asyncio +async def test_iter_candidates_keeps_latest_messages_when_window_exceeds_limit(adapter, monkeypatch): + class RealisticChannel(FakeChannel): + def history(self, **kwargs): + async def _gen(): + messages = list(self._history_messages) + if not kwargs["oldest_first"]: + messages.reverse() + for message in messages[:kwargs["limit"]]: + yield message + + return _gen() + + channel = RealisticChannel( + channel_id=123, + history_messages=[ + make_message(message_id=1), + make_message(message_id=2), + make_message(message_id=3), + make_message(message_id=4), + ], + ) + adapter._client.get_channel = lambda _channel_id: channel + monkeypatch.setattr(adapter, "_missed_message_backfill_limit", lambda: 3) + + got = [] + async for msg in adapter._iter_missed_message_backfill_candidates({"123"}): + got.append(msg.id) + + assert got == [2, 3, 4]