fix(channels): offload blocking filesystem IO in Wechat channel (#3925)

WechatChannel made synchronous filesystem calls (mkdir, write_text,
read_bytes, Path.replace, unlink) directly inside async entry points:
_poll_loop, _bind_via_qrcode, _ensure_authenticated, _extract_image_file,
_extract_file_item, start, _send_image_attachment, _send_file_attachment.
Under slow disks, large files, or concurrent load these blocked the
asyncio event loop and stalled the channel worker.

Construction was also blocking: __init__ called _load_state() (os.stat +
read_text) synchronously, and ChannelService._start_channel() instantiates
the channel directly on the async path, so constructing WechatChannel in an
async context raised BlockingError. Persisted state (auth token + cursor)
is now loaded in start() via asyncio.to_thread, leaving __init__ IO-free.

Offload each call to a thread via asyncio.to_thread, matching the existing
pattern in channels/manager.py and dingtalk.py. The sync helpers
(_save_state, _save_auth_state, _load_auth_state, _stage_downloaded_file)
keep their signatures; only the async call sites wrap them.

Adds tests/blocking_io/test_wechat_channel_state.py as a regression anchor
covering the IO-free constructor (the production _start_channel path), the
staging write path, and the auth-state read path. Detected by
`make detect-blocking-io`.

backend/app/channels/wechat.py

@@ -255,19 +255,29 @@ class WechatChannel(Channel):
         self._state_dir = self._resolve_state_dir(config.get("state_dir"))
         self._cursor_path = self._state_dir / "wechat-getupdates.json" if self._state_dir else None
         self._auth_path = self._state_dir / "wechat-auth.json" if self._state_dir else None
-        self._load_state()
+        # NOTE: persisted state (auth token + cursor) is intentionally NOT loaded
+        # here. ChannelService._start_channel() constructs the channel directly
+        # on the async path, so filesystem IO in __init__ would block the event
+        # loop (the strict blocking-IO gate raises BlockingError on os.stat).
+        # State is loaded in start() via asyncio.to_thread instead.
 
     async def start(self) -> None:
         if self._running:
             return
 
+        # Load persisted state off the event loop before the bot_token check
+        # below: a token restored from the auth file must be visible here so
+        # the qrcode-login fallback isn't taken unnecessarily. __init__ defers
+        # this load precisely so construction stays IO-free on the async path.
+        await asyncio.to_thread(self._load_state)
+
         if not self._bot_token and not self._qrcode_login_enabled:
             logger.error("WeChat channel requires bot_token or qrcode_login_enabled")
             return
 
         self._main_loop = asyncio.get_running_loop()
         if self._state_dir:
-            self._state_dir.mkdir(parents=True, exist_ok=True)
+            await asyncio.to_thread(self._state_dir.mkdir, parents=True, exist_ok=True)
 
         await self._ensure_client()
         self._running = True
@@ -372,7 +382,7 @@ class WechatChannel(Channel):
             return False
 
         try:
-            plaintext = attachment.actual_path.read_bytes()
+            plaintext = await asyncio.to_thread(attachment.actual_path.read_bytes)
         except OSError:
             logger.exception("[WeChat] failed to read outbound image %s", attachment.actual_path)
             return False
@@ -462,7 +472,7 @@ class WechatChannel(Channel):
             return False
 
         try:
-            plaintext = attachment.actual_path.read_bytes()
+            plaintext = await asyncio.to_thread(attachment.actual_path.read_bytes)
         except OSError:
             logger.exception("[WeChat] failed to read outbound file %s", attachment.actual_path)
             return False
@@ -555,8 +565,8 @@ class WechatChannel(Channel):
                     if errcode == -14:
                         self._bot_token = ""
                         self._get_updates_buf = ""
-                        self._save_state()
-                        self._save_auth_state(status="expired", bot_token="")
+                        await asyncio.to_thread(self._save_state)
+                        await asyncio.to_thread(self._save_auth_state, status="expired", bot_token="")
                         logger.error("[WeChat] bot token expired; scan again or update bot_token and restart the channel")
                         self._running = False
                         break
@@ -574,7 +584,7 @@ class WechatChannel(Channel):
                 next_buf = data.get("get_updates_buf")
                 if isinstance(next_buf, str) and next_buf != self._get_updates_buf:
                     self._get_updates_buf = next_buf
-                    self._save_state()
+                    await asyncio.to_thread(self._save_state)
 
                 for raw_message in data.get("msgs", []):
                     await self._handle_update(raw_message)
@@ -692,7 +702,7 @@ class WechatChannel(Channel):
             if self._bot_token:
                 return True
 
-            self._load_auth_state()
+            await asyncio.to_thread(self._load_auth_state)
             if self._bot_token:
                 return True
 
@@ -720,7 +730,8 @@ class WechatChannel(Channel):
         if qrcode_img_content:
             logger.warning("[WeChat] qrcode_img_content=%s", qrcode_img_content)
 
-        self._save_auth_state(
+        await asyncio.to_thread(
+            self._save_auth_state,
             status="pending",
             qrcode=qrcode,
             qrcode_img_content=qrcode_img_content or None,
@@ -742,7 +753,8 @@ class WechatChannel(Channel):
                 if ilink_bot_id:
                     self._ilink_bot_id = ilink_bot_id
 
-                return self._save_auth_state(
+                return await asyncio.to_thread(
+                    self._save_auth_state,
                     status="confirmed",
                     bot_token=token,
                     ilink_bot_id=self._ilink_bot_id,
@@ -751,7 +763,8 @@ class WechatChannel(Channel):
                 )
 
             if status in {"expired", "canceled", "cancelled", "invalid", "failed"}:
-                self._save_auth_state(
+                await asyncio.to_thread(
+                    self._save_auth_state,
                     status=status,
                     qrcode=qrcode,
                     qrcode_img_content=qrcode_img_content or None,
@@ -760,7 +773,8 @@ class WechatChannel(Channel):
 
             await asyncio.sleep(max(self._qrcode_poll_interval, 0.1))
 
-        self._save_auth_state(
+        await asyncio.to_thread(
+            self._save_auth_state,
             status="timeout",
             qrcode=qrcode,
             qrcode_img_content=qrcode_img_content or None,
@@ -1046,7 +1060,7 @@ class WechatChannel(Channel):
         detected_image = _detect_image_extension_and_mime(decrypted)
         image_extension = detected_image[0] if detected_image else ".jpg"
         filename = _safe_media_filename("wechat-image", image_extension, message_id=message_id, index=index)
-        stored_path = self._stage_downloaded_file(filename, decrypted)
+        stored_path = await asyncio.to_thread(self._stage_downloaded_file, filename, decrypted)
         if stored_path is None:
             return None
 
@@ -1097,7 +1111,7 @@ class WechatChannel(Channel):
             logger.warning("[WeChat] inbound file exceeds size limit (%d bytes), skipping message_id=%s", len(decrypted), message_id)
             return None
 
-        stored_path = self._stage_downloaded_file(filename, decrypted)
+        stored_path = await asyncio.to_thread(self._stage_downloaded_file, filename, decrypted)
         if stored_path is None:
             return None
 

backend/tests/blocking_io/test_wechat_channel_state.py

@@ -0,0 +1,114 @@
+"""Regression anchor: WeChat channel filesystem IO must not block the event loop.
+
+Two production paths touch the filesystem, and both must stay off the asyncio
+loop:
+
+1. **Construction** — ``ChannelService._start_channel()`` instantiates the
+   channel directly on the async path (``channel_cls(...)`` with no thread
+   offload), so ``WechatChannel.__init__`` must be IO-free; persisted state is
+   loaded later in ``start()`` via ``asyncio.to_thread``. Constructing the
+   channel in an async context used to raise ``BlockingError: Blocking call to
+   os.stat`` because ``__init__`` called ``_load_state`` synchronously.
+
+2. **Runtime** — ``_handle_update`` stages downloaded inbound files
+   (``mkdir`` + ``write_bytes``) and ``_ensure_authenticated`` reads persisted
+   auth state (``read_text``); both offload via ``asyncio.to_thread``.
+
+If any of this regresses back onto the event loop, the strict Blockbuster gate
+raises ``BlockingError`` and these tests fail.
+
+The constructors below are invoked *directly* on the event loop (no
+``asyncio.to_thread`` wrapper) on purpose: that mirrors the production
+``_start_channel`` path the gate is meant to protect. Test-only file setup
+(writing the auth fixture) is still wrapped in ``asyncio.to_thread`` because
+that scaffolding IO would otherwise trip the gate even though it is not the
+code under test.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+
+import pytest
+
+from app.channels.message_bus import MessageBus
+from app.channels.wechat import WechatChannel, _encrypt_aes_128_ecb
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_wechat_constructor_is_io_free_on_async_path(tmp_path: Path) -> None:
+    """``__init__`` must not touch the filesystem — ``_start_channel`` constructs inline."""
+    bus = MessageBus()
+    # Seed an auth file that WOULD restore a different token if __init__ read it.
+    auth_path = tmp_path / "wechat-auth.json"
+    await asyncio.to_thread(auth_path.write_text, json.dumps({"status": "confirmed", "bot_token": "from-disk"}))
+
+    # Direct construction on the event loop, no asyncio.to_thread wrapper —
+    # this is exactly the production _start_channel path. If __init__ regresses
+    # to doing os.stat / read_text (e.g. re-adding _load_state here), the gate
+    # raises BlockingError right at this line.
+    channel = WechatChannel(bus=bus, config={"bot_token": "from-config", "state_dir": str(tmp_path)})
+    # Token came from config, not the disk file — proving __init__ did not read it.
+    assert channel._bot_token == "from-config"
+
+
+async def test_wechat_inbound_file_staging_does_not_block_event_loop(tmp_path: Path) -> None:
+    """Staging a downloaded inbound image writes through ``asyncio.to_thread``."""
+    bus = MessageBus()
+    published = []
+
+    async def capture(msg):
+        published.append(msg)
+
+    bus.publish_inbound = capture  # type: ignore[method-assign]
+
+    channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)})
+
+    plaintext = b"fake-image-bytes"
+    aes_key = b"1234567890abcdef"
+    encrypted = _encrypt_aes_128_ecb(plaintext, aes_key)
+
+    async def _fake_download(_url: str, *, timeout: float | None = None):
+        return encrypted
+
+    channel._download_cdn_bytes = _fake_download  # type: ignore[method-assign]
+
+    await channel._handle_update(
+        {
+            "message_type": 1,
+            "message_id": 101,
+            "from_user_id": "wx-1",
+            "context_token": "ctx-img",
+            "item_list": [
+                {
+                    "type": 2,
+                    "image_item": {"aeskey": aes_key.hex(), "media": {"full_url": "https://cdn.example/image.bin"}},
+                }
+            ],
+        }
+    )
+
+    assert len(published) == 1
+    assert len(published[0].files) == 1
+    staged = Path(published[0].files[0]["path"])
+    staged_exists = await asyncio.to_thread(staged.exists)
+    assert staged_exists, "inbound image should be staged under the tmp state dir"
+
+
+async def test_wechat_auth_state_load_does_not_block_event_loop(tmp_path: Path) -> None:
+    """``_ensure_authenticated`` reads persisted auth state through ``asyncio.to_thread``."""
+    bus = MessageBus()
+    # bot_token="" forces _ensure_authenticated to fall through to the
+    # _load_auth_state branch (offloaded) instead of returning early.
+    channel = WechatChannel(bus=bus, config={"bot_token": "", "state_dir": str(tmp_path)})
+
+    auth_path = tmp_path / "wechat-auth.json"
+    await asyncio.to_thread(auth_path.write_text, json.dumps({"status": "confirmed", "bot_token": "loaded-token"}))
+
+    result = await channel._ensure_authenticated()
+
+    assert result is True
+    assert channel._bot_token == "loaded-token"

backend/tests/test_wechat_channel.py

@@ -1248,6 +1248,9 @@ def test_state_cursor_is_loaded_from_disk(tmp_path: Path):
         bus=MessageBus(),
         config={"bot_token": "bot-token", "state_dir": str(state_dir)},
     )
+    # State load moved out of __init__ (it does filesystem IO that would block
+    # the async path); mirror start() by loading explicitly here.
+    channel._load_state()
 
     assert channel._get_updates_buf == "cursor-123"
 
@@ -1266,6 +1269,9 @@ def test_auth_state_is_loaded_from_disk(tmp_path: Path):
         bus=MessageBus(),
         config={"state_dir": str(state_dir), "qrcode_login_enabled": True},
     )
+    # State load moved out of __init__ (it does filesystem IO that would block
+    # the async path); mirror start() by loading explicitly here.
+    channel._load_state()
 
     assert channel._bot_token == "saved-token"
     assert channel._ilink_bot_id == "bot-1"