fix(gateway): recover from stale Weixin context tokens

gateway/platforms/weixin.py

@@ -118,12 +118,21 @@ def _wx_secret(name: str, default: Optional[str] = None) -> Optional[str]:
 def _is_stale_session_ret(
     ret: "Optional[int]", errcode: "Optional[int]", errmsg: "Optional[str]",
 ) -> bool:
-    """True when iLink returns ret=-2 / errcode=-2 with 'unknown error',
-    which is a stale-session signal (same as errcode=-14) rather than
-    a genuine rate limit."""
+    """True when iLink returns the known stale-session ret/errcode -2."""
     if ret != RATE_LIMIT_ERRCODE and errcode != RATE_LIMIT_ERRCODE:
         return False
-    return (errmsg or "").lower() == "unknown error"
+    return (errmsg or "").strip().lower() == "unknown error"
+
+
+def _is_stale_context_token_ret(
+    ret: "Optional[int]", errcode: "Optional[int]", errmsg: "Optional[str]",
+) -> bool:
+    """True when an outbound send reports a stale ``context_token``."""
+    if _is_stale_session_ret(ret, errcode, errmsg):
+        return True
+    if ret != RATE_LIMIT_ERRCODE and errcode != RATE_LIMIT_ERRCODE:
+        return False
+    return (errmsg or "").strip().lower() == "prepare failed"
 
 
 MEDIA_IMAGE = 1
@@ -332,6 +341,14 @@ def set(self, account_id: str, user_id: str, token: str) -> None:
         self._cache[self._key(account_id, user_id)] = token
         self._persist(account_id)
 
+    def delete(self, account_id: str, user_id: str, expected_token: str) -> bool:
+        key = self._key(account_id, user_id)
+        if self._cache.get(key) != expected_token:
+            return False
+        self._cache.pop(key, None)
+        self._persist(account_id)
+        return True
+
     def _persist(self, account_id: str) -> None:
         prefix = f"{account_id}:"
         payload = {
@@ -1202,9 +1219,10 @@ def __init__(self, config: PlatformConfig):
         self._send_chunk_delay_seconds = float(
             extra.get("send_chunk_delay_seconds") or os.getenv("WEIXIN_SEND_CHUNK_DELAY_SECONDS", "1.5")
         )
-        self._send_chunk_retries = int(
-            extra.get("send_chunk_retries") or os.getenv("WEIXIN_SEND_CHUNK_RETRIES", "4")
-        )
+        send_chunk_retries = extra.get("send_chunk_retries")
+        if send_chunk_retries is None:
+            send_chunk_retries = os.getenv("WEIXIN_SEND_CHUNK_RETRIES", "4")
+        self._send_chunk_retries = int(send_chunk_retries)
         self._send_chunk_retry_delay_seconds = float(
             extra.get("send_chunk_retry_delay_seconds")
             or os.getenv("WEIXIN_SEND_CHUNK_RETRY_DELAY_SECONDS", "1.0")
@@ -1762,17 +1780,16 @@ async def _send_text_chunk(
         *,
         chat_id: str,
         chunk: str,
-        context_token: Optional[str],
         client_id: str,
     ) -> None:
         """Send a single text chunk with per-chunk retry and backoff.
 
-        On session-expired errors (errcode -14), automatically retries
-        *without* ``context_token`` — iLink accepts tokenless sends as a
-        degraded fallback, which keeps cron-initiated push messages working
-        even when no user message has refreshed the session recently.
+        On session-expired or stale-context errors, automatically retries
+        *without* ``context_token``.  The recovery send does not consume the
+        configured transient retry budget.
         """
         async with self._send_text_gate:
+            context_token = self._token_store.get(self._account_id, chat_id)
             await self._send_text_chunk_locked(
                 chat_id=chat_id,
                 chunk=chunk,
@@ -1790,8 +1807,8 @@ async def _send_text_chunk_locked(
     ) -> None:
         """Send a text chunk while holding the adapter-wide outbound text gate."""
         last_error: Optional[Exception] = None
-        retried_without_token = False
-        for attempt in range(self._send_chunk_retries + 1):
+        attempt = 0
+        while attempt <= self._send_chunk_retries:
             if self._rate_limit_cooldown_remaining() > 0:
                 raise self._rate_limit_error()
             try:
@@ -1809,23 +1826,35 @@ async def _send_text_chunk_locked(
                     ret = resp.get("ret")
                     errcode = resp.get("errcode")
                     if (ret is not None and ret not in {0,}) or (errcode is not None and errcode not in {0,}):
+                        is_stale_context_token = _is_stale_context_token_ret(
+                            ret, errcode, resp.get("errmsg")
+                        )
                         is_session_expired = (
                             ret == SESSION_EXPIRED_ERRCODE
                             or errcode == SESSION_EXPIRED_ERRCODE
-                            or _is_stale_session_ret(ret, errcode, resp.get("errmsg"))
+                            or is_stale_context_token
                         )
                         # Session expired — strip token and retry once
-                        if is_session_expired and not retried_without_token and context_token:
-                            retried_without_token = True
+                        if is_session_expired and context_token:
+                            stale_token = context_token
                             context_token = None
-                            self._token_store._cache.pop(
-                                self._token_store._key(self._account_id, chat_id), None
+                            self._token_store.delete(
+                                self._account_id,
+                                chat_id,
+                                stale_token,
                             )
                             logger.warning(
                                 "[%s] session expired for %s; retrying without context_token",
                                 self.name, _safe_id(chat_id),
                             )
                             continue
+                        if is_stale_context_token:
+                            errmsg = resp.get("errmsg") or resp.get("msg") or "unknown error"
+                            last_error = RuntimeError(
+                                f"iLink sendmessage stale session: "
+                                f"ret={ret} errcode={errcode} errmsg={errmsg}"
+                            )
+                            break
                         # Rate limit (-2) — backoff and retry
                         is_rate_limited = (
                             ret == RATE_LIMIT_ERRCODE
@@ -1850,6 +1879,7 @@ async def _send_text_chunk_locked(
                                 self.name, _safe_id(chat_id), wait,
                             )
                             await asyncio.sleep(wait)
+                            attempt += 1
                             continue
                         errmsg = resp.get("errmsg") or resp.get("msg") or "unknown error"
                         raise RuntimeError(
@@ -1873,6 +1903,7 @@ async def _send_text_chunk_locked(
                 )
                 if wait > 0:
                     await asyncio.sleep(wait)
+                attempt += 1
         assert last_error is not None
         raise last_error
 
@@ -1885,7 +1916,6 @@ async def send(
     ) -> SendResult:
         if not self._send_session or not self._token:
             return SendResult(success=False, error="Not connected")
-        context_token = self._token_store.get(self._account_id, chat_id)
         last_message_id: Optional[str] = None
 
         # Extract MEDIA: tags and bare local file paths before text delivery.
@@ -1932,7 +1962,6 @@ async def _deliver_media(path: str, is_voice: bool = False) -> None:
                 await self._send_text_chunk(
                     chat_id=chat_id,
                     chunk=chunk,
-                    context_token=context_token,
                     client_id=client_id,
                 )
                 last_message_id = client_id