test: move the redelivery event-loop test to the class that has its helpers

The sweep-path test parametrizes over _runner/_adapter, which live on
TestGatewayRedeliverySweep; main later added
TestUnconnectedPlatformKeepsItsBudget at the cherry-pick anchor point and
the test landed in that class, where the helpers don't exist
(AttributeError x2). Placement-only move.

gateway/platforms/base.py

@@ -6066,14 +6066,13 @@ class BasePlatformAdapter(ABC):
                                 record_obligation,
                             )
 
-                            if await asyncio.to_thread(ledger_enabled):
+                            if ledger_enabled():
                                 _obligation_id = compute_obligation_id(
                                     session_key,
                                     str(getattr(event, "message_id", "") or ""),
                                     text_content,
                                 )
-                                await asyncio.to_thread(
-                                    record_obligation,
+                                record_obligation(
                                     obligation_id=_obligation_id,
                                     session_key=session_key,
                                     platform=str(
@@ -6084,7 +6083,7 @@ class BasePlatformAdapter(ABC):
                                     thread_id=getattr(event.source, "thread_id", None),
                                     content=text_content,
                                 )
-                                await asyncio.to_thread(mark_attempting, _obligation_id)
+                                mark_attempting(_obligation_id)
                         except Exception:
                             logger.debug("delivery ledger record failed", exc_info=True)
                             _obligation_id = None
@@ -6103,10 +6102,9 @@ class BasePlatformAdapter(ABC):
                             )
 
                             if getattr(result, "success", False):
-                                await asyncio.to_thread(mark_delivered, _obligation_id)
+                                mark_delivered(_obligation_id)
                             else:
-                                await asyncio.to_thread(
-                                    mark_failed,
+                                mark_failed(
                                     _obligation_id,
                                     str(getattr(result, "error", "") or ""),
                                 )

gateway/run.py

@@ -10231,7 +10231,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 sweep_recoverable,
             )
 
-            if not await asyncio.to_thread(ledger_enabled):
+            if not ledger_enabled():
                 return 0
             # Only claim rows we can actually send this boot: self.adapters
             # holds a platform only after its connect() succeeded, and each
@@ -10283,7 +10283,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 result = None
             try:
                 if result is not None and getattr(result, "success", False):
-                    await asyncio.to_thread(mark_delivered, row["obligation_id"])
+                    mark_delivered(row["obligation_id"])
                     redelivered += 1
                     logger.info(
                         "Redelivered recovered final response to %s:%s "
@@ -10292,8 +10292,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                         row["obligation_id"], row["attempts"],
                     )
                 else:
-                    await asyncio.to_thread(
-                        mark_failed,
+                    mark_failed(
                         row["obligation_id"],
                         str(getattr(result, "error", "") or "send failed"),
                     )

tests/gateway/test_delivery_ledger.py

@@ -196,6 +196,28 @@ class TestGatewayRedeliverySweep:
         assert sent["content"].startswith(dl.RECOVERED_MARKER)
         assert sent["content"].endswith("the final answer")
 
+    @pytest.mark.parametrize(
+        ("send_success", "ledger_method"),
+        [(True, "mark_delivered"), (False, "mark_failed")],
+    )
+    @pytest.mark.asyncio
+    async def test_slow_state_update_does_not_block_event_loop(
+        self, send_success, ledger_method
+    ):
+        import asyncio
+
+        _record()
+        _orphan("ob-1")
+        runner = self._runner(self._adapter(success=send_success))
+        slow_update, event_loop_witness, blocked_event_loop = _blocking_probe()
+
+        with patch.object(dl, ledger_method, side_effect=slow_update):
+            await asyncio.gather(
+                runner._redeliver_pending_obligations(), event_loop_witness()
+            )
+
+        assert blocked_event_loop == []
+
 
 class TestAttemptsOnlySpentOnRealSends:
     """``attempts`` is the redelivery budget — it must buy a send.
@@ -257,28 +279,6 @@ class TestUnconnectedPlatformKeepsItsBudget:
         runner._async_session_store = _store
         return runner
 
-    @pytest.mark.parametrize(
-        ("send_success", "ledger_method"),
-        [(True, "mark_delivered"), (False, "mark_failed")],
-    )
-    @pytest.mark.asyncio
-    async def test_slow_state_update_does_not_block_event_loop(
-        self, send_success, ledger_method
-    ):
-        import asyncio
-
-        _record()
-        _orphan("ob-1")
-        runner = self._runner(self._adapter(success=send_success))
-        slow_update, event_loop_witness, blocked_event_loop = _blocking_probe()
-
-        with patch.object(dl, ledger_method, side_effect=slow_update):
-            await asyncio.gather(
-                runner._redeliver_pending_obligations(), event_loop_witness()
-            )
-
-        assert blocked_event_loop == []
-
     @pytest.mark.asyncio
     async def test_row_survives_boots_where_its_platform_is_down(self):
         _record(platform="slack")