From d5b9c1ee37bba5da3de61b49defe5d7fc40b0819 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 17 Jul 2026 02:49:40 -0700 Subject: [PATCH] fix(discord): guard recovery claims and ledger failures Honor configured bot senders at shared ingress, fail closed when durable state is unavailable, and suppress duplicate reconnect work while a fresh queued/processing claim is active. --- plugins/platforms/discord/adapter.py | 34 ++++++- .../test_discord_missed_message_backfill.py | 94 +++++++++++++++++++ 2 files changed, 126 insertions(+), 2 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index a7a07f2847fe..041b349b5732 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -2013,6 +2013,16 @@ class DiscordAdapter(BasePlatformAdapter): 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 = self._record_recovery_scan_start(channels) if not channels: logger.info("[%s] Missed-message backfill enabled but no channels configured", self.name) @@ -2185,10 +2195,10 @@ class DiscordAdapter(BasePlatformAdapter): 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 + 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): @@ -2412,6 +2422,26 @@ class DiscordAdapter(BasePlatformAdapter): 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() diff --git a/tests/gateway/test_discord_missed_message_backfill.py b/tests/gateway/test_discord_missed_message_backfill.py index 27a55bdfd14e..ed7e85e33300 100644 --- a/tests/gateway/test_discord_missed_message_backfill.py +++ b/tests/gateway/test_discord_missed_message_backfill.py @@ -114,6 +114,17 @@ def make_message(*, message_id=1, author_id=42, content="please ingest", reactio ) +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)]) @@ -121,6 +132,19 @@ async def test_backfills_message_with_only_own_success_reaction(adapter): 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( @@ -248,6 +272,21 @@ async def test_run_backfill_counts_only_messages_that_reach_dispatch(adapter, mo 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) @@ -564,6 +603,61 @@ def test_empty_successful_turn_is_not_persistently_complete(adapter): 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 + + def test_final_delivery_remains_complete_after_processing_hook(adapter): message = make_message(message_id=91) event = MessageEvent(