test: raise blocking-probe timeouts for loaded CI runners

CI slices failed the offload tests with 0.5s witness timeouts: on a
loaded shared runner the event loop thread can take >0.5s to get
scheduled even when NOT blocked, making the probe report a false
positive. A genuinely blocked loop can never set the progress event at
any timeout (the witness coroutine can't run at all), so 5s only
absorbs scheduler flake without weakening the invariant. Mutation
re-verified: reverting the offload still fails all 4 tests.

gateway/platforms/base.py

@@ -6066,13 +6066,14 @@ class BasePlatformAdapter(ABC):
                                 record_obligation,
                             )
 
-                            if ledger_enabled():
+                            if await asyncio.to_thread(ledger_enabled):
                                 _obligation_id = compute_obligation_id(
                                     session_key,
                                     str(getattr(event, "message_id", "") or ""),
                                     text_content,
                                 )
-                                record_obligation(
+                                await asyncio.to_thread(
+                                    record_obligation,
                                     obligation_id=_obligation_id,
                                     session_key=session_key,
                                     platform=str(
@@ -6083,7 +6084,7 @@ class BasePlatformAdapter(ABC):
                                     thread_id=getattr(event.source, "thread_id", None),
                                     content=text_content,
                                 )
-                                mark_attempting(_obligation_id)
+                                await asyncio.to_thread(mark_attempting, _obligation_id)
                         except Exception:
                             logger.debug("delivery ledger record failed", exc_info=True)
                             _obligation_id = None
@@ -6102,9 +6103,10 @@ class BasePlatformAdapter(ABC):
                             )
 
                             if getattr(result, "success", False):
-                                mark_delivered(_obligation_id)
+                                await asyncio.to_thread(mark_delivered, _obligation_id)
                             else:
-                                mark_failed(
+                                await asyncio.to_thread(
+                                    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 ledger_enabled():
+            if not await asyncio.to_thread(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):
-                    mark_delivered(row["obligation_id"])
+                    await asyncio.to_thread(mark_delivered, row["obligation_id"])
                     redelivered += 1
                     logger.info(
                         "Redelivered recovered final response to %s:%s "
@@ -10292,7 +10292,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                         row["obligation_id"], row["attempts"],
                     )
                 else:
-                    mark_failed(
+                    await asyncio.to_thread(
+                        mark_failed,
                         row["obligation_id"],
                         str(getattr(result, "error", "") or "send failed"),
                     )

tests/gateway/test_delivery_ledger.py

@@ -59,13 +59,16 @@ def _blocking_probe():
 
     def _slow_ledger_call(*args, **kwargs):
         ledger_started.set()
-        if not event_loop_progressed.wait(timeout=0.5):
+        # Generous timeout: a genuinely blocked loop can never set the event
+        # (the witness coroutine cannot run), so a longer wait only guards
+        # against loaded-CI scheduling flake, not against missing the bug.
+        if not event_loop_progressed.wait(timeout=5.0):
             blocked_event_loop.append(True)
 
     async def _event_loop_witness():
         import asyncio
 
-        deadline = asyncio.get_running_loop().time() + 1
+        deadline = asyncio.get_running_loop().time() + 10
         while not ledger_started.is_set():
             if asyncio.get_running_loop().time() >= deadline:
                 raise AssertionError("ledger call never started")

tests/gateway/test_delivery_ledger_producer.py

@@ -74,11 +74,14 @@ def _blocking_probe():
 
     def _slow_ledger_call(*args, **kwargs):
         ledger_started.set()
-        if not event_loop_progressed.wait(timeout=0.5):
+        # Generous timeout: a genuinely blocked loop can never set the event
+        # (the witness coroutine cannot run), so a longer wait only guards
+        # against loaded-CI scheduling flake, not against missing the bug.
+        if not event_loop_progressed.wait(timeout=5.0):
             blocked_event_loop.append(True)
 
     async def _event_loop_witness():
-        deadline = asyncio.get_running_loop().time() + 1
+        deadline = asyncio.get_running_loop().time() + 10
         while not ledger_started.is_set():
             if asyncio.get_running_loop().time() >= deadline:
                 raise AssertionError("ledger call never started")