fix(weixin): recover from ret=-2 with empty stale-context errors

gateway/platforms/weixin.py

@@ -93,19 +93,24 @@
 RETRY_DELAY_SECONDS = 2
 BACKOFF_DELAY_SECONDS = 30
 SESSION_EXPIRED_ERRCODE = -14
-RATE_LIMIT_ERRCODE = -2  # iLink frequency limit — backoff and retry
+RATE_LIMIT_ERRCODE = -2  # iLink overloads -2 for rate limits and stale/invalid context
 MESSAGE_DEDUP_TTL_SECONDS = 300
 
 
 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."""
+    """Return whether ``-2`` is iLink's stale-context variant.
+
+    Cron pushes to an inactive peer can return ``ret=-2`` with either an empty
+    error message or ``"unknown error"``.  Both recover when the saved
+    ``context_token`` is removed and the message is retried tokenless.  An
+    explicit frequency/rate-limit message must continue through the normal
+    backoff path instead.
+    """
     if ret != RATE_LIMIT_ERRCODE and errcode != RATE_LIMIT_ERRCODE:
         return False
-    return (errmsg or "").lower() == "unknown error"
+    return (errmsg or "").strip().lower() in {"", "unknown error"}
 
 
 MEDIA_IMAGE = 1
@@ -314,6 +319,11 @@ 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) -> None:
+        """Remove a stale peer token from memory and durable storage."""
+        self._cache.pop(self._key(account_id, user_id), None)
+        self._persist(account_id)
+
     def _persist(self, account_id: str) -> None:
         prefix = f"{account_id}:"
         payload = {
@@ -1812,12 +1822,12 @@ async def _send_text_chunk_locked(
                         if is_session_expired and not retried_without_token and context_token:
                             retried_without_token = True
                             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)
                             logger.warning(
-                                "[%s] session expired for %s; retrying without context_token",
-                                self.name, _safe_id(chat_id),
+                                "[%s] stale context_token for %s "
+                                "(ret=%s errcode=%s errmsg=%r); retrying tokenless",
+                                self.name, _safe_id(chat_id), ret, errcode,
+                                resp.get("errmsg"),
                             )
                             continue
                         # Rate limit (-2) — backoff and retry

tests/gateway/test_weixin.py

@@ -318,6 +318,20 @@ def _boom(_src, _dst):
         assert json.loads(token_path.read_text(encoding="utf-8")) == {"user-a": "old-token"}
         warning_mock.assert_called_once()
 
+    def test_context_token_delete_removes_memory_and_disk_entry(self, tmp_path):
+        store = ContextTokenStore(str(tmp_path))
+        store.set("acct", "user-a", "stale-token")
+        store.set("acct", "user-b", "fresh-token")
+
+        store.delete("acct", "user-a")
+
+        assert store.get("acct", "user-a") is None
+        assert store.get("acct", "user-b") == "fresh-token"
+        token_path = tmp_path / "weixin" / "accounts" / "acct.context-tokens.json"
+        assert json.loads(token_path.read_text(encoding="utf-8")) == {
+            "user-b": "fresh-token"
+        }
+
     def test_save_sync_buf_preserves_existing_file_on_replace_failure(self, tmp_path, monkeypatch):
         sync_path = tmp_path / "weixin" / "accounts" / "acct.sync.json"
         sync_path.parent.mkdir(parents=True, exist_ok=True)
@@ -446,6 +460,32 @@ async def flaky_send(*args, **kwargs):
         assert first_try["text"] == retry["text"]
         assert first_try["client_id"] == retry["client_id"]
 
+    @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock)
+    def test_empty_errmsg_minus_2_retries_without_stale_context_token(
+        self, send_message_mock, tmp_path
+    ):
+        adapter = self._connected_adapter()
+        token_store = ContextTokenStore(str(tmp_path))
+        token_store.set("test-account", "wxid_test123", "stale-ctx-token")
+        adapter._token_store = token_store
+        send_message_mock.side_effect = [
+            {"ret": -2, "errcode": None, "errmsg": None},
+            {"ret": 0, "errcode": 0},
+        ]
+
+        result = asyncio.run(adapter.send("wxid_test123", "scheduled journal"))
+
+        assert result.success is True
+        assert send_message_mock.await_count == 2
+        first_attempt = send_message_mock.await_args_list[0].kwargs
+        tokenless_retry = send_message_mock.await_args_list[1].kwargs
+        assert first_attempt["context_token"] == "stale-ctx-token"
+        assert tokenless_retry["context_token"] is None
+        assert token_store.get("test-account", "wxid_test123") is None
+        token_path = tmp_path / "weixin" / "accounts" / "test-account.context-tokens.json"
+        assert json.loads(token_path.read_text(encoding="utf-8")) == {}
+        assert adapter._rate_limit_events == []
+
     @patch("gateway.platforms.weixin.asyncio.sleep", new_callable=AsyncMock)
     @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock)
     def test_repeated_rate_limits_open_circuit_for_followup_sends(self, send_message_mock, sleep_mock):
@@ -958,9 +998,9 @@ def test_ret_minus_2_with_freq_limit_is_not_stale(self):
         # Genuine rate limit — must NOT be treated as stale session.
         assert weixin._is_stale_session_ret(-2, None, "freq limit") is False
 
-    def test_ret_minus_2_with_no_errmsg_is_not_stale(self):
-        assert weixin._is_stale_session_ret(-2, None, None) is False
-        assert weixin._is_stale_session_ret(-2, None, "") is False
+    def test_ret_minus_2_with_no_errmsg_is_stale(self):
+        assert weixin._is_stale_session_ret(-2, None, None) is True
+        assert weixin._is_stale_session_ret(-2, None, "") is True
 
     def test_errcode_minus_14_is_not_matched_here(self):
         # -14 is handled by the separate SESSION_EXPIRED_ERRCODE path; the