fix(channels): isolate per-message failures in WeChat poll loop (#4231)

_poll_loop persisted the long-poll cursor (get_updates_buf) for the whole
getupdates batch immediately on receipt, then iterated data["msgs"] with no
per-message error handling. If any single message's processing raised (e.g.
an attachment that fails to decrypt), the for loop aborted and every message
after the failing one in that batch was never handled -- but since the cursor
had already advanced past the whole batch, the next poll would never re-fetch
them. The loss was silent and permanent, not a delay.

Fix is two parts:
- Wrap each _handle_update call in its own try/except (re-raising
  CancelledError) so one bad message is logged and skipped instead of
  aborting the rest of the batch.
- Move the cursor advance/persist to after the per-message loop instead of
  before it, so a hard crash mid-batch leaves the cursor unmoved. Worst case
  becomes re-fetching and re-processing the batch on restart, not silently
  skipping messages that were never actually handled.

Regression test drives the real _handle_update over a 3-message batch where
the middle message is a WeChat image item with deliberately non-block-aligned
"encrypted" bytes, so decryption raises a genuine cryptography ValueError
(the same failure shape a corrupt real attachment would produce). Confirms
the first and third messages both still reach the bus, the failure is logged
with the message id, and the persisted cursor reflects the fully-attempted
batch.

backend/app/channels/wechat.py

@@ -581,13 +581,30 @@ class WechatChannel(Channel):
 
                 self._update_longpoll_timeout(data)
 
+                # Each message is isolated in its own try/except: one message that
+                # fails to process (e.g. an attachment that fails to decrypt) must
+                # not abort the whole batch and strand every message after it.
+                for raw_message in data.get("msgs", []):
+                    try:
+                        await self._handle_update(raw_message)
+                    except asyncio.CancelledError:
+                        raise
+                    except Exception:
+                        message_id = raw_message.get("message_id") or raw_message.get("msg_id") if isinstance(raw_message, dict) else None
+                        logger.exception(
+                            "[WeChat] failed to handle inbound message message_id=%s; skipping it and continuing with the rest of the batch",
+                            message_id,
+                        )
+
+                # The cursor is advanced only after the whole batch has been
+                # attempted (not before the loop above), so a hard crash mid-batch
+                # leaves it unmoved -- the worst case on restart is re-fetching and
+                # re-processing this batch, not silently skipping past messages
+                # that were never actually handled.
                 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
                     await asyncio.to_thread(self._save_state)
-
-                for raw_message in data.get("msgs", []):
-                    await self._handle_update(raw_message)
             except asyncio.CancelledError:
                 raise
             except Exception:

backend/tests/test_wechat_channel.py

@@ -1234,6 +1234,139 @@ def test_poll_loop_updates_server_timeout(monkeypatch):
     _run(go())
 
 
+def test_poll_loop_one_bad_message_does_not_permanently_lose_its_siblings(monkeypatch, tmp_path: Path, caplog):
+    """A single message that fails to process must not sink the rest of its batch.
+
+    Regression test for a permanent message-loss bug: the long-poll cursor was
+    persisted for the *whole* batch before the per-message loop ran, and the loop
+    had no per-message error isolation, so one bad message (e.g. an attachment
+    that fails to decrypt) aborted processing of every message after it in the
+    same batch. Because the cursor had already advanced past the whole batch,
+    the next poll would never re-fetch the unprocessed tail -- silent, permanent
+    loss of every message after the first failure in a batch.
+
+    This test sends a real 3-message batch through the *real* (unstubbed)
+    ``_handle_update`` -- message 1 is fine, message 2 is a WeChat image item
+    whose "encrypted" bytes are deliberately not a multiple of the AES block
+    size, so ``_decrypt_aes_128_ecb`` raises a genuine ``cryptography`` library
+    ``ValueError`` (matching how a real corrupt/undecryptable attachment would
+    fail), and message 3 is fine again. The fix must:
+      - still deliver message 1 and message 3 to the bus despite message 2's
+        failure (per-message isolation instead of one bad apple aborting the
+        for loop), and
+      - log message 2's failure instead of swallowing it silently, and
+      - only advance/persist the cursor once the whole batch has been
+        attempted (so a crash mid-batch re-delivers rather than silently
+        drops).
+    """
+    from app.channels.wechat import WechatChannel
+
+    async def go():
+        bus = MessageBus()
+        published: list[Any] = []
+
+        async def capture(msg):
+            published.append(msg)
+
+        bus.publish_inbound = capture  # type: ignore[method-assign]
+
+        aes_key = b"1234567890abcdef"
+        non_block_aligned_ciphertext = b"\x01\x02\x03\x04\x05"  # 5 bytes: not a multiple of 16
+
+        msg_good_1 = {
+            "message_type": 1,
+            "message_id": "msg-1-good",
+            "from_user_id": "wx-user-1",
+            "context_token": "ctx-1",
+            "item_list": [{"type": 1, "text_item": {"text": "message one"}}],
+        }
+        msg_bad_2 = {
+            "message_type": 1,
+            "message_id": "msg-2-bad",
+            "from_user_id": "wx-user-1",
+            "context_token": "ctx-2",
+            "item_list": [
+                {
+                    "type": 2,
+                    "image_item": {
+                        "aeskey": aes_key.hex(),
+                        "media": {"full_url": "https://cdn.example/corrupt-attachment.bin"},
+                    },
+                }
+            ],
+        }
+        msg_good_3 = {
+            "message_type": 1,
+            "message_id": "msg-3-good",
+            "from_user_id": "wx-user-1",
+            "context_token": "ctx-3",
+            "item_list": [{"type": 1, "text_item": {"text": "message three"}}],
+        }
+
+        state_dir = tmp_path / "wechat-state"
+
+        def _client_factory(*args, **kwargs):
+            return _MockAsyncClient(
+                post_responses=[
+                    {
+                        "ret": 0,
+                        "msgs": [msg_good_1, msg_bad_2, msg_good_3],
+                        "get_updates_buf": "cursor-after-batch",
+                    }
+                ],
+                **kwargs,
+            )
+
+        monkeypatch.setattr("app.channels.wechat.httpx.AsyncClient", _client_factory)
+
+        channel = WechatChannel(
+            bus=bus,
+            config={"bot_token": "test-token", "state_dir": str(state_dir), "polling_retry_delay": 0.001},
+        )
+
+        async def _fake_download(_url: str, *, timeout: float | None = None) -> bytes:
+            return non_block_aligned_ciphertext
+
+        channel._download_cdn_bytes = _fake_download  # type: ignore[method-assign]
+
+        # _handle_update is intentionally left as the REAL implementation so
+        # message 2 hits the genuine cryptography ValueError. To keep the test
+        # deterministic without depending on the bug/fix under test, force the
+        # poll loop to stop after exactly one getupdates cycle via
+        # _ensure_authenticated (called at the top of every iteration, before
+        # any message is processed) rather than via message content.
+        real_ensure_authenticated = channel._ensure_authenticated
+        auth_calls = {"n": 0}
+
+        async def _ensure_auth_then_stop() -> bool:
+            auth_calls["n"] += 1
+            if auth_calls["n"] > 1:
+                channel._running = False
+                return False
+            return await real_ensure_authenticated()
+
+        channel._ensure_authenticated = _ensure_auth_then_stop  # type: ignore[method-assign]
+        channel._running = True
+
+        with caplog.at_level(logging.INFO, logger="app.channels.wechat"):
+            await channel._poll_loop()
+
+        # Message 1 and message 3 must both survive message 2's failure.
+        assert [m.text for m in published] == ["message one", "message three"]
+
+        # Message 2's failure must be logged, not silently swallowed.
+        messages = [record.getMessage() for record in caplog.records]
+        assert any("msg-2-bad" in message for message in messages)
+
+        # The cursor must reflect that the whole batch was attempted -- both
+        # in memory and in the persisted state file used to resume polling.
+        assert channel._get_updates_buf == "cursor-after-batch"
+        persisted = json.loads((state_dir / "wechat-getupdates.json").read_text(encoding="utf-8"))
+        assert persisted["get_updates_buf"] == "cursor-after-batch"
+
+    _run(go())
+
+
 def test_state_cursor_is_loaded_from_disk(tmp_path: Path):
     from app.channels.wechat import WechatChannel