diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py index 12caef0f1449..0abe156b3df9 100644 --- a/gateway/platforms/qqbot/adapter.py +++ b/gateway/platforms/qqbot/adapter.py @@ -100,6 +100,7 @@ def __init__(self, code, reason=""): CONNECT_TIMEOUT_SECONDS, RECONNECT_BACKOFF, MAX_RECONNECT_ATTEMPTS, + MAX_RESUME_ATTEMPTS, RATE_LIMIT_DELAY, QUICK_DISCONNECT_THRESHOLD, MAX_QUICK_DISCONNECT_COUNT, @@ -205,6 +206,7 @@ def __init__(self, config: PlatformConfig): self._heartbeat_interval: float = 30.0 # seconds, updated by Hello self._session_id: Optional[str] = None self._last_seq: Optional[int] = None + self._resume_attempts: int = 0 self._chat_type_map: Dict[str, str] = {} # chat_id → "c2c"|"group"|"guild"|"dm" # Request/response correlation @@ -576,6 +578,7 @@ async def _listen_loop(self) -> None: ) self._session_id = None self._last_seq = None + self._resume_attempts = 0 if await self._reconnect(backoff_idx): backoff_idx = 0 @@ -734,6 +737,7 @@ async def _send_resume(self) -> None: # If resume fails, clear session and fall back to identify on next Hello self._session_id = None self._last_seq = None + self._resume_attempts = 0 @staticmethod def _create_task(coro): @@ -772,8 +776,21 @@ def _dispatch_payload(self, payload: Dict[str, Any]) -> None: # Authenticate: send Resume if we have a session, else Identify. # Use _create_task which is safe when no event loop is running (tests). if self._session_id and self._last_seq is not None: - self._create_task(self._send_resume()) + if self._resume_attempts >= MAX_RESUME_ATTEMPTS: + logger.warning( + "[%s] Resume attempt cap reached (%d). Falling back to Identify.", + self._log_tag, + MAX_RESUME_ATTEMPTS, + ) + self._session_id = None + self._last_seq = None + self._resume_attempts = 0 + self._create_task(self._send_identify()) + else: + self._resume_attempts += 1 + self._create_task(self._send_resume()) else: + self._resume_attempts = 0 self._create_task(self._send_identify()) return @@ -782,6 +799,7 @@ def _dispatch_payload(self, payload: Dict[str, Any]) -> None: if t == "READY": self._handle_ready(d) elif t == "RESUMED": + self._resume_attempts = 0 logger.info("[%s] Session resumed", self._log_tag) elif t in ( "C2C_MESSAGE_CREATE", @@ -807,6 +825,7 @@ def _handle_ready(self, d: Any) -> None: """Handle the READY event — store session_id for resume.""" if isinstance(d, dict): self._session_id = d.get("session_id") + self._resume_attempts = 0 logger.info("[%s] Ready, session_id=%s", self._log_tag, self._session_id) # ------------------------------------------------------------------ diff --git a/gateway/platforms/qqbot/constants.py b/gateway/platforms/qqbot/constants.py index ddae3c133edc..5ef0c69ccc8e 100644 --- a/gateway/platforms/qqbot/constants.py +++ b/gateway/platforms/qqbot/constants.py @@ -40,6 +40,7 @@ RECONNECT_BACKOFF = [2, 5, 10, 30, 60] MAX_RECONNECT_ATTEMPTS = 100 +MAX_RESUME_ATTEMPTS = 3 RATE_LIMIT_DELAY = 60 # seconds QUICK_DISCONNECT_THRESHOLD = 5.0 # seconds MAX_QUICK_DISCONNECT_COUNT = 3 diff --git a/tests/gateway/test_qqbot.py b/tests/gateway/test_qqbot.py index a0c9fa6573cd..ff7eda75175c 100644 --- a/tests/gateway/test_qqbot.py +++ b/tests/gateway/test_qqbot.py @@ -478,6 +478,53 @@ def test_resumed_preserves_session(self): assert adapter._last_seq == 60 +class TestHelloAuthModeSelection: + def _make_adapter(self, **extra): + from gateway.platforms.qqbot import QQAdapter + return QQAdapter(_make_config(**extra)) + + def test_hello_uses_resume_before_cap(self): + adapter = self._make_adapter(app_id="a", client_secret="b") + adapter._session_id = "sess_1" + adapter._last_seq = 10 + + scheduled = [] + adapter._create_task = lambda coro: scheduled.append(coro) # type: ignore[assignment] + adapter._send_resume = mock.AsyncMock(name="send_resume") + adapter._send_identify = mock.AsyncMock(name="send_identify") + + adapter._dispatch_payload({"op": 10, "d": {"heartbeat_interval": 30000}}) + + assert len(scheduled) == 1 + assert adapter._resume_attempts == 1 + adapter._send_resume.assert_called_once() + adapter._send_identify.assert_not_called() + scheduled[0].close() + + def test_hello_falls_back_to_identify_when_resume_cap_reached(self): + from gateway.platforms.qqbot.constants import MAX_RESUME_ATTEMPTS + + adapter = self._make_adapter(app_id="a", client_secret="b") + adapter._session_id = "stale_sess" + adapter._last_seq = 99 + adapter._resume_attempts = MAX_RESUME_ATTEMPTS + + scheduled = [] + adapter._create_task = lambda coro: scheduled.append(coro) # type: ignore[assignment] + adapter._send_resume = mock.AsyncMock(name="send_resume") + adapter._send_identify = mock.AsyncMock(name="send_identify") + + adapter._dispatch_payload({"op": 10, "d": {"heartbeat_interval": 30000}}) + + assert len(scheduled) == 1 + assert adapter._session_id is None + assert adapter._last_seq is None + assert adapter._resume_attempts == 0 + adapter._send_resume.assert_not_called() + adapter._send_identify.assert_called_once() + scheduled[0].close() + + # --------------------------------------------------------------------------- # _parse_json # ---------------------------------------------------------------------------