From 303949acdc334377278e91d4245d3be0669f4518 Mon Sep 17 00:00:00 2001 From: James Date: Thu, 4 Jun 2026 12:02:45 -0700 Subject: [PATCH] fix: backfill missed Discord messages on startup (#3) --- plugins/platforms/discord/adapter.py | 525 +++++++++++++++++- .../test_discord_missed_message_backfill.py | 241 ++++++++ 2 files changed, 758 insertions(+), 8 deletions(-) create mode 100644 tests/gateway/test_discord_missed_message_backfill.py diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index b373ef4b887..609020a5456 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -17,6 +17,7 @@ import logging import math import os import re +import sqlite3 import struct import subprocess import tempfile @@ -50,6 +51,7 @@ _DISCORD_COMMAND_SYNC_POLICIES = {"safe", "bulk", "off"} _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_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 @@ -913,6 +915,8 @@ class DiscordAdapter(BasePlatformAdapter): # bot task's done callback uses this to distinguish an operator/service # shutdown from a runtime websocket crash. self._disconnecting = False + self._missed_message_backfill_task: Optional[asyncio.Task] = None + self._discord_recovery_db_lock = threading.Lock() # Dedup cache: prevents duplicate bot responses when Discord # RESUME replays events after reconnects. self._dedup = MessageDeduplicator() @@ -1160,6 +1164,15 @@ class DiscordAdapter(BasePlatformAdapter): adapter_self._post_connect_task = asyncio.create_task( adapter_self._run_post_connect_initialization() ) + if adapter_self._missed_message_backfill_enabled(): + if ( + adapter_self._missed_message_backfill_task + and not adapter_self._missed_message_backfill_task.done() + ): + adapter_self._missed_message_backfill_task.cancel() + adapter_self._missed_message_backfill_task = asyncio.create_task( + adapter_self._run_missed_message_backfill() + ) @self._client.event async def on_message(message: DiscordMessage): @@ -1636,12 +1649,19 @@ class DiscordAdapter(BasePlatformAdapter): await self._post_connect_task except asyncio.CancelledError: pass + if self._missed_message_backfill_task and not self._missed_message_backfill_task.done(): + self._missed_message_backfill_task.cancel() + try: + await self._missed_message_backfill_task + except asyncio.CancelledError: + pass self._running = False self._client = None self._ready_event.clear() self._post_connect_task = None self._liveness_task = None + self._missed_message_backfill_task = None self._release_platform_lock() @@ -1909,6 +1929,474 @@ class DiscordAdapter(BasePlatformAdapter): except Exception as e: # pragma: no cover - defensive logging logger.warning("[%s] Slash command sync failed: %s", self.name, e, exc_info=True) + def _missed_message_backfill_enabled(self) -> bool: + """Whether to reconcile Discord messages missed while the gateway was down.""" + raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL", "false") + return str(raw).strip().lower() in ("true", "1", "yes", "on") + + def _missed_message_backfill_channels(self) -> set[str]: + """Channels to scan for missed messages after Discord reconnects. + + Defaults to free-response channels because those are the places where + mention-free posts are expected to start work. Operators can set + DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS="*" to scan every reachable + text channel, but the safe default is scoped. + """ + raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", "") + if not raw.strip(): + return self._discord_free_response_channels() + return {item.strip() for item in raw.split(",") if item.strip()} + + def _missed_message_backfill_window_seconds(self) -> float: + raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_WINDOW_SECONDS", "21600") + try: + value = float(raw) + except (TypeError, ValueError): + value = 21600.0 + return max(60.0, value) + + def _missed_message_backfill_limit(self) -> int: + raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_LIMIT", "100") + try: + value = int(raw) + except (TypeError, ValueError): + value = 100 + return max(1, min(value, 500)) + + def _missed_message_backfill_max_dispatches(self) -> int: + raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_MAX_DISPATCHES", "10") + try: + value = int(raw) + except (TypeError, ValueError): + value = 10 + return max(1, min(value, 100)) + + async def _run_missed_message_backfill(self) -> None: + """Find and enqueue recent Discord messages missed while the bot was down. + + Discord gateway events are not replayed for messages sent while the bot + is offline. Normal startup resume only handles sessions already marked + resume_pending; this pass scans recent channel/thread history, records + what it saw durably, and reuses the normal message handler for messages + that lack a substantive non-outage Hermes response. Emoji-only acks are + deliberately not sufficient completion evidence. + """ + if not self._client: + return + channels = self._missed_message_backfill_channels() + scan_id = self._record_recovery_scan_start(channels) + if not channels: + logger.info("[%s] Missed-message backfill enabled but no channels configured", self.name) + self._record_recovery_scan_complete(scan_id, status="skipped", scanned=0, missed=0, dispatched=0) + return + + max_dispatches = self._missed_message_backfill_max_dispatches() + dispatched = 0 + scanned = 0 + missed = 0 + try: + async for message in self._iter_missed_message_backfill_candidates(channels): + scanned += 1 + message_id = str(getattr(message, "id", "")) + self._record_discord_message_seen(message, status="discovered") + if self._dedup.is_duplicate(message_id): + continue + if not await self._should_backfill_discord_message(message): + continue + missed += 1 + logger.info( + "[%s] Backfilling missed Discord message %s in channel %s", + self.name, + getattr(message, "id", "unknown"), + getattr(getattr(message, "channel", None), "id", "unknown"), + ) + self._record_recovery_attempt(message, status="queued") + try: + await self._handle_message(message) + dispatched += 1 + except Exception as exc: + self._record_recovery_attempt(message, status="failed", error=str(exc)) + raise + if dispatched >= max_dispatches: + break + self._record_recovery_scan_complete(scan_id, status="success", scanned=scanned, missed=missed, dispatched=dispatched) + logger.info( + "[%s] Missed-message backfill complete: scanned=%d missed=%d dispatched=%d", + self.name, + scanned, + missed, + dispatched, + ) + except asyncio.CancelledError: + self._record_recovery_scan_complete(scan_id, status="cancelled", scanned=scanned, missed=missed, dispatched=dispatched) + raise + except Exception as exc: # pragma: no cover - defensive logging + 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 _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( + seconds=self._missed_message_backfill_window_seconds() + ) + limit = self._missed_message_backfill_limit() + seen: set[str] = set() + + candidate_channels = [] + if "*" in channel_ids: + for guild in getattr(self._client, "guilds", []) or []: + candidate_channels.extend(getattr(guild, "text_channels", []) or []) + else: + for channel_id in sorted(channel_ids): + channel = None + try: + channel = self._client.get_channel(int(channel_id)) + except Exception: + channel = None + if channel is None: + try: + channel = await self._client.fetch_channel(int(channel_id)) + except Exception as exc: + logger.debug("[%s] Cannot fetch backfill channel %s: %s", self.name, channel_id, exc) + continue + candidate_channels.append(channel) + + 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 + + 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.""" + channel_key = str(getattr(channel, "id", "")) + if not channel_key or channel_key in seen_channels: + return + seen_channels.add(channel_key) + + history = getattr(channel, "history", None) + if callable(history): + try: + messages = [] + async for message in history(limit=limit, after=after, oldest_first=True): + messages.append(message) + for message in messages: + yield message + except Exception as exc: + logger.debug("[%s] Cannot read history for %s: %s", self.name, channel_key, exc) + + child_threads = list(getattr(channel, "threads", []) or []) + archived_threads = getattr(channel, "archived_threads", None) + if callable(archived_threads): + try: + async for thread in archived_threads(limit=limit): + child_threads.append(thread) + except Exception as exc: + logger.debug("[%s] Cannot list archived threads for %s: %s", self.name, channel_key, exc) + + for thread in child_threads: + thread_key = str(getattr(thread, "id", "")) + if not thread_key or thread_key in seen_channels: + continue + async for message in self._iter_channel_and_thread_messages(thread, limit=limit, after=after, seen_channels=seen_channels): + yield message + + 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): + return False + if getattr(getattr(message, "author", None), "id", None) == getattr(self._client.user, "id", None): + return False + if getattr(getattr(message, "author", None), "bot", False): + return False + if self._discord_message_is_persistently_complete(str(getattr(message, "id", ""))): + return False + # A success reaction alone is only an acknowledgement. It is not + # enough evidence that the substantive response/action completed. + if await self._message_has_non_down_bot_response(message): + return False + return True + + async def _message_has_own_reaction(self, message: Any, emojis: set[str]) -> bool: + bot_user = getattr(self._client, "user", None) if self._client else None + bot_id = getattr(bot_user, "id", None) + for reaction in getattr(message, "reactions", []) or []: + if str(getattr(reaction, "emoji", "")) not in emojis: + continue + if getattr(reaction, "me", False): + return True + users = getattr(reaction, "users", None) + if not callable(users) or bot_id is None: + continue + try: + async for user in users(): + if getattr(user, "id", None) == bot_id: + return True + except Exception: + continue + return False + + def _is_down_notice_content(self, content: str) -> bool: + 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) + + async def _message_has_non_down_bot_response(self, message: Any) -> bool: + """Detect an already-addressed message without trusting down notices.""" + bot_user = getattr(self._client, "user", None) if self._client else None + bot_id = getattr(bot_user, "id", None) + if bot_id is None: + return False + + async def _scan_history(channel: Any) -> bool: + history = getattr(channel, "history", None) + if not callable(history): + return False + try: + async for candidate in history(limit=25, after=getattr(message, "created_at", None), oldest_first=True): + author = getattr(candidate, "author", None) + if getattr(author, "id", None) != bot_id: + continue + if self._is_down_notice_content(getattr(candidate, "content", "")): + continue + reference = getattr(candidate, "reference", None) + ref_id = str(getattr(reference, "message_id", "") or "") + if not ref_id or ref_id == str(getattr(message, "id", "")): + return True + except Exception: + return False + return False + + if await _scan_history(getattr(message, "channel", None)): + return True + + thread = getattr(message, "thread", None) + if thread is not None and await _scan_history(thread): + return True + return False + + def _discord_recovery_db_path(self) -> _Path: + from hermes_constants import get_hermes_home + + directory = get_hermes_home() / "gateway" + directory.mkdir(parents=True, exist_ok=True) + return directory / _DISCORD_RECOVERY_DB_FILENAME + + def _with_discord_recovery_db(self, fn, default=None): + try: + with self._discord_recovery_db_lock: + conn = sqlite3.connect(self._discord_recovery_db_path()) + try: + 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, + action_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 + ) + """) + result = fn(conn) + conn.commit() + return result + finally: + conn.close() + except Exception as exc: + logger.debug("[%s] Discord recovery DB operation failed: %s", self.name, exc, exc_info=True) + return default + + @staticmethod + def _utc_now_iso() -> str: + import datetime as _dt + return _dt.datetime.now(_dt.timezone.utc).isoformat() + + def _message_channel_ids(self, message: Any) -> tuple[str, Optional[str], Optional[str]]: + channel = getattr(message, "channel", None) + channel_id = str(getattr(channel, "id", "") or "") + parent_id = str(getattr(channel, "parent_id", "") or "") or None + thread_id = channel_id if parent_id else None + return channel_id, thread_id, parent_id + + def _record_discord_message_seen(self, message: Any, *, status: str) -> None: + message_id = str(getattr(message, "id", "") or "") + if not message_id: + return + channel_id, thread_id, parent_id = self._message_channel_ids(message) + author_id = str(getattr(getattr(message, "author", None), "id", "") or "") + created_at = getattr(message, "created_at", None) + created_text = created_at.isoformat() if hasattr(created_at, "isoformat") else None + now = self._utc_now_iso() + + def _op(conn): + existing = conn.execute("SELECT status FROM discord_messages WHERE message_id=?", (message_id,)).fetchone() + final_status = existing[0] if existing and existing[0] == "responded" else status + conn.execute( + """ + INSERT INTO discord_messages (message_id, channel_id, thread_id, parent_channel_id, author_id, created_at, status, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(message_id) DO UPDATE SET + channel_id=excluded.channel_id, + thread_id=excluded.thread_id, + parent_channel_id=excluded.parent_channel_id, + author_id=excluded.author_id, + created_at=COALESCE(discord_messages.created_at, excluded.created_at), + status=?, + updated_at=excluded.updated_at + """, + (message_id, channel_id, thread_id, parent_id, author_id, created_text, final_status, now, final_status), + ) + + self._with_discord_recovery_db(_op) + + def _record_recovery_attempt(self, message: Any, *, status: str, error: Optional[str] = None) -> None: + self._record_discord_message_seen(message, status=status) + message_id = str(getattr(message, "id", "") or "") + if not message_id: + return + now = self._utc_now_iso() + + def _op(conn): + conn.execute( + """ + UPDATE discord_messages + SET status=?, attempts=attempts+1, last_attempt_at=?, last_error=?, updated_at=? + WHERE message_id=? + """, + (status, now, error, now, message_id), + ) + + self._with_discord_recovery_db(_op) + + def _record_discord_processing_start(self, event: MessageEvent, *, emoji_ack: bool) -> None: + message = event.raw_message + self._record_discord_message_seen(message, status="processing") + message_id = str(getattr(message, "id", "") or getattr(event, "message_id", "") or "") + if not message_id: + return + now = self._utc_now_iso() + + def _op(conn): + conn.execute( + "UPDATE discord_messages SET status='processing', emoji_ack=?, updated_at=? WHERE message_id=?", + (1 if emoji_ack else 0, now, message_id), + ) + + self._with_discord_recovery_db(_op) + + def _record_discord_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None: + 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") + 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), + ) + + self._with_discord_recovery_db(_op) + + def _record_discord_response(self, *, reply_to: Optional[str], result: SendResult, content: str) -> None: + if not reply_to: + return + now = self._utc_now_iso() + outage = self._is_down_notice_content(content) + status = "missed" if outage else ("responded" if result.success else "failed") + + def _op(conn): + conn.execute( + """ + INSERT INTO discord_messages (message_id, status, replied, outage_response, response_message_id, updated_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(message_id) DO UPDATE SET + status=?, + replied=CASE WHEN ? THEN 1 ELSE replied END, + outage_response=CASE WHEN ? THEN 1 ELSE outage_response END, + response_message_id=COALESCE(?, response_message_id), + updated_at=? + """, + (reply_to, status, 1 if result.success and not outage else 0, 1 if outage else 0, result.message_id, now, status, 1 if result.success and not outage else 0, 1 if outage else 0, result.message_id, now), + ) + + self._with_discord_recovery_db(_op) + + def _discord_message_is_persistently_complete(self, message_id: str) -> bool: + if not message_id: + return False + + def _op(conn): + row = conn.execute("SELECT status, replied, outage_response FROM discord_messages WHERE message_id=?", (message_id,)).fetchone() + if not row: + return False + status, replied, outage = row + return status == "responded" and bool(replied) and not bool(outage) + + return bool(self._with_discord_recovery_db(_op, default=False)) + + def _record_recovery_scan_start(self, channels: set[str]) -> str: + scan_id = f"{int(time.time() * 1000)}-{os.getpid()}" + now = self._utc_now_iso() + + def _op(conn): + conn.execute( + "INSERT OR REPLACE INTO discord_recovery_scans (scan_id, started_at, status, channels, window_seconds, limit_count) VALUES (?, ?, ?, ?, ?, ?)", + (scan_id, now, "running", json.dumps(sorted(channels)), self._missed_message_backfill_window_seconds(), self._missed_message_backfill_limit()), + ) + + self._with_discord_recovery_db(_op) + return scan_id + + def _record_recovery_scan_complete(self, scan_id: str, *, status: str, scanned: int, missed: int, dispatched: int, error: Optional[str] = None) -> None: + now = self._utc_now_iso() + + def _op(conn): + conn.execute( + "UPDATE discord_recovery_scans SET completed_at=?, status=?, scanned=?, missed=?, dispatched=?, error=? WHERE scan_id=?", + (now, status, scanned, missed, dispatched, error, scan_id), + ) + + self._with_discord_recovery_db(_op) + def _get_discord_command_sync_policy(self) -> str: raw = str(os.getenv("DISCORD_COMMAND_SYNC_POLICY", "safe") or "").strip().lower() if raw in _DISCORD_COMMAND_SYNC_POLICIES: @@ -2131,15 +2619,16 @@ class DiscordAdapter(BasePlatformAdapter): return os.getenv("DISCORD_REACTIONS", "true").lower() not in {"false", "0", "no"} async def on_processing_start(self, event: MessageEvent) -> None: - """Add an in-progress reaction for normal Discord message events.""" - if not self._reactions_enabled(): - return + """Add an in-progress reaction and record durable handling state.""" message = event.raw_message - if hasattr(message, "add_reaction"): - await self._add_reaction(message, "👀") + acked = False + if self._reactions_enabled() and hasattr(message, "add_reaction"): + acked = await self._add_reaction(message, "👀") + self._record_discord_processing_start(event, emoji_ack=acked) async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None: - """Swap the in-progress reaction for a final success/failure reaction.""" + """Swap the in-progress reaction for final reaction and durable state.""" + self._record_discord_processing_complete(event, outcome) if not self._reactions_enabled(): return message = event.raw_message @@ -2256,15 +2745,19 @@ class DiscordAdapter(BasePlatformAdapter): elif not _looks_like_nonconversational_history_message(content): self._last_self_message_id[_target_id] = message_ids[-1] - return SendResult( + result = SendResult( success=True, message_id=message_ids[0] if message_ids else None, raw_response={"message_ids": message_ids} ) + self._record_discord_response(reply_to=reply_to, result=result, content=content) + return result except Exception as e: # pragma: no cover - defensive logging logger.error("[%s] Failed to send Discord message: %s", self.name, e, exc_info=True) - return SendResult(success=False, error=str(e)) + result = SendResult(success=False, error=str(e)) + self._record_discord_response(reply_to=reply_to, result=result, content=content) + return result async def _send_to_forum(self, forum_channel: Any, content: str) -> SendResult: """Create a thread post in a forum channel with the message as starter content. @@ -8633,6 +9126,22 @@ def _apply_yaml_config(yaml_cfg: dict, discord_cfg: dict) -> dict | None: os.environ["DISCORD_AUTO_THREAD"] = str(discord_cfg["auto_thread"]).lower() if "reactions" in discord_cfg and not os.getenv("DISCORD_REACTIONS"): os.environ["DISCORD_REACTIONS"] = str(discord_cfg["reactions"]).lower() + backfill_cfg = discord_cfg.get("missed_message_backfill") + if isinstance(backfill_cfg, dict): + _backfill_env = { + "enabled": "DISCORD_MISSED_MESSAGE_BACKFILL", + "channels": "DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", + "window_seconds": "DISCORD_MISSED_MESSAGE_BACKFILL_WINDOW_SECONDS", + "limit": "DISCORD_MISSED_MESSAGE_BACKFILL_LIMIT", + "max_dispatches": "DISCORD_MISSED_MESSAGE_BACKFILL_MAX_DISPATCHES", + } + for _key, _env in _backfill_env.items(): + if _key not in backfill_cfg or os.getenv(_env): + continue + _value = backfill_cfg[_key] + if isinstance(_value, list): + _value = ",".join(str(v) for v in _value) + os.environ[_env] = str(_value).lower() if isinstance(_value, bool) else str(_value) # ignored_channels: channels where bot never responds (even when mentioned) ic = discord_cfg.get("ignored_channels") if ic is not None and not os.getenv("DISCORD_IGNORED_CHANNELS"): diff --git a/tests/gateway/test_discord_missed_message_backfill.py b/tests/gateway/test_discord_missed_message_backfill.py new file mode 100644 index 00000000000..408ddc84a8c --- /dev/null +++ b/tests/gateway/test_discord_missed_message_backfill.py @@ -0,0 +1,241 @@ +"""Tests for Discord missed-message startup backfill.""" + +import os +import sys +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import PlatformConfig + + +def _ensure_discord_mock(): + if "discord" in sys.modules and hasattr(sys.modules["discord"], "__file__"): + return + + discord_mod = MagicMock() + discord_mod.Intents.default.return_value = MagicMock() + discord_mod.Client = MagicMock + discord_mod.File = MagicMock + discord_mod.DMChannel = type("DMChannel", (), {}) + discord_mod.Thread = type("Thread", (), {}) + discord_mod.ForumChannel = type("ForumChannel", (), {}) + discord_mod.ui = SimpleNamespace(View=object, button=lambda *a, **k: (lambda fn: fn), Button=object) + discord_mod.ButtonStyle = SimpleNamespace(success=1, primary=2, secondary=2, danger=3, green=1, grey=2, blurple=2, red=3) + 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.app_commands = SimpleNamespace( + describe=lambda **kwargs: (lambda fn: fn), + choices=lambda **kwargs: (lambda fn: fn), + Choice=lambda **kwargs: SimpleNamespace(**kwargs), + ) + + ext_mod = MagicMock() + commands_mod = MagicMock() + commands_mod.Bot = MagicMock + ext_mod.commands = commands_mod + + sys.modules.setdefault("discord", discord_mod) + sys.modules.setdefault("discord.ext", ext_mod) + sys.modules.setdefault("discord.ext.commands", commands_mod) + + +_ensure_discord_mock() + +from gateway.platforms.discord import DiscordAdapter # noqa: E402 + + +class FakeReaction: + def __init__(self, emoji, *, me=False, users=None): + self.emoji = emoji + self.me = me + self._users = list(users or []) + + async def users(self): + for user in self._users: + yield user + + +class FakeChannel: + def __init__(self, channel_id=123, history_messages=None): + self.id = channel_id + self.name = "wiki-inbox" + self.guild = SimpleNamespace(name="emo") + self.topic = None + self._history_messages = list(history_messages or []) + + def history(self, **kwargs): + async def _gen(): + for message in self._history_messages: + yield message + + return _gen() + + +@pytest.fixture +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) + adapter._handle_message = AsyncMock() + monkeypatch.setenv("DISCORD_MISSED_MESSAGE_BACKFILL", "true") + return adapter + + +def make_message(*, message_id=1, author_id=42, content="please ingest", reactions=None, channel=None): + channel = channel or FakeChannel() + return SimpleNamespace( + id=message_id, + content=content, + reactions=list(reactions or []), + author=SimpleNamespace(id=author_id, bot=False, display_name="Emo", name="emo"), + channel=channel, + created_at=datetime.now(timezone.utc), + attachments=[], + mentions=[], + reference=None, + type=None, + ) + + +@pytest.mark.asyncio +async def test_backfills_message_with_only_own_success_reaction(adapter): + message = make_message(reactions=[FakeReaction("✅", me=True)]) + + assert await adapter._should_backfill_discord_message(message) is True + + +@pytest.mark.asyncio +async def test_should_not_backfill_message_with_non_down_bot_response(adapter): + bot_reply = SimpleNamespace( + id=2, + content="Done — captured it.", + 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_backfills_when_only_down_notice_exists(adapter): + down_notice = SimpleNamespace( + id=2, + content="The agent is down right now.", + author=SimpleNamespace(id=999, bot=True), + reference=SimpleNamespace(message_id=1), + created_at=datetime.now(timezone.utc), + ) + channel = FakeChannel(history_messages=[down_notice]) + message = make_message(message_id=1, channel=channel) + + assert await adapter._should_backfill_discord_message(message) is True + + +@pytest.mark.asyncio +async def test_run_backfill_dispatches_unaddressed_messages(adapter, monkeypatch): + message = make_message(message_id=1) + + async def fake_candidates(_channels): + yield message + + monkeypatch.setenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", "123") + monkeypatch.setattr(adapter, "_iter_missed_message_backfill_candidates", fake_candidates) + monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True)) + monkeypatch.setattr(adapter, "_missed_message_backfill_max_dispatches", lambda: 10) + monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"}) + monkeypatch.setattr("asyncio.sleep", AsyncMock()) + + await adapter._run_missed_message_backfill() + + adapter._handle_message.assert_awaited_once_with(message) + + +def test_missed_message_backfill_config_bridge(monkeypatch, tmp_path): + from gateway.config import load_gateway_config + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + for key in ( + "DISCORD_MISSED_MESSAGE_BACKFILL", + "DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", + "DISCORD_MISSED_MESSAGE_BACKFILL_WINDOW_SECONDS", + "DISCORD_MISSED_MESSAGE_BACKFILL_LIMIT", + "DISCORD_MISSED_MESSAGE_BACKFILL_MAX_DISPATCHES", + ): + monkeypatch.delenv(key, raising=False) + + (tmp_path / "config.yaml").write_text( + "platforms:\n" + " discord:\n" + " enabled: true\n" + "discord:\n" + " missed_message_backfill:\n" + " enabled: true\n" + " channels: ['1501971993405292796']\n" + " window_seconds: 3600\n" + " limit: 25\n" + " max_dispatches: 3\n" + ) + + load_gateway_config() + + assert os.environ["DISCORD_MISSED_MESSAGE_BACKFILL"] == "true" + assert os.environ["DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS"] == "1501971993405292796" + assert os.environ["DISCORD_MISSED_MESSAGE_BACKFILL_WINDOW_SECONDS"] == "3600" + assert os.environ["DISCORD_MISSED_MESSAGE_BACKFILL_LIMIT"] == "25" + assert os.environ["DISCORD_MISSED_MESSAGE_BACKFILL_MAX_DISPATCHES"] == "3" + + +@pytest.mark.asyncio +async def test_persistent_responded_record_suppresses_backfill(adapter): + message = make_message(message_id=77) + adapter._record_discord_message_seen(message, status="responded") + adapter._record_discord_response( + reply_to="77", + result=SimpleNamespace(success=True, message_id="9001"), + content="Done — captured it.", + ) + + assert await adapter._should_backfill_discord_message(message) is False + + +def test_down_notice_response_does_not_mark_message_complete(adapter): + adapter._record_discord_response( + reply_to="88", + result=SimpleNamespace(success=True, message_id="9002"), + content="The agent is down right now.", + ) + + assert adapter._discord_message_is_persistently_complete("88") 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)) + archived_msg = make_message(message_id=202, channel=FakeChannel(channel_id=2020)) + active_thread = FakeChannel(channel_id=2010, history_messages=[active_msg]) + archived_thread = FakeChannel(channel_id=2020, history_messages=[archived_msg]) + + class ParentChannel(FakeChannel): + threads = [active_thread] + + def archived_threads(self, **kwargs): + async def _gen(): + yield archived_thread + return _gen() + + parent = ParentChannel(channel_id=123, history_messages=[]) + adapter._client.get_channel = lambda _id: parent + + got = [] + async for msg in adapter._iter_missed_message_backfill_candidates({"123"}): + got.append(msg.id) + + assert got == [201, 202]