diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index 68a0e0a9e1b19..4b488efb1ab3c 100644 --- a/gateway/platforms/weixin.py +++ b/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 diff --git a/tests/gateway/test_weixin.py b/tests/gateway/test_weixin.py index ed20aeb8f351f..e028fc0f39941 100644 --- a/tests/gateway/test_weixin.py +++ b/tests/gateway/test_weixin.py @@ -139,6 +139,44 @@ def test_get_connected_platforms_includes_weixin_with_token(self): assert config.get_connected_platforms() == [Platform.WEIXIN] + def test_get_connected_platforms_requires_account_id(self): + config = GatewayConfig( + platforms={ + Platform.WEIXIN: PlatformConfig( + enabled=True, + token="bot-token", + ) + } + ) + + assert config.get_connected_platforms() == [] + + def test_explicit_zero_send_chunk_retries_overrides_environment(self, monkeypatch): + monkeypatch.setenv("WEIXIN_SEND_CHUNK_RETRIES", "7") + + adapter = WeixinAdapter( + PlatformConfig( + enabled=True, + token="bot-token", + extra={"account_id": "bot-account", "send_chunk_retries": 0}, + ) + ) + + assert adapter._send_chunk_retries == 0 + + def test_missing_send_chunk_retries_uses_environment(self, monkeypatch): + monkeypatch.setenv("WEIXIN_SEND_CHUNK_RETRIES", "7") + + adapter = WeixinAdapter( + PlatformConfig( + enabled=True, + token="bot-token", + extra={"account_id": "bot-account"}, + ) + ) + + assert adapter._send_chunk_retries == 7 + class TestWeixinStatePersistence: def test_save_weixin_account_preserves_existing_file_on_replace_failure(self, tmp_path, monkeypatch): @@ -167,6 +205,61 @@ def _boom(_src, _dst): assert json.loads(account_path.read_text(encoding="utf-8")) == original + def test_context_token_persist_preserves_existing_file_on_replace_failure(self, tmp_path, monkeypatch): + token_path = tmp_path / "weixin" / "accounts" / "acct.context-tokens.json" + token_path.parent.mkdir(parents=True, exist_ok=True) + token_path.write_text(json.dumps({"user-a": "old-token"}), encoding="utf-8") + + def _boom(_src, _dst): + raise OSError("disk full") + + monkeypatch.setattr("utils.os.replace", _boom) + + store = ContextTokenStore(str(tmp_path)) + with patch.object(weixin.logger, "warning") as warning_mock: + store.set("acct", "user-b", "new-token") + + 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_only_selected_account_peer(self, tmp_path): + store = ContextTokenStore(str(tmp_path)) + store.set("acct-a", "peer-a", "token-a") + store.set("acct-a", "peer-b", "token-b") + store.set("acct-b", "peer-a", "token-c") + + store.delete("acct-a", "peer-a", "token-a") + + assert store.get("acct-a", "peer-a") is None + assert store.get("acct-a", "peer-b") == "token-b" + assert store.get("acct-b", "peer-a") == "token-c" + + restored = ContextTokenStore(str(tmp_path)) + restored.restore("acct-a") + restored.restore("acct-b") + assert restored.get("acct-a", "peer-a") is None + assert restored.get("acct-a", "peer-b") == "token-b" + assert restored.get("acct-b", "peer-a") == "token-c" + + 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) + sync_path.write_text(json.dumps({"get_updates_buf": "old-sync"}), encoding="utf-8") + + def _boom(_src, _dst): + raise OSError("disk full") + + monkeypatch.setattr("utils.os.replace", _boom) + + try: + weixin._save_sync_buf(str(tmp_path), "acct", "new-sync") + except OSError: + pass + else: + raise AssertionError("expected _save_sync_buf to propagate replace failure") + + assert json.loads(sync_path.read_text(encoding="utf-8")) == {"get_updates_buf": "old-sync"} + class TestWeixinQrLogin: @pytest.mark.asyncio @@ -241,6 +334,183 @@ 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_stale_context_token_recovers_with_zero_retry_budget( + self, + send_message_mock, + tmp_path, + ): + adapter = self._connected_adapter() + adapter._send_chunk_retries = 0 + adapter._token_store = weixin.ContextTokenStore(str(tmp_path)) + adapter._token_store.set(adapter._account_id, "wxid_test123", "ctx-token") + send_message_mock.side_effect = [ + {"ret": -2, "errmsg": "prepare failed"}, + {"ret": 0}, + ] + + result = asyncio.run(adapter.send("wxid_test123", "hello")) + + assert result.success is True + assert [ + call.kwargs["context_token"] + for call in send_message_mock.await_args_list + ] == ["ctx-token", None] + assert adapter._token_store.get(adapter._account_id, "wxid_test123") is None + + restored_store = weixin.ContextTokenStore(str(tmp_path)) + restored_store.restore(adapter._account_id) + assert restored_store.get(adapter._account_id, "wxid_test123") is None + + @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock) + def test_stale_context_token_is_not_reused_for_later_chunks( + self, + send_message_mock, + tmp_path, + ): + adapter = self._connected_adapter() + adapter.MAX_MESSAGE_LENGTH = 12 + adapter._send_chunk_delay_seconds = 0 + adapter._token_store = weixin.ContextTokenStore(str(tmp_path)) + adapter._token_store.set(adapter._account_id, "wxid_test123", "ctx-token") + send_message_mock.side_effect = [ + {"ret": -2, "errmsg": "prepare failed"}, + {"ret": 0}, + {"ret": 0}, + ] + + result = asyncio.run(adapter.send("wxid_test123", "first\n\nsecond")) + + assert result.success is True + assert [ + call.kwargs["context_token"] + for call in send_message_mock.await_args_list + ] == ["ctx-token", None, None] + + @patch("gateway.platforms.weixin.asyncio.sleep", new_callable=AsyncMock) + @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock) + def test_repeated_stale_response_does_not_trigger_rate_limit_handling( + self, + send_message_mock, + sleep_mock, + tmp_path, + ): + adapter = self._connected_adapter() + adapter._token_store = ContextTokenStore(str(tmp_path)) + adapter._token_store.set(adapter._account_id, "wxid_test123", "ctx-token") + send_message_mock.side_effect = [ + {"ret": -2, "errmsg": "prepare failed"}, + {"ret": -2, "errmsg": "prepare failed"}, + ] + + result = asyncio.run(adapter.send("wxid_test123", "hello")) + + assert result.success is False + assert "stale session" in (result.error or "") + assert [ + call.kwargs["context_token"] + for call in send_message_mock.await_args_list + ] == ["ctx-token", None] + assert adapter._rate_limit_events == [] + assert adapter._rate_limit_circuit_until == 0.0 + sleep_mock.assert_not_awaited() + + @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock) + def test_failed_stale_recovery_does_not_restart_with_original_token( + self, + send_message_mock, + tmp_path, + ): + adapter = self._connected_adapter() + adapter._send_chunk_retries = 1 + adapter._send_chunk_retry_delay_seconds = 0 + adapter._token_store = weixin.ContextTokenStore(str(tmp_path)) + adapter._token_store.set(adapter._account_id, "wxid_test123", "ctx-token") + send_message_mock.side_effect = [ + {"ret": -2, "errmsg": "prepare failed"}, + RuntimeError("tokenless recovery failed"), + RuntimeError("tokenless retry failed"), + {"ret": 0}, + ] + + result = asyncio.run(adapter.send("wxid_test123", "hello")) + + assert result.success is False + assert [ + call.kwargs["context_token"] + for call in send_message_mock.await_args_list + ] == ["ctx-token", None, None] + + @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock) + def test_stale_response_does_not_delete_newer_context_token( + self, + send_message_mock, + tmp_path, + ): + adapter = self._connected_adapter() + adapter._send_chunk_retries = 0 + adapter._token_store = weixin.ContextTokenStore(str(tmp_path)) + adapter._token_store.set(adapter._account_id, "wxid_test123", "ctx-token") + + async def refresh_before_stale_response(*args, **kwargs): + if kwargs["context_token"] == "ctx-token": + adapter._token_store.set( + adapter._account_id, + "wxid_test123", + "fresh-token", + ) + return {"ret": -2, "errmsg": "prepare failed"} + return {"ret": 0} + + send_message_mock.side_effect = refresh_before_stale_response + + result = asyncio.run(adapter.send("wxid_test123", "hello")) + + assert result.success is True + assert adapter._token_store.get( + adapter._account_id, + "wxid_test123", + ) == "fresh-token" + + restored_store = weixin.ContextTokenStore(str(tmp_path)) + restored_store.restore(adapter._account_id) + assert restored_store.get( + adapter._account_id, + "wxid_test123", + ) == "fresh-token" + + @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock) + def test_context_token_is_loaded_after_send_gate_is_acquired( + self, + send_message_mock, + tmp_path, + ): + adapter = self._connected_adapter() + adapter._token_store = weixin.ContextTokenStore(str(tmp_path)) + adapter._token_store.set(adapter._account_id, "wxid_test123", "old-token") + send_message_mock.return_value = {"ret": 0} + + async def send_after_refresh(): + await adapter._send_text_gate.acquire() + try: + send_task = asyncio.create_task( + adapter.send("wxid_test123", "hello") + ) + await asyncio.sleep(0) + adapter._token_store.set( + adapter._account_id, + "wxid_test123", + "fresh-token", + ) + finally: + adapter._send_text_gate.release() + return await send_task + + result = asyncio.run(send_after_refresh()) + + assert result.success is True + assert send_message_mock.await_args.kwargs["context_token"] == "fresh-token" + @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): @@ -500,6 +770,22 @@ class TestIsStaleSessionRet: """Regression test for #17228: distinguish stale-session ret=-2 from rate-limit ret=-2.""" + def test_prepare_failed_is_only_an_outbound_context_token_signal(self): + assert weixin._is_stale_session_ret(-2, None, "prepare failed") is False + assert weixin._is_stale_context_token_ret( + -2, + None, + "prepare failed", + ) is True + + + def test_prepare_failed_with_case_and_whitespace_is_stale_context_token(self): + assert weixin._is_stale_context_token_ret( + -2, + None, + " Prepare Failed ", + ) is True + 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 @@ -511,6 +797,30 @@ def test_errcode_minus_14_is_not_matched_here(self): assert weixin._is_stale_session_ret(-14, None, "session expired") is False +class TestContextTokenStore: + def test_delete_removes_only_matching_peer_and_persists(self, tmp_path): + store = weixin.ContextTokenStore(str(tmp_path)) + store.set("account-a", "peer-a", "stale-token") + store.set("account-a", "peer-b", "fresh-token") + store.set("account-b", "peer-a", "other-account-token") + + assert store.delete("account-a", "peer-a", "replacement-token") is False + assert store.get("account-a", "peer-a") == "stale-token" + + assert store.delete("account-a", "peer-a", "stale-token") is True + + assert store.get("account-a", "peer-a") is None + assert store.get("account-a", "peer-b") == "fresh-token" + assert store.get("account-b", "peer-a") == "other-account-token" + + restored_store = weixin.ContextTokenStore(str(tmp_path)) + restored_store.restore("account-a") + restored_store.restore("account-b") + assert restored_store.get("account-a", "peer-a") is None + assert restored_store.get("account-a", "peer-b") == "fresh-token" + assert restored_store.get("account-b", "peer-a") == "other-account-token" + + class TestWeixinContentDedup: """Regression tests for Issue #16182 — upstream API sends duplicate content with different message_ids, bypassing message_id deduplication.