diff --git a/contributors/emails/emodoteth@gmail.com b/contributors/emails/emodoteth@gmail.com new file mode 100644 index 000000000000..a227d61e64d2 --- /dev/null +++ b/contributors/emails/emodoteth@gmail.com @@ -0,0 +1 @@ +emo-eth diff --git a/contributors/emails/wenzel.james.r@gmail.com b/contributors/emails/wenzel.james.r@gmail.com new file mode 100644 index 000000000000..a227d61e64d2 --- /dev/null +++ b/contributors/emails/wenzel.james.r@gmail.com @@ -0,0 +1 @@ +emo-eth diff --git a/gateway/platforms/helpers.py b/gateway/platforms/helpers.py index 7af36280cf4a..85edb02c5140 100644 --- a/gateway/platforms/helpers.py +++ b/gateway/platforms/helpers.py @@ -70,6 +70,22 @@ def is_duplicate(self, msg_id: str) -> bool: 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 discard(self, msg_id: str) -> None: + """Release a claimed message ID after cancelled/failed handoff.""" + self._seen.pop(msg_id, None) + def clear(self): """Clear all tracked messages.""" self._seen.clear() diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index fca8bf43847f..f1aaa393829c 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -240,6 +240,8 @@ def _metadata_for_send( final-message delivery. """ meta = dict(self.metadata) if self.metadata else {} + if self._initial_reply_to_id: + meta["reply_to_message_id"] = self._initial_reply_to_id if expect_edits: meta["expect_edits"] = True if final: diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index 02a96ee3dcfd..2a25217160c0 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -774,6 +774,7 @@ def run_import(args) -> None: "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 diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 55ad02afd3ee..fc9d75470aeb 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2532,6 +2532,13 @@ def _ensure_hermes_home_managed(home: Path): "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 diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index b373ef4b8872..215ae0d46d18 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -10,6 +10,7 @@ """ import asyncio +import datetime as dt import hashlib import inspect import json @@ -50,6 +51,7 @@ def __init__(self, id: int) -> None: # noqa: A002 - matches discord API _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_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,10 @@ def __init__(self, config: PlatformConfig): # 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 + from hermes_constants import get_hermes_home + from plugins.platforms.discord.recovery import DiscordRecoveryStore + self._discord_recovery_store = DiscordRecoveryStore(get_hermes_home()) # Dedup cache: prevents duplicate bot responses when Discord # RESUME replays events after reconnects. self._dedup = MessageDeduplicator() @@ -1160,116 +1166,12 @@ async def on_ready(): adapter_self._post_connect_task = asyncio.create_task( adapter_self._run_post_connect_initialization() ) + if adapter_self._missed_message_backfill_enabled(): + adapter_self._ensure_missed_message_backfill_task() @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): @@ -1344,6 +1246,96 @@ async def on_voice_state_update(member, before, after): self._release_platform_lock() return False + def _discord_message_admission( + self, + message: Any, + *, + claim: bool, + ) -> tuple[bool, bool]: + """Return ``(admitted, role_authorized)`` for one Discord event.""" + message_id = str(getattr(message, "id", "")) + if claim: + if self._dedup.is_duplicate(message_id): + return False, False + elif self._dedup.contains(message_id): + return False, False + if message.author == self._client.user: + return False, False + if message.type not in {discord.MessageType.default, discord.MessageType.reply}: + return False, 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, False + if allow_bots == "mentions" and not self._self_is_explicitly_mentioned(message): + return False, False + if ( + self._discord_bots_require_inline_mention() + and not self._self_is_raw_mentioned(message) + ): + return False, 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, 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, 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, False + + return True, role_authorized + + async def _dispatch_discord_message(self, message: Any) -> bool: + """Apply Discord ingress policy and dispatch one live event.""" + if not self._ready_event.is_set(): + try: + await asyncio.wait_for(self._ready_event.wait(), timeout=30.0) + except asyncio.TimeoutError: + pass + admitted, role_authorized = self._discord_message_admission( + message, claim=True, + ) + if not admitted: + return False + return await self._handle_message( + message, role_authorized=role_authorized, + ) + 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(): @@ -1636,12 +1628,19 @@ async def disconnect(self) -> None: 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 +1908,661 @@ async def _run_post_connect_initialization(self) -> None: 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.""" + configured = self.config.extra.get("missed_message_backfill") + if isinstance(configured, dict) and "enabled" in configured: + value = configured["enabled"] + if isinstance(value, str): + return value.strip().lower() in ("true", "1", "yes", "on") + return bool(value) + 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 the union of allowed and free-response channels so both + mention-gated requests and mention-free work can be recovered. + Operators can set ``channels: "*"`` to scan every reachable text + channel, but the safe default is scoped. + """ + configured = self.config.extra.get("missed_message_backfill") + if isinstance(configured, dict) and "channels" in configured: + raw = configured.get("channels") + if isinstance(raw, list): + return {str(item).strip() for item in raw if str(item).strip()} + raw = str(raw or "") + if raw.strip(): + return {item.strip() for item in raw.split(",") if item.strip()} + raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", "") + if not raw.strip(): + 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: + configured = self.config.extra.get("missed_message_backfill") + raw = ( + configured.get("window_seconds", 21600) + if isinstance(configured, dict) + else 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: + configured = self.config.extra.get("missed_message_backfill") + raw = ( + configured.get("limit", 100) + if isinstance(configured, dict) + else 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: + configured = self.config.extra.get("missed_message_backfill") + raw = ( + configured.get("max_dispatches", 10) + if isinstance(configured, dict) + else os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_MAX_DISPATCHES", "10") + ) + try: + value = int(raw) + except (TypeError, ValueError): + value = 10 + return max(1, min(value, 100)) + + def _ensure_missed_message_backfill_task(self) -> asyncio.Task: + """Return the active recovery task, or start one when none is running.""" + task = self._missed_message_backfill_task + if task is not None and not task.done(): + return task + task = asyncio.create_task(self._run_missed_message_backfill()) + self._missed_message_backfill_task = task + runner = getattr(self, "gateway_runner", None) + if runner is not None and getattr(runner, "_startup_restore_in_progress", False): + tasks = getattr(runner, "_startup_restore_tasks", None) + if tasks is None: + tasks = [] + runner._startup_restore_tasks = tasks + tasks.append(task) + return task + + 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() + ledger_ok = await self._with_discord_recovery_db_async( + lambda conn: conn.execute("SELECT 1").fetchone() is not None, + False, + ) + if not ledger_ok: + logger.error( + "[%s] Missed-message recovery aborted: durable ledger unavailable", + self.name, + ) + return + scan_id = await asyncio.to_thread( + self._record_recovery_scan_start, + channels, + ) + if not channels: + logger.info("[%s] Missed-message backfill enabled but no channels configured", self.name) + await asyncio.to_thread( + 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") + # 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 + 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: + admitted = await self._dispatch_recovered_message(message) + if admitted: + dispatched += 1 + except asyncio.CancelledError: + self._dedup.discard(message_id) + self._record_recovery_attempt(message, status="cancelled") + raise + except Exception as exc: + self._dedup.discard(message_id) + self._record_recovery_attempt(message, status="failed", error=str(exc)) + raise + if dispatched >= max_dispatches: + break + await asyncio.to_thread( + 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: + await asyncio.to_thread( + 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 + await asyncio.to_thread( + 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.""" + if not isinstance(message.channel, discord.DMChannel): + parent_id = self._get_parent_channel_id(message.channel) + channel_keys = self._discord_channel_keys(message, parent_id) + free_channels = self._discord_free_response_channels() + in_bot_thread = ( + isinstance(message.channel, discord.Thread) + and str(message.channel.id) in self._threads + and not self._discord_thread_require_mention() + ) + if ( + self._discord_require_mention() + and "*" not in free_channels + and not (channel_keys & free_channels) + and not in_bot_thread + and not self._self_is_explicitly_mentioned(message) + ): + return False + admitted, role_authorized = self._discord_message_admission( + message, claim=False, + ) + if not admitted: + return False + return await self._handle_message( + message, + role_authorized=role_authorized, + recovered=True, + ) + + async def _iter_missed_message_backfill_candidates(self, channel_ids: set[str]): + if not self._client: + return + 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) + + iterators = [ + self._iter_channel_and_thread_messages( + channel, + limit=limit, + after=after, + seen_channels=seen, + ).__aiter__() + for channel in candidate_channels + ] + yielded = 0 + while iterators and yielded < limit: + next_round = [] + for iterator in iterators: + try: + item = await iterator.__anext__() + except StopAsyncIteration: + continue + yield item + yielded += 1 + next_round.append(iterator) + if yielded >= limit: + return + iterators = next_round + + 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) + + 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: + # 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_iter: # type: ignore[attr-defined] + messages.append(message) + 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) + + 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 + + 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): + return False + if getattr(getattr(message, "author", None), "id", None) == getattr(self._client.user, "id", None): + return False + if self._discord_message_is_persistently_complete(str(getattr(message, "id", ""))): + return False + if self._discord_message_has_active_claim(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 + + def _is_down_notice_content(self, content: str) -> bool: + """Recognize only explicit Hermes/gateway outage notices.""" + text = (content or "").lower() + 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.""" + 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 ref_id == str(getattr(message, "id", "")): + return True + except Exception: + return False + return False + + message_channel = getattr(message, "channel", None) + # Only an explicit reply reference proves which input a bot response + # completed. An arbitrary later bot post can otherwise mask multiple + # unanswered requests in the same parent channel or thread. + if await _scan_history(message_channel): + 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: + return self._discord_recovery_store.path() + + def _with_discord_recovery_db(self, fn, default=None): + return self._discord_recovery_store.call(fn, default) + + async def _with_discord_recovery_db_async(self, fn, default=None): + return await asyncio.to_thread( + self._discord_recovery_store.call, + fn, + 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: + if not self._missed_message_backfill_enabled(): + return + 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: + 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: + 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: + 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 "") + 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: + 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 + 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=CASE WHEN status='responded' THEN status ELSE ? END, " + "updated_at=? WHERE message_id=?", + (status, now, message_id), + ) + + self._with_discord_recovery_db(_op) + + def _record_discord_response( + self, + *, + reply_to: Optional[str], + result: SendResult, + content: str, + final: bool, + ) -> None: + if not self._missed_message_backfill_enabled() or not reply_to: + return + now = self._utc_now_iso() + completed = bool(final and result.success) + status = "responded" if completed else "failed" + + def _op(conn): + conn.execute( + """ + INSERT INTO discord_messages (message_id, status, replied, outage_response, response_message_id, updated_at) + VALUES (?, ?, ?, 0, ?, ?) + ON CONFLICT(message_id) DO UPDATE SET + status=CASE WHEN ? THEN 'responded' ELSE discord_messages.status END, + replied=CASE WHEN ? THEN 1 ELSE discord_messages.replied END, + outage_response=CASE WHEN ? THEN 0 ELSE discord_messages.outage_response END, + response_message_id=COALESCE(?, response_message_id), + updated_at=? + """, + ( + reply_to, + status, + 1 if completed else 0, + result.message_id, + now, + 1 if completed else 0, + 1 if completed else 0, + 1 if completed else 0, + result.message_id, + now, + ), + ) + + self._with_discord_recovery_db(_op) + if completed: + def _channel_for_message(conn): + row = conn.execute( + "SELECT COALESCE(thread_id, channel_id) FROM discord_messages " + "WHERE message_id=?", + (reply_to,), + ).fetchone() + return str(row[0]) if row and row[0] else None + + channel_id = self._with_discord_recovery_db(_channel_for_message) + if channel_id: + self._advance_discord_recovery_cursor(channel_id, reply_to) + + 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 _discord_message_has_active_claim(self, message_id: str) -> bool: + if not message_id: + return False + cutoff = ( + dt.datetime.now(dt.timezone.utc) - dt.timedelta(minutes=10) + ).isoformat() + + def _op(conn): + row = conn.execute( + "SELECT status, updated_at FROM discord_messages WHERE message_id=?", + (message_id,), + ).fetchone() + return bool( + row + and row[0] in {"queued", "processing"} + and row[1] >= cutoff + ) + + return bool(self._with_discord_recovery_db(_op, default=True)) + + 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 +2785,24 @@ def _reactions_enabled(self) -> bool: 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, "👀") + await asyncio.to_thread( + 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.""" + await asyncio.to_thread( + self._record_discord_processing_complete, + event, + outcome, + ) if not self._reactions_enabled(): return message = event.raw_message @@ -2174,6 +2837,7 @@ async def send( if metadata and metadata.get("thread_id"): thread_id = metadata["thread_id"] nonconversational = _metadata_marks_nonconversational(metadata) + final_delivery = bool(metadata and metadata.get("notify")) if thread_id: # Fetch the thread directly — threads are addressed by their own ID. @@ -2192,7 +2856,15 @@ async def send( # 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) + await asyncio.to_thread( + self._record_discord_response, + reply_to=reply_to, + result=result, + content=content, + final=final_delivery, + ) + return result # Format and split message if needed formatted = self.format_message(content) @@ -2256,15 +2928,31 @@ async def send( 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} ) + await asyncio.to_thread( + self._record_discord_response, + reply_to=reply_to, + result=result, + content=content, + final=final_delivery, + ) + 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)) + await asyncio.to_thread( + self._record_discord_response, + reply_to=reply_to, + result=result, + content=content, + final=bool(metadata and metadata.get("notify")), + ) + 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. @@ -2389,6 +3077,7 @@ async def edit_message( content: str, *, finalize: bool = False, + metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Edit a previously sent Discord message. @@ -2468,7 +3157,16 @@ async def edit_message( self._last_overflow_preview[_preview_key] = truncated else: raise - return SendResult(success=True, message_id=message_id) + result = SendResult(success=True, message_id=message_id) + if finalize: + await asyncio.to_thread( + self._record_discord_response, + reply_to=(metadata or {}).get("reply_to_message_id"), + result=result, + content=content, + final=True, + ) + return result except Exception as e: # pragma: no cover - defensive logging logger.error("[%s] Failed to edit Discord message %s: %s", self.name, message_id, e, exc_info=True) return SendResult(success=False, error=str(e)) @@ -6356,8 +7054,14 @@ async def _cache_discord_document(self, att, ext: str) -> bytes: raise Exception(f"HTTP {resp.status}") return await resp.read() - async def _handle_message(self, message: DiscordMessage, role_authorized: bool = False) -> None: - """Handle incoming Discord messages.""" + async def _handle_message( + self, + message: DiscordMessage, + role_authorized: bool = False, + *, + recovered: bool = False, + ) -> bool: + """Handle one Discord message and report whether it reached dispatch.""" # In server channels (not DMs), require the bot to be @mentioned # UNLESS the channel is in the free-response list or the message is # in a thread where the bot has already participated. @@ -6413,14 +7117,14 @@ async def _handle_message(self, message: DiscordMessage, role_authorized: bool = allowed_channels = {ch.strip() for ch in allowed_channels_raw.split(",") if ch.strip()} if "*" not in allowed_channels and not (channel_keys & allowed_channels): logger.debug("[%s] Ignoring message in non-allowed channel: %s", self.name, channel_keys) - return + return False # Check ignored channels - never respond even when mentioned ignored_channels_raw = os.getenv("DISCORD_IGNORED_CHANNELS", "") ignored_channels = {ch.strip() for ch in ignored_channels_raw.split(",") if ch.strip()} if "*" in ignored_channels or (channel_keys & ignored_channels): logger.debug("[%s] Ignoring message in ignored channel: %s", self.name, channel_keys) - return + return False free_channels = self._discord_free_response_channels() @@ -6449,7 +7153,7 @@ async def _handle_message(self, message: DiscordMessage, role_authorized: bool = if require_mention and not is_free_channel and not in_bot_thread: if not self._self_is_explicitly_mentioned(message) and not mention_prefix: - return + return False # Auto-thread: when enabled, automatically create a thread for every # @mention in a text channel so each conversation is isolated (like Slack). # Messages already inside threads or DMs are unaffected. @@ -6498,7 +7202,7 @@ async def _handle_message(self, message: DiscordMessage, role_authorized: bool = self.name, notify_error, ) - return + return False referenced_attachments = [] reference = getattr(message, "reference", None) @@ -6795,7 +7499,7 @@ async def _handle_message(self, message: DiscordMessage, role_authorized: bool = getattr(message.author, "display_name", getattr(message.author, "name", "unknown")), getattr(message.channel, "id", "unknown"), ) - return + return False event_text = "(The user sent a message with no text content)" _chan = message.channel @@ -6832,12 +7536,18 @@ async def _handle_message(self, message: DiscordMessage, role_authorized: bool = if thread_id: self._threads.mark(thread_id) - # Only batch plain text messages — commands, media, etc. dispatch - # immediately since they won't be split by the Discord client. - if msg_type == MessageType.TEXT and self._text_batch_delay_seconds > 0: + # Only live plain text messages use split-message batching. Recovery + # candidates are already complete historical messages; coalescing them + # would lose constituent IDs and make later restarts replay them. + if ( + not recovered + and msg_type == MessageType.TEXT + and self._text_batch_delay_seconds > 0 + ): self._enqueue_text_event(event) else: await self.handle_message(event) + return True # ------------------------------------------------------------------ # Text message aggregation (handles Discord client-side splits) @@ -8633,6 +9343,10 @@ 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() + seeded_extra = {} + backfill_cfg = discord_cfg.get("missed_message_backfill") + if isinstance(backfill_cfg, dict): + seeded_extra["missed_message_backfill"] = dict(backfill_cfg) # 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"): @@ -8709,16 +9423,15 @@ def _apply_yaml_config(yaml_cfg: dict, discord_cfg: dict) -> dict | None: ("websocket_heartbeat_ack_max_age_seconds", None, None), ("websocket_max_latency_seconds", None, None), ) - seeded = {} for primary_key, legacy_key, env_key in _websocket_liveness_keys: value = _websocket_liveness_cfg.get(primary_key) if value is None and legacy_key: value = _websocket_liveness_cfg.get(legacy_key) if value is not None: - seeded[primary_key] = value + seeded_extra[primary_key] = value if env_key and not os.getenv(env_key): os.environ[env_key] = str(value) - return seeded or None + return seeded_extra or None def _is_connected(config) -> bool: diff --git a/plugins/platforms/discord/recovery.py b/plugins/platforms/discord/recovery.py new file mode 100644 index 000000000000..060197e481aa --- /dev/null +++ b/plugins/platforms/discord/recovery.py @@ -0,0 +1,110 @@ +"""Durable state for Discord reconnect message recovery.""" + +from __future__ import annotations + +import datetime as dt +import logging +import os +import sqlite3 +import threading +from contextlib import suppress +from pathlib import Path +from typing import Any, Callable + +from hermes_constants import get_hermes_home + +logger = logging.getLogger(__name__) + +_DB_FILENAME = "discord_message_recovery.db" +_RETENTION_DAYS = 30 + + +class DiscordRecoveryStore: + """Small profile-scoped SQLite ledger for completed Discord messages.""" + + def __init__(self, hermes_home: Path | None = None) -> None: + self._lock = threading.Lock() + self._initialized = False + self._hermes_home = Path(hermes_home or get_hermes_home()) + + def path(self) -> Path: + directory = self._hermes_home / "gateway" + directory.mkdir(parents=True, exist_ok=True) + return directory / _DB_FILENAME + + def call(self, fn: Callable[[sqlite3.Connection], Any], default: Any = None) -> Any: + try: + with self._lock: + path = self.path() + conn = sqlite3.connect(path, timeout=0.1) + try: + if not self._initialized: + self._initialize(conn) + self._initialized = True + with suppress(OSError): + os.chmod(path, 0o600) + result = fn(conn) + conn.commit() + return result + finally: + conn.close() + except Exception as exc: + logger.warning("Discord recovery ledger unavailable: %s", exc) + return default + + def _initialize(self, conn: sqlite3.Connection) -> None: + 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, + 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 + ) + """) + 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() + conn.execute("DELETE FROM discord_messages WHERE updated_at < ?", (cutoff,)) + conn.execute( + "DELETE FROM discord_recovery_scans " + "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 new file mode 100644 index 000000000000..d33e5e38755a --- /dev/null +++ b/tests/gateway/test_discord_missed_message_backfill.py @@ -0,0 +1,990 @@ +"""Tests for Discord missed-message startup backfill.""" + +import asyncio +import datetime as dt +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 Platform, PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType, ProcessingOutcome + + +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.Object = lambda *, id: SimpleNamespace(id=id) + 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() + +import discord # noqa: E402 +from plugins.platforms.discord.adapter import ( # noqa: E402 + DiscordAdapter, + _apply_yaml_config, +) + + +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, parent_id=None): + self.id = channel_id + self.parent_id = parent_id + self.name = "wiki-inbox" + self.guild = SimpleNamespace(id=777, 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) + 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(return_value=True) + 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, mentions=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, + guild=getattr(channel, "guild", None), + created_at=datetime.now(timezone.utc), + attachments=[], + mentions=list(mentions or []), + reference=None, + type=discord.MessageType.default, + ) + + +def make_bot_message(*, message_id=1, content="please ingest", channel=None, mentions=None): + message = make_message( + message_id=message_id, + content=content, + channel=channel, + mentions=mentions, + ) + message.author.bot = True + return message + + +@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_configured_bot_sender_is_left_for_shared_ingress_policy(adapter, monkeypatch): + bot_user = adapter._client.user + monkeypatch.setenv("DISCORD_ALLOW_BOTS", "mentions") + message = make_bot_message( + message_id=98, + content=f"<@{bot_user.id}> run this", + mentions=[bot_user], + ) + + 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_parent_channel_unreferenced_bot_message_does_not_suppress_backfill(adapter): + unrelated_bot_post = SimpleNamespace( + id=2, + content="Done — captured a different item.", + author=SimpleNamespace(id=999, bot=True), + reference=None, + created_at=datetime.now(timezone.utc), + ) + channel = FakeChannel(history_messages=[unrelated_bot_post]) + message = make_message(message_id=1, channel=channel) + + assert await adapter._should_backfill_discord_message(message) is True + + +@pytest.mark.asyncio +async def test_thread_unreferenced_bot_message_does_not_mask_request(adapter): + bot_post = SimpleNamespace( + id=2, + content="Done — captured a different request.", + author=SimpleNamespace(id=999, bot=True), + reference=None, + created_at=datetime.now(timezone.utc), + ) + thread = FakeChannel(channel_id=456, parent_id=123, history_messages=[bot_post]) + message = make_message(message_id=1, channel=thread) + + assert await adapter._should_backfill_discord_message(message) is True + + +@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_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): + bot_user = adapter._client.user + message = make_message( + message_id=1, + content=f"<@{bot_user.id}> please ingest", + mentions=[bot_user], + ) + + 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, + role_authorized=False, + recovered=True, + ) + + +@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_recovery_aborts_when_durable_ledger_is_unavailable(adapter, monkeypatch): + dispatch = AsyncMock() + monkeypatch.setattr(adapter, "_dispatch_recovered_message", dispatch) + monkeypatch.setattr( + adapter, + "_with_discord_recovery_db_async", + AsyncMock(return_value=False), + ) + + await adapter._run_missed_message_backfill() + + dispatch.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_recovery_releases_dedup_claim_when_dispatch_is_cancelled(adapter, monkeypatch): + message = make_message(message_id=97) + started = asyncio.Event() + + async def cancelled_dispatch(_message): + adapter._dedup.is_duplicate(str(message.id)) + started.set() + await asyncio.Event().wait() + + monkeypatch.setattr(adapter, "_dispatch_recovered_message", cancelled_dispatch) + monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True)) + monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"}) + + async def candidates(_channels): + yield message + + monkeypatch.setattr(adapter, "_iter_missed_message_backfill_candidates", candidates) + task = asyncio.create_task(adapter._run_missed_message_backfill()) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert adapter._dedup.contains(str(message.id)) is False + + +@pytest.mark.asyncio +async def test_repeated_ready_coalesces_instead_of_cancelling_active_recovery(adapter): + started = asyncio.Event() + release = asyncio.Event() + + async def slow_recovery(): + started.set() + await release.wait() + + first = asyncio.create_task(slow_recovery()) + adapter._missed_message_backfill_task = first + await started.wait() + + second = adapter._ensure_missed_message_backfill_task() + + assert second is first + assert first.cancelled() is False + release.set() + await first + + +@pytest.mark.asyncio +async def test_recovery_task_joins_gateway_startup_restore(adapter, monkeypatch): + release = asyncio.Event() + + async def recovery(): + await release.wait() + + runner = SimpleNamespace( + _startup_restore_in_progress=True, + _startup_restore_tasks=[], + ) + adapter.gateway_runner = runner + monkeypatch.setattr(adapter, "_run_missed_message_backfill", recovery) + + task = adapter._ensure_missed_message_backfill_task() + + assert runner._startup_restore_tasks == [task] + release.set() + await task + + +@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, + recovered=True, + ) + + +@pytest.mark.asyncio +async def test_recovery_does_not_treat_unmentioned_message_as_dispatched(adapter, monkeypatch): + monkeypatch.setenv("DISCORD_REQUIRE_MENTION", "true") + monkeypatch.setenv("DISCORD_AUTO_THREAD", "false") + adapter.config.extra["free_response_channels"] = "" + adapter.handle_message = AsyncMock() + message = make_message(message_id=95, content="not addressed") + + assert await adapter._dispatch_recovered_message(message) is False + adapter.handle_message.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_recovered_messages_bypass_live_text_debounce(adapter, monkeypatch): + bot_user = adapter._client.user + message = make_message( + message_id=96, + content=f"<@{bot_user.id}> recover", + mentions=[bot_user], + ) + adapter._text_batch_delay_seconds = 0.6 + adapter._handle_message = DiscordAdapter._handle_message.__get__( + adapter, DiscordAdapter + ) + adapter.handle_message = AsyncMock() + monkeypatch.setenv("DISCORD_AUTO_THREAD", "false") + + assert await adapter._dispatch_recovered_message(message) is True + adapter.handle_message.assert_awaited_once() + assert adapter._pending_text_batches == {} + + +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" + ) + + config = load_gateway_config() + backfill = config.platforms[Platform.DISCORD].extra[ + "missed_message_backfill" + ] + + assert backfill == { + "enabled": True, + "channels": ["1501971993405292796"], + "window_seconds": 3600, + "limit": 25, + "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, + } + + +def test_missed_message_backfill_config_stays_per_adapter(): + first_extra = _apply_yaml_config( + {}, + { + "missed_message_backfill": { + "enabled": True, + "channels": ["111"], + "window_seconds": 60, + "limit": 5, + "max_dispatches": 2, + } + }, + ) + second_extra = _apply_yaml_config( + {}, + { + "missed_message_backfill": { + "enabled": False, + "channels": ["222"], + "window_seconds": 120, + "limit": 6, + "max_dispatches": 3, + } + }, + ) + + first = DiscordAdapter(PlatformConfig(enabled=True, token="one", extra=first_extra or {})) + second = DiscordAdapter(PlatformConfig(enabled=True, token="two", extra=second_extra or {})) + + assert first._missed_message_backfill_enabled() is True + assert first._missed_message_backfill_channels() == {"111"} + assert first._missed_message_backfill_window_seconds() == 60 + assert first._missed_message_backfill_limit() == 5 + assert first._missed_message_backfill_max_dispatches() == 2 + assert second._missed_message_backfill_enabled() is False + assert second._missed_message_backfill_channels() == {"222"} + assert second._missed_message_backfill_window_seconds() == 120 + assert second._missed_message_backfill_limit() == 6 + assert second._missed_message_backfill_max_dispatches() == 3 + + +def test_recovery_store_pins_profile_home_at_adapter_construction(monkeypatch, tmp_path): + first_home = tmp_path / "first" + second_home = tmp_path / "second" + monkeypatch.setenv("HERMES_HOME", str(first_home)) + adapter = DiscordAdapter(PlatformConfig(enabled=True, token="one")) + monkeypatch.setenv("HERMES_HOME", str(second_home)) + + assert adapter._discord_recovery_db_path() == ( + first_home / "gateway" / "discord_message_recovery.db" + ) + + +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) + 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.", + final=True, + ) + + 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=False, message_id="9002"), + content="The agent is down right now.", + final=True, + ) + + 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._discord_recovery_store._initialized = False + 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 + + +def test_fresh_processing_claim_suppresses_duplicate_recovery(adapter): + message = make_message(message_id=99) + 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) + + assert adapter._discord_message_has_active_claim("99") is True + + +def test_stale_processing_claim_is_recoverable(adapter): + message = make_message(message_id=100) + 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) + stale = (datetime.now(timezone.utc) - dt.timedelta(minutes=11)).isoformat() + adapter._with_discord_recovery_db( + lambda conn: conn.execute( + "UPDATE discord_messages SET updated_at=? WHERE message_id='100'", + (stale,), + ) + ) + + assert adapter._discord_message_has_active_claim("100") is False + + +@pytest.mark.asyncio +async def test_processing_hook_offloads_contended_ledger(adapter, monkeypatch): + message = make_message(message_id=101) + event = MessageEvent( + text=message.content, + message_type=MessageType.TEXT, + raw_message=message, + message_id=str(message.id), + ) + + def slow_record(*_args, **_kwargs): + import time + time.sleep(0.1) + + monkeypatch.setattr(adapter, "_record_discord_processing_start", slow_record) + processing = asyncio.create_task(adapter.on_processing_start(event)) + await asyncio.sleep(0.01) + + assert processing.done() is False + await processing + + +@pytest.mark.asyncio +async def test_recovery_scan_offloads_ledger_writes(adapter, monkeypatch): + def slow_scan_start(_channels): + import time + time.sleep(0.1) + return "scan" + + monkeypatch.setattr(adapter, "_record_recovery_scan_start", slow_scan_start) + monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: set()) + scan = asyncio.create_task(adapter._run_missed_message_backfill()) + await asyncio.sleep(0.01) + + assert scan.done() is False + await scan + + +@pytest.mark.asyncio +async def test_send_offloads_final_delivery_ledger_write(adapter, monkeypatch): + channel = FakeChannel(channel_id=123) + channel.send = AsyncMock(return_value=SimpleNamespace(id=9011)) + channel.fetch_message = AsyncMock() + adapter._client.get_channel = lambda _channel_id: channel + + def slow_record(**_kwargs): + import time + time.sleep(0.1) + + monkeypatch.setattr(adapter, "_record_discord_response", slow_record) + sending = asyncio.create_task( + adapter.send( + "123", + "done", + reply_to="104", + metadata={"notify": True}, + ) + ) + await asyncio.sleep(0.01) + + assert sending.done() is False + assert (await sending).success is True + + +def test_final_delivery_remains_complete_after_processing_hook(adapter): + message = make_message(message_id=91) + 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_response( + reply_to="91", + result=SimpleNamespace(success=True, message_id="9004"), + content="Done", + final=True, + ) + adapter._record_discord_processing_complete(event, ProcessingOutcome.SUCCESS) + + assert adapter._discord_message_is_persistently_complete("91") is True + + +def test_preview_delivery_does_not_mark_message_complete(adapter): + adapter._record_discord_response( + reply_to="92", + result=SimpleNamespace(success=True, message_id="9005"), + content="partial", + final=False, + ) + + assert adapter._discord_message_is_persistently_complete("92") is False + + +def test_successful_final_delivery_clears_prior_outage_state(adapter): + adapter._record_discord_response( + reply_to="93", + result=SimpleNamespace(success=False, message_id="9006"), + content="Hermes is offline", + final=True, + ) + assert adapter._discord_message_is_persistently_complete("93") is False + + adapter._record_discord_response( + reply_to="93", + result=SimpleNamespace(success=True, message_id="9007"), + content="Recovered successfully", + final=True, + ) + + assert adapter._discord_message_is_persistently_complete("93") is True + + +@pytest.mark.asyncio +async def test_send_uses_notify_metadata_as_final_delivery_signal(adapter): + channel = FakeChannel(channel_id=123) + channel.send = AsyncMock(return_value=SimpleNamespace(id=9008)) + channel.fetch_message = AsyncMock() + adapter._client.get_channel = lambda _channel_id: channel + + preview = await adapter.send( + "123", + "partial", + reply_to="94", + metadata={"expect_edits": True}, + ) + assert preview.success is True + assert adapter._discord_message_is_persistently_complete("94") is False + + final = await adapter.send( + "123", + "complete", + reply_to="94", + metadata={"notify": True}, + ) + assert final.success is True + assert adapter._discord_message_is_persistently_complete("94") is True + + +@pytest.mark.asyncio +async def test_final_stream_edit_marks_original_request_complete(adapter): + channel = FakeChannel(channel_id=123) + message = SimpleNamespace(edit=AsyncMock()) + channel.fetch_message = AsyncMock(return_value=message) + adapter._client.get_channel = lambda _channel_id: channel + + result = await adapter.edit_message( + "123", + "9009", + "complete streamed response", + finalize=True, + metadata={"reply_to_message_id": "102"}, + ) + + assert result.success is True + assert adapter._discord_message_is_persistently_complete("102") is True + + +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", + final=True, + ) + + 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)) + 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] + + +@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 len(got) == 3 + assert set(got).issubset({1, 2, 3, 4}) + + +@pytest.mark.asyncio +async def test_iter_candidates_round_robins_configured_channels(adapter, monkeypatch): + first = FakeChannel( + channel_id=123, + history_messages=[ + make_message(message_id=1), + make_message(message_id=2), + make_message(message_id=3), + ], + ) + second = FakeChannel( + channel_id=456, + history_messages=[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 message in adapter._iter_missed_message_backfill_candidates({"123", "456"}): + got.append(message.id) + + assert 4 in got + + +@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] + + +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_cursor_does_not_advance_past_incomplete_dispatched_message(adapter, monkeypatch): + channel = FakeChannel( + channel_id=123, + history_messages=[ + make_message(message_id=1), + make_message(message_id=2), + ], + ) + for message in channel._history_messages: + message.channel = channel + adapter._client.get_channel = lambda _channel_id: channel + monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"}) + monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True)) + monkeypatch.setattr(adapter, "_dispatch_recovered_message", AsyncMock(side_effect=[True, True])) + monkeypatch.setattr(adapter, "_missed_message_backfill_max_dispatches", lambda: 10) + + await adapter._run_missed_message_backfill() + + assert adapter._discord_recovery_cursor("123") is None + + +def test_final_delivery_advances_channel_cursor(adapter): + message = make_message(message_id=103, channel=FakeChannel(channel_id=123)) + adapter._record_discord_message_seen(message, status="processing") + + adapter._record_discord_response( + reply_to="103", + result=SimpleNamespace(success=True, message_id="9010"), + content="done", + final=True, + ) + + assert adapter._discord_recovery_cursor("123") == "103" + + +@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 diff --git a/tests/gateway/test_message_deduplicator.py b/tests/gateway/test_message_deduplicator.py index e6470075284c..cf70e4944752 100644 --- a/tests/gateway/test_message_deduplicator.py +++ b/tests/gateway/test_message_deduplicator.py @@ -60,6 +60,19 @@ def test_empty_id_never_duplicate(self): 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) diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index cd49d3d74782..b43bacf046a5 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -9,6 +9,22 @@ from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig +def test_stream_send_metadata_carries_original_reply_anchor(): + consumer = GatewayStreamConsumer( + adapter=MagicMock(), + chat_id="123", + initial_reply_to_id="456", + ) + + assert consumer._metadata_for_send(final=False) == { + "reply_to_message_id": "456", + } + assert consumer._metadata_for_send(final=True) == { + "reply_to_message_id": "456", + "notify": True, + } + + # ── _clean_for_display unit tests ──────────────────────────────────────── diff --git a/tests/gateway/test_stream_consumer_thread_routing.py b/tests/gateway/test_stream_consumer_thread_routing.py index a133a0784664..009be5c388bf 100644 --- a/tests/gateway/test_stream_consumer_thread_routing.py +++ b/tests/gateway/test_stream_consumer_thread_routing.py @@ -103,7 +103,11 @@ async def test_metadata_passed_on_first_send(self): await consumer._send_or_edit("Test") call_kwargs = adapter.send.call_args[1] - assert call_kwargs["metadata"] == {**metadata, "expect_edits": True} + assert call_kwargs["metadata"] == { + **metadata, + "reply_to_message_id": "om_msg_000", + "expect_edits": True, + } assert metadata == {"thread_id": "omt_topic789"} @pytest.mark.asyncio @@ -140,7 +144,11 @@ async def test_nonfinal_first_send_does_not_mark_notify(self): await consumer._send_or_edit("Preview", finalize=False) metadata = adapter.send.call_args[1]["metadata"] - assert metadata == {"thread_id": "root_post_123", "expect_edits": True} + assert metadata == { + "thread_id": "root_post_123", + "reply_to_message_id": "reply_post_456", + "expect_edits": True, + } class TestOverflowFirstMessage: diff --git a/tests/hermes_cli/test_backup.py b/tests/hermes_cli/test_backup.py index fc6de295230e..cbac29eae503 100644 --- a/tests/hermes_cli/test_backup.py +++ b/tests/hermes_cli/test_backup.py @@ -1489,6 +1489,26 @@ def test_copies_nested_files(self, hermes_home): 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) diff --git a/website/docs/user-guide/messaging/discord.md b/website/docs/user-guide/messaging/discord.md index 6b55e1c4949f..3414e14414e6 100644 --- a/website/docs/user-guide/messaging/discord.md +++ b/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` diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md index ade7ce08bd6a..8df87a340e2c 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/discord.md +++ b/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`