refactor(gateway): give the tracker persist a real _to_thread seam

Commit 93fdc661b4 patched ``helpers.asyncio.to_thread``, but
``helpers.asyncio is asyncio`` so that was a global patch, not a module
seam.  ``helpers._to_thread = asyncio.to_thread`` is now the attribute
``mark_async`` awaits, and the read-after-mark test patches that name and
leaves ``asyncio.to_thread`` untouched.

PROOF: with ``mark_async`` mutated back to ``asyncio.to_thread(self._save)``
the test fails ``mark_async never reached the to_thread handoff``; reverted,
``helpers._to_thread is asyncio.to_thread`` under PYTHONSAFEPATH=1 and
run_tests.sh over the 4 branch test files: 7 passed, 0 failed.

gateway/platforms/helpers.py

@@ -61,6 +61,12 @@ class MessageDeduplicator:
         self._seen.clear()
 
 
+# Worker-thread handoff used by the off-loop persist paths.  A module attribute
+# so tests can replace THIS seam instead of patching ``asyncio.to_thread``
+# globally.
+_to_thread = asyncio.to_thread
+
+
 async def cancel_task(task: Optional[asyncio.Task]) -> None:
     """Cancel *task* and wait for it to unwind. ``None``/finished tasks are no-ops; awaiting the
     current task would deadlock, so a self-cancel only requests cancellation. Exceptions the task
@@ -181,7 +187,7 @@ class ThreadParticipationTracker:
         long as it runs.
         """
         if self._remember(thread_id):
-            await asyncio.to_thread(self._save)
+            await _to_thread(self._save)
 
     def __contains__(self, thread_id: str) -> bool:
         with self._lock:

tests/gateway/test_thread_tracker_mark_off_loop.py

@@ -123,7 +123,7 @@ def test_a_mark_is_visible_in_memory_before_the_persist_completes(tracker, monke
     would reopen the mention-gating hole the tracker exists to close -- and
     would make membership depend on executor availability.
 
-    The oracle is a handoff that NEVER RUNS: ``asyncio.to_thread`` is replaced
+    The oracle is a handoff that NEVER RUNS: ``helpers._to_thread`` is replaced
     by a coroutine that parks forever without ever invoking the callable.  That
     is what makes this non-vacuous.  Merely waiting for the worker to reach the
     rename -- the obvious shape -- also observes a True under the deferred
@@ -136,8 +136,9 @@ def test_a_mark_is_visible_in_memory_before_the_persist_completes(tracker, monke
         handed_off.append(fn)
         await asyncio.Event().wait()  # park forever; fn is never called
 
-    # Patch through the module seam the tracker actually calls.
-    monkeypatch.setattr(helpers.asyncio, "to_thread", _never, raising=True)
+    # Patch the module seam the tracker actually calls; ``asyncio.to_thread``
+    # itself is left alone.
+    monkeypatch.setattr(helpers, "_to_thread", _never, raising=True)
 
     async def scenario():
         mark = asyncio.create_task(tracker.mark_async("!first:example.org"))