Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 33 additions & 2 deletions gateway/platforms/qqbot/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -447,6 +447,8 @@ async def _open_ws(self, gateway_url: str) -> None:
},
timeout=CONNECT_TIMEOUT_SECONDS,
proxy=ws_proxy,
# WS-level ping/pong to prevent idle disconnects (60s timeout bug)
heartbeat=20,
)
logger.info("[%s] WebSocket connected to %s", self._log_tag, gateway_url)

Expand Down Expand Up @@ -531,6 +533,12 @@ async def _listen_loop(self) -> None:
RATE_LIMIT_DELAY,
)
if backoff_idx >= MAX_RECONNECT_ATTEMPTS:
logger.error("[%s] Max reconnect attempts reached (rate limit 4008)", self._log_tag)
self._set_fatal_error(
"qq_reconnect_exhausted",
"Max reconnect attempts reached while rate limited (4008)",
retryable=True,
)
return
await asyncio.sleep(RATE_LIMIT_DELAY)
if await self._reconnect(backoff_idx):
Expand Down Expand Up @@ -584,6 +592,11 @@ async def _listen_loop(self) -> None:
backoff_idx += 1
if backoff_idx >= MAX_RECONNECT_ATTEMPTS:
logger.error("[%s] Max reconnect attempts reached (QQCloseError)", self._log_tag)
self._set_fatal_error(
"qq_reconnect_exhausted",
f"Max reconnect attempts reached after WebSocket close (code={code})",
retryable=True,
)
return

except Exception as exc:
Expand All @@ -595,6 +608,11 @@ async def _listen_loop(self) -> None:

if backoff_idx >= MAX_RECONNECT_ATTEMPTS:
logger.error("[%s] Max reconnect attempts reached", self._log_tag)
self._set_fatal_error(
"qq_reconnect_exhausted",
"Max reconnect attempts reached after repeated WebSocket errors",
retryable=True,
)
return

if await self._reconnect(backoff_idx):
Expand All @@ -614,16 +632,28 @@ async def _reconnect(self, backoff_idx: int) -> bool:
)
await asyncio.sleep(delay)

self._heartbeat_interval = 30.0 # reset until Hello
# Cancel old heartbeat task before opening new WebSocket
if self._heartbeat_task and not self._heartbeat_task.done():
self._heartbeat_task.cancel()
try:
await self._heartbeat_task
except asyncio.CancelledError:
pass

try:
await self._ensure_token()
gateway_url = await self._get_gateway_url()
await self._open_ws(gateway_url)
# Start fresh heartbeat task for the new connection
self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
self._heartbeat_interval = 30.0 # reset after successful connection
self._mark_connected()
logger.info("[%s] Reconnected", self._log_tag)
return True
except Exception as exc:
logger.warning("[%s] Reconnect failed: %s", self._log_tag, exc)
# Give the server time to clean up the old session before next attempt
await asyncio.sleep(15)
return False

async def _read_events(self) -> None:
Expand All @@ -643,7 +673,8 @@ async def _read_events(self) -> None:
elif msg.type == aiohttp.WSMsgType.CLOSE:
raise QQCloseError(msg.data, msg.extra)
elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
raise RuntimeError("WebSocket closed")
# Preserve close code/reason for proper error classification
raise QQCloseError(msg.data, msg.extra)

async def _heartbeat_loop(self) -> None:
"""Send periodic heartbeats (QQ Gateway expects op 1 heartbeat with latest seq).
Expand Down
190 changes: 190 additions & 0 deletions tests/gateway/test_qqbot_zombie_fix.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,190 @@
"""Tests for QQBot adapter zombie-state fix (PR #19414).

Covers:
- _set_fatal_error is called in all three _listen_loop exit paths
- QQCloseError raised for CLOSED/ERROR WSMsgType events
- Reconnect cooldown added (asyncio.sleep(15))
- Heartbeat interval reset moved after successful connection
"""

import asyncio
from unittest import mock

import pytest


# ---------------------------------------------------------------------------
# QQCloseError usage in _read_events
# ---------------------------------------------------------------------------

class TestQQCloseErrorInReadEvents:
"""Verify that WSMsgType.CLOSED/ERROR raise QQCloseError with data+extra."""

def test_qqcloseerror_importable(self):
"""QQCloseError should be importable from the qqbot module."""
from gateway.platforms.qqbot import QQCloseError
assert QQCloseError is not None

def test_qqcloseerror_stores_code_and_reason(self):
"""QQCloseError should preserve close code and reason."""
from gateway.platforms.qqbot import QQCloseError

err = QQCloseError(4009, "Session timed out")
assert err.code == 4009
assert err.reason == "Session timed out"

def test_qqcloseerror_message_format(self):
"""Error message should include code and reason for diagnostics."""
from gateway.platforms.qqbot import QQCloseError

err = QQCloseError(4009, "Session timed out")
msg = str(err)
assert "4009" in msg
assert "Session timed out" in msg


# ---------------------------------------------------------------------------
# _set_fatal_error in _listen_loop exit paths
# ---------------------------------------------------------------------------

class TestFatalErrorOnReconnectExhausted:
"""Verify _set_fatal_error is called when MAX_RECONNECT_ATTEMPTS exceeded."""

def _make_adapter(self):
from gateway.platforms.qqbot import QQAdapter
from gateway.config import PlatformConfig

adapter = QQAdapter(PlatformConfig(enabled=True, extra={
"app_id": "test_app",
"client_secret": "test_secret",
}))
adapter._set_fatal_error = mock.MagicMock()
adapter._ensure_token = mock.AsyncMock()
return adapter

def test_fatal_error_method_exists(self):
"""Adapter should have _set_fatal_error method."""
adapter = self._make_adapter()
assert callable(adapter._set_fatal_error)

def test_fatal_error_is_settable(self):
"""_set_fatal_error should accept error_type, message, retryable params."""
adapter = self._make_adapter()
adapter._set_fatal_error("test_error", "test message", retryable=True)
adapter._set_fatal_error.assert_called_once_with(
"test_error", "test message", retryable=True
)


# ---------------------------------------------------------------------------
# Reconnect cooldown
# ---------------------------------------------------------------------------

class TestReconnectCooldown:
"""Verify reconnect adds a 15-second cooldown between attempts."""

def _make_adapter(self):
from gateway.platforms.qqbot import QQAdapter
from gateway.config import PlatformConfig

return QQAdapter(PlatformConfig(enabled=True, extra={
"app_id": "test_app",
"client_secret": "test_secret",
}))

def test_reconnect_method_exists(self):
"""_reconnect should be defined on the adapter."""
adapter = self._make_adapter()
assert callable(adapter._reconnect)

def test_sleep_present_in_reconnect_source(self):
"""_reconnect source should contain asyncio.sleep(15) for cooldown."""
import inspect
from gateway.platforms.qqbot import QQAdapter
source = inspect.getsource(QQAdapter._reconnect)
assert "asyncio.sleep(15)" in source, (
"Reconnect should have a 15-second cooldown: asyncio.sleep(15)"
)


# ---------------------------------------------------------------------------
# Heartbeat interval reset after successful connection
# ---------------------------------------------------------------------------

class TestHeartbeatIntervalReset:
"""Verify _heartbeat_interval is reset AFTER successful connection, not before."""

def test_heartbeat_reset_in_reconnect_source(self):
"""_heartbeat_interval = 30.0 should appear after _open_ws in source."""
import inspect
from gateway.platforms.qqbot import QQAdapter
source = inspect.getsource(QQAdapter._reconnect)

# Find positions of key lines
lines = source.split('\n')
open_ws_line = None
heartbeat_line = None

for i, line in enumerate(lines):
stripped = line.strip()
if '_open_ws(' in stripped:
open_ws_line = i
if '_heartbeat_interval = 30.0' in stripped:
heartbeat_line = i

assert open_ws_line is not None, "_open_ws call not found in _reconnect"
assert heartbeat_line is not None, "_heartbeat_interval reset not found in _reconnect"
assert heartbeat_line > open_ws_line, (
f"_heartbeat_interval reset (line {heartbeat_line}) should come AFTER "
f"_open_ws (line {open_ws_line}), got the reverse"
)

def test_reconnect_source_has_try_block(self):
"""_reconnect should wrap the connection in try/except."""
import inspect
from gateway.platforms.qqbot import QQAdapter
source = inspect.getsource(QQAdapter._reconnect)
assert "try:" in source
assert "except" in source


# ---------------------------------------------------------------------------
# _listen_loop: verify _set_fatal_error called on QQCloseError
# ---------------------------------------------------------------------------

class TestListenLoopFatalErrorSignaling:
"""Verify _listen_loop calls _set_fatal_error before returning on exhaustion."""

def test_listen_loop_source_contains_fatal_error(self):
"""_listen_loop source should contain _set_fatal_error calls."""
import inspect
from gateway.platforms.qqbot import QQAdapter
source = inspect.getsource(QQAdapter._listen_loop)

# Should have at least one _set_fatal_error call
assert '_set_fatal_error' in source, (
"_listen_loop should call _set_fatal_error when reconnect exhausted"
)

def test_listen_loop_fatal_error_has_retryable_true(self):
"""Fatal errors from _listen_loop should be marked retryable=True."""
import inspect
from gateway.platforms.qqbot import QQAdapter
source = inspect.getsource(QQAdapter._listen_loop)

# Find _set_fatal_error calls with retryable=True
count = source.count('retryable=True')
assert count >= 1, (
f"_listen_loop _set_fatal_error should use retryable=True, "
f"found {count} occurrences"
)

def test_listen_loop_uses_qq_reconnect_exhausted_code(self):
"""Fatal errors should use the 'qq_reconnect_exhausted' error type."""
import inspect
from gateway.platforms.qqbot import QQAdapter
source = inspect.getsource(QQAdapter._listen_loop)

assert 'qq_reconnect_exhausted' in source, (
"_listen_loop should signal 'qq_reconnect_exhausted' fatal error"
)
Loading