fix(discord): keep recovery ledger I/O off event loop

Offload scan bookkeeping and final-delivery writes so SQLite contention cannot stall Discord heartbeats or message delivery.

plugins/platforms/discord/adapter.py

@@ -2023,10 +2023,20 @@ class DiscordAdapter(BasePlatformAdapter):
                 self.name,
             )
             return
-        scan_id = self._record_recovery_scan_start(channels)
+        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)
-            self._record_recovery_scan_complete(scan_id, status="skipped", scanned=0, missed=0, dispatched=0)
+            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()
@@ -2067,7 +2077,14 @@ class DiscordAdapter(BasePlatformAdapter):
                     raise
                 if dispatched >= max_dispatches:
                     break
-            self._record_recovery_scan_complete(scan_id, status="success", scanned=scanned, missed=missed, dispatched=dispatched)
+            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,
@@ -2076,10 +2093,25 @@ class DiscordAdapter(BasePlatformAdapter):
                 dispatched,
             )
         except asyncio.CancelledError:
-            self._record_recovery_scan_complete(scan_id, status="cancelled", scanned=scanned, missed=missed, dispatched=dispatched)
+            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
-            self._record_recovery_scan_complete(scan_id, status="failed", scanned=scanned, missed=missed, dispatched=dispatched, error=str(exc))
+            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:
@@ -2825,7 +2857,8 @@ class DiscordAdapter(BasePlatformAdapter):
             # Forum channels reject channel.send() — create a thread post instead.
             if self._is_forum_parent(channel):
                 result = await self._send_to_forum(channel, content)
-                self._record_discord_response(
+                await asyncio.to_thread(
+                    self._record_discord_response,
                     reply_to=reply_to,
                     result=result,
                     content=content,
@@ -2900,7 +2933,8 @@ class DiscordAdapter(BasePlatformAdapter):
                 message_id=message_ids[0] if message_ids else None,
                 raw_response={"message_ids": message_ids}
             )
-            self._record_discord_response(
+            await asyncio.to_thread(
+                self._record_discord_response,
                 reply_to=reply_to,
                 result=result,
                 content=content,
@@ -2911,7 +2945,8 @@ class DiscordAdapter(BasePlatformAdapter):
         except Exception as e:  # pragma: no cover - defensive logging
             logger.error("[%s] Failed to send Discord message: %s", self.name, e, exc_info=True)
             result = SendResult(success=False, error=str(e))
-            self._record_discord_response(
+            await asyncio.to_thread(
+                self._record_discord_response,
                 reply_to=reply_to,
                 result=result,
                 content=content,
@@ -3124,7 +3159,8 @@ class DiscordAdapter(BasePlatformAdapter):
                     raise
             result = SendResult(success=True, message_id=message_id)
             if finalize:
-                self._record_discord_response(
+                await asyncio.to_thread(
+                    self._record_discord_response,
                     reply_to=(metadata or {}).get("reply_to_message_id"),
                     result=result,
                     content=content,

tests/gateway/test_discord_missed_message_backfill.py

@@ -659,6 +659,48 @@ async def test_processing_hook_offloads_contended_ledger(adapter, monkeypatch):
     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(