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

@@ -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()

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(