fix(qqbot): converge heartbeat lifecycle

app/modules/qqbot/gateway.py

@@ -39,27 +39,116 @@ def run_gateway(
     :param ws_holder: 调用方持有的单元素列表，存放当前 WebSocketApp，供 stop() 时 close 以打断 run_forever
     """
     last_seq: Optional[int] = None
-    heartbeat_interval_ms: Optional[int] = None
-    heartbeat_timer: Optional[threading.Timer] = None
+    heartbeat_thread: Optional[threading.Thread] = None
+    heartbeat_stop_event: Optional[threading.Event] = None
+    heartbeat_epoch = 0
+    heartbeat_state_lock = threading.Lock()
+    heartbeat_replace_lock = threading.RLock()
 
-    def send_heartbeat():
-        nonlocal heartbeat_timer
-        if stop_event.is_set():
+    def heartbeat_loop(
+        generation_epoch: int,
+        generation_stop_event: threading.Event,
+        interval_seconds: float,
+        ws: websocket.WebSocketApp,
+    ) -> None:
+        """在当前连接 generation 内串行发送心跳，停止后不再派生新线程。"""
+        while not generation_stop_event.wait(interval_seconds):
+            if stop_event.is_set():
+                return
+            with heartbeat_state_lock:
+                if (
+                    generation_epoch != heartbeat_epoch
+                    or heartbeat_stop_event is not generation_stop_event
+                ):
+                    return
+                sequence = last_seq
+            try:
+                payload = {"op": 1, "d": sequence}
+                ws.send(json.dumps(payload))
+                logger.debug(
+                    f"[QQ Gateway:{config_name}] Heartbeat sent, seq={sequence}"
+                )
+            except Exception as err:
+                logger.debug(f"[QQ Gateway:{config_name}] Heartbeat error: {err}")
+
+    def invalidate_heartbeat_generation() -> tuple[Optional[threading.Thread], int]:
+        """锁内失效当前 generation，并返回待等待的 owner 与新 epoch。"""
+        nonlocal heartbeat_epoch
+        with heartbeat_state_lock:
+            heartbeat_epoch += 1
+            current_thread = heartbeat_thread
+            if heartbeat_stop_event is not None:
+                heartbeat_stop_event.set()
+            return current_thread, heartbeat_epoch
+
+    def clear_heartbeat_generation(current_thread: Optional[threading.Thread]) -> None:
+        """仅在快照仍是当前 owner 且已经终止时清除 generation 引用。"""
+        nonlocal heartbeat_thread, heartbeat_stop_event
+        if current_thread is not None and current_thread.is_alive():
             return
-        try:
-            if ws_holder and ws_holder[0]:
-                payload = {"op": 1, "d": last_seq}
-                ws_holder[0].send(json.dumps(payload))
-                logger.debug(f"[QQ Gateway:{config_name}] Heartbeat sent, seq={last_seq}")
-        except Exception as err:
-            logger.debug(f"[QQ Gateway:{config_name}] Heartbeat error: {err}")
-        if heartbeat_interval_ms and not stop_event.is_set():
-            heartbeat_timer = threading.Timer(heartbeat_interval_ms / 1000.0, send_heartbeat)
-            heartbeat_timer.daemon = True
-            heartbeat_timer.start()
+        with heartbeat_state_lock:
+            if heartbeat_thread is current_thread:
+                heartbeat_thread = None
+                heartbeat_stop_event = None
 
-    def on_ws_message(_, message):
-        nonlocal last_seq, heartbeat_interval_ms, heartbeat_timer
+    def stop_heartbeat(*, wait: bool) -> None:
+        """停止当前心跳；等待阶段不持有状态锁，避免阻塞 close 回调。"""
+        if wait:
+            heartbeat_replace_lock.acquire()
+        try:
+            current_thread, _ = invalidate_heartbeat_generation()
+            if (
+                wait
+                and current_thread is not None
+                and current_thread.is_alive()
+                and current_thread is not threading.current_thread()
+            ):
+                current_thread.join()
+            clear_heartbeat_generation(current_thread)
+        finally:
+            if wait:
+                heartbeat_replace_lock.release()
+
+    def start_heartbeat(ws: websocket.WebSocketApp, interval_ms: int) -> None:
+        """停止旧 generation 后启动一个由 Gateway 聚合的新心跳 owner。"""
+        nonlocal heartbeat_thread, heartbeat_stop_event
+        with heartbeat_replace_lock:
+            current_thread, generation_epoch = invalidate_heartbeat_generation()
+            if (
+                current_thread is not None
+                and current_thread.is_alive()
+                and current_thread is not threading.current_thread()
+            ):
+                current_thread.join()
+            clear_heartbeat_generation(current_thread)
+            with heartbeat_state_lock:
+                if (
+                    generation_epoch != heartbeat_epoch
+                    or stop_event.is_set()
+                    or not ws_holder
+                    or ws_holder[0] is not ws
+                ):
+                    return
+                generation_stop_event = threading.Event()
+                # 心跳是 Gateway 的常驻子 owner，不占用进程共享 ThreadHelper；Gateway 退出前统一 join。
+                generation_thread = threading.Thread(
+                    target=heartbeat_loop,
+                    args=(
+                        generation_epoch,
+                        generation_stop_event,
+                        interval_ms / 1000.0,
+                        ws,
+                    ),
+                    daemon=True,
+                    name=f"qq-gateway-heartbeat-{config_name}",
+                )
+                heartbeat_stop_event = generation_stop_event
+                heartbeat_thread = generation_thread
+                generation_thread.start()
+
+    def on_ws_message(ws, message):
+        """处理当前 WebSocket 连接收到的 QQ Gateway 消息。"""
+        nonlocal last_seq
         try:
             payload = json.loads(message)
         except json.JSONDecodeError as err:
@@ -72,7 +161,8 @@ def run_gateway(
         t = payload.get("t")
 
         if s is not None:
-            last_seq = s
+            with heartbeat_state_lock:
+                last_seq = s
 
         logger.debug(f"[QQ Gateway:{config_name}] op={op} t={t}")
 
@@ -89,15 +179,10 @@ def run_gateway(
                     "shard": [0, 1],
                 },
             }
-            ws_holder[0].send(json.dumps(identify))
+            ws.send(json.dumps(identify))
             logger.info(f"[QQ Gateway:{config_name}] Identify sent")
 
-            # 启动心跳
-            if heartbeat_timer:
-                heartbeat_timer.cancel()
-            heartbeat_timer = threading.Timer(heartbeat_interval_ms / 1000.0, send_heartbeat)
-            heartbeat_timer.daemon = True
-            heartbeat_timer.start()
+            start_heartbeat(ws, heartbeat_interval_ms)
 
         elif op == 0:  # Dispatch
             if t == "READY":
@@ -147,59 +232,69 @@ def run_gateway(
 
         elif op == 9:  # Invalid Session
             logger.warning(f"[QQ Gateway:{config_name}] Invalid session")
-            if ws_holder and ws_holder[0]:
-                ws_holder[0].close()
+            ws.close()
 
     def on_ws_error(_, error):
+        """记录当前 WebSocket 连接上报的错误。"""
         logger.error(f"[QQ Gateway:{config_name}] WebSocket error: {error}")
 
-    def on_ws_close(_, close_status_code, close_msg):
+    def on_ws_close(ws, close_status_code, close_msg):
+        """失效当前连接及其心跳 generation，但不在回调线程等待。"""
         logger.info(f"[QQ Gateway:{config_name}] WebSocket closed: {close_status_code} {close_msg}")
-        if heartbeat_timer:
-            heartbeat_timer.cancel()
-        ws_holder.clear()
+        # close 回调可能由 stop() 调用线程触发，只发信号，统一由 Gateway 线程等待终态。
+        with heartbeat_state_lock:
+            is_current_connection = bool(ws_holder and ws_holder[0] is ws)
+            if is_current_connection:
+                ws_holder.clear()
+        if is_current_connection:
+            stop_heartbeat(wait=False)
 
     reconnect_delays = [1, 2, 5, 10, 30, 60]
     attempt = 0
 
-    while not stop_event.is_set():
-        try:
-            token = get_token_fn(app_id, app_secret)
-            gateway_url = get_gateway_url_fn(token)
-            logger.info(f"[QQ Gateway:{config_name}] Connecting to {gateway_url[:60]}...")
+    try:
+        while not stop_event.is_set():
+            try:
+                token = get_token_fn(app_id, app_secret)
+                gateway_url = get_gateway_url_fn(token)
+                logger.info(f"[QQ Gateway:{config_name}] Connecting to {gateway_url[:60]}...")
 
-            ws = websocket.WebSocketApp(
-                gateway_url,
-                on_message=on_ws_message,
-                on_error=on_ws_error,
-                on_close=on_ws_close,
-            )
-            ws_holder.clear()
-            ws_holder.append(ws)
+                ws = websocket.WebSocketApp(
+                    gateway_url,
+                    on_message=on_ws_message,
+                    on_error=on_ws_error,
+                    on_close=on_ws_close,
+                )
+                with heartbeat_state_lock:
+                    ws_holder.clear()
+                    ws_holder.append(ws)
 
-            # run_forever 会阻塞，需要传入 stop_event 的检查
-            # websocket-client 的 run_forever 支持 ping_interval, ping_timeout
-            # 我们使用自定义心跳，所以不设置 ping
-            ws.run_forever(
-                ping_interval=None,
-                ping_timeout=None,
-                skip_utf8_validation=True,
-            )
+                # websocket-client 的 run_forever 会阻塞；QQ 协议使用自定义心跳，不启用 ping。
+                try:
+                    ws.run_forever(
+                        ping_interval=None,
+                        ping_timeout=None,
+                        skip_utf8_validation=True,
+                    )
+                finally:
+                    # 旧连接的心跳必须先终止，下一轮 reconnect 才能取得 owner。
+                    stop_heartbeat(wait=True)
 
-        except Exception as e:
-            logger.error(f"[QQ Gateway:{config_name}] Connection error: {e}")
+            except Exception as err:
+                logger.error(f"[QQ Gateway:{config_name}] Connection error: {err}")
 
-        if stop_event.is_set():
-            break
-
-        delay = reconnect_delays[min(attempt, len(reconnect_delays) - 1)]
-        attempt += 1
-        logger.info(f"[QQ Gateway:{config_name}] Reconnecting in {delay}s (attempt {attempt})")
-        for _ in range(delay * 10):
             if stop_event.is_set():
                 break
-            time.sleep(0.1)
 
-    if heartbeat_timer:
-        heartbeat_timer.cancel()
-    logger.info(f"[QQ Gateway:{config_name}] Gateway thread stopped")
+            delay = reconnect_delays[min(attempt, len(reconnect_delays) - 1)]
+            attempt += 1
+            logger.info(
+                f"[QQ Gateway:{config_name}] Reconnecting in {delay}s (attempt {attempt})"
+            )
+            for _ in range(delay * 10):
+                if stop_event.is_set():
+                    break
+                time.sleep(0.1)
+    finally:
+        stop_heartbeat(wait=True)
+        logger.info(f"[QQ Gateway:{config_name}] Gateway thread stopped")

docs/refactor/backend-architecture-review.md

@@ -85,9 +85,9 @@
 |---|---|---|---|
 | 0 | 历史任务清账、现行架构图、外部契约核对和宿主基线对齐 | 已推送 | `d234c7132`；远端同 SHA；ahead/behind `0/0`；架构契约 `71 passed` |
 | 1 | Mypy fail-closed，并把 Ruff/Mypy 已下降债务固化为真实低水位 | 已推送 | `6062b0661`；远端同 SHA；ahead/behind `0/0`；Mypy 11994、Ruff 976 |
-| 2 | 用全量串行测试初始化非零 Coverage 低水位，并补齐 CI/文档防回退契约 | 已本地验证 | Ubuntu canonical：Application `9292/11949`（77.76%），Domain `3390/4278`（79.24%）；专项 `26 passed` |
-| 3 | 收口阶段 62 遗留的 QQ Gateway heartbeat Timer 所有权 | 待批次 2 | Timer 只 cancel 不 join，Gateway 主线程可能在 heartbeat 仍执行时报告停止成功 |
-| Final | 全仓回归、插件兼容复核、台账定稿和远端一致性验证 | 待前置批次 | 所有准入项已推送；全量测试和适用门禁通过；本地/远端 0/0 |
+| 2 | 用全量串行测试初始化非零 Coverage 低水位，并补齐 CI/文档防回退契约 | 已推送 | `265d3c6d1`；远端同 SHA；ahead/behind `0/0`；Application 77.76%，Domain 79.24% |
+| 3 | 收口阶段 62 遗留的 QQ Gateway heartbeat Timer 所有权 | 已全量验证 | generation owner 已实现；全量 `6391 passed, 6 skipped`；等待提交推送 |
+| Final | 全仓回归、插件兼容复核、台账定稿和远端一致性验证 | 进行中 | 本地门禁已通过；等待批次 3 推送及远端 0/0 复核 |
 
 ### 批次 0：审计与基线对齐
 
@@ -162,7 +162,8 @@
   最新专项覆盖，质量/CI 专项 `26 passed`，Ruff 通过，Pylint `10.00/10`，宿主、Schema、
   Coverage 和锁文件门禁均通过。
 
-待完成：提交推送并记录远端证据。
+交付证据：提交 `265d3c6d1` 已推送到 `origin/v3`；`git ls-remote` 返回同一 SHA，
+`HEAD...origin/v3` 为 `0/0`，提交严格包含批次 2 的 8 个主仓路径。
 
 ### 批次 3：QQ Gateway heartbeat owner
 
@@ -171,11 +172,29 @@
 WebSocket 和 Gateway 退出路径只调用 `cancel()`，而 `QQBot.stop()` 只等待 Gateway 主线程。
 当 heartbeat callback 已进入 `send()` 时，`cancel()` 不能终止它，外层因此可能误报收敛。
 
-停止条件：Gateway 最终退出路径在现有外层 20 秒硬预算内等待当前 Timer 真正终止；
-Timer 未终止时 Gateway 线程保持存活，`QQBot` 保留 owner 并允许再次停止；故障注入证明首次
+停止条件：Gateway 最终退出路径等待当前 heartbeat owner 真正终止，不为 heartbeat 叠加
+第二份调用方等待预算；heartbeat 未终止时 Gateway 线程保持存活，`QQBot` 在现有 20 秒
+Gateway join 预算后保留 owner 并允许再次停止；故障注入证明首次
 停止返回 `False`、释放 callback 后第二次返回 `True`；生命周期专项和 task ownership 门禁通过；
 批次独立提交推送。
 
+本地实现与专项验收：
+
+* 递归 Timer 已替换为一个由 Gateway 聚合、按 connection generation 捕获精确 WebSocket 的
+  常驻心跳线程；新 Hello 必须先失效并等待旧 generation，旧连接 close 不能终止新连接心跳；
+* 状态锁只保护 epoch、owner 和 sequence 快照，任何 `join()`、网络发送及 close 回调都不持锁；
+  Gateway 的每次连接和最终退出都等待子 owner 终止，所以未收敛时外层 Gateway 线程保持存活；
+* 回调中的 Identify、Invalid Session 均使用回调携带的精确 WebSocket，避免 `ws_holder`
+  被并发 close 清空或替换后误发到别的连接；公开 `QQBot.stop()` 合同和等待预算未改变；
+* 三条阻塞发送、重复 Hello 和双连接重连故障注入测试并行重复 4 轮均通过；完整生命周期专项
+  `96 passed`，Ruff 通过，Pylint `10.00/10`，task ownership、复杂度、async blocking、
+  宿主基线及 Ruff/Mypy ratchet 均通过；Mypy 低水位从 11994 净减至 11993。
+* 双连接用例直接证明重连前必须等待旧心跳终止，且迟到的旧连接 close 不会终止新 generation；
+  独立代码审查未发现生产线程生命周期、锁序、generation 或 reconnect 合同缺陷；
+* 4 分片全量回归 `6391 passed, 6 skipped`；锁文件、Schema、Coverage、service locator 和
+  官方插件 ABI 语义门禁均通过，架构/质量专项 `50 passed`。插件参考仓只有不参与语义判定的
+  schema/provenance 漂移，且本地分支落后其远端 2 个提交，本轮保持参考仓只读且不刷新指纹。
+
 ## 五、验证矩阵
 
 每个批次按改动范围选择下列命令，Final 全部执行：
@@ -217,6 +236,10 @@ git rev-list --left-right --count HEAD...origin/v3
 | 2026-08-26 | 批次 1 交付 | `6062b0661` 已推送；远端同 SHA；ahead/behind `0/0`；显式变更 10 个路径 |
 | 2026-08-26 | 批次 2 canonical Coverage | run `32977180133` 与本机计数一致；Application 77.76%，Domain 79.24%；零 fixture 已初始化 |
 | 2026-08-26 | 批次 2 本地验收 | 串行全量 `6379 passed, 6 skipped`；最新专项 `26 passed`；Coverage/宿主/Schema/锁文件门禁通过 |
+| 2026-08-26 | 批次 2 交付 | `265d3c6d1` 已推送；远端同 SHA；ahead/behind `0/0`；显式变更 8 个路径 |
+| 2026-08-26 | 批次 3 启动 | 仅收口阶段 62 已记录的 QQ heartbeat owner，不扩张公开 `QQBot.stop()` 合同 |
+| 2026-08-26 | 批次 3 专项验收 | 三条故障注入并行重复 4 轮稳定；生命周期 `96 passed`；静态、所有权和架构门禁通过 |
+| 2026-08-26 | 批次 3 全量验收 | 最终快照 4 分片 `6391 passed, 6 skipped`；全部适用主仓门禁通过；插件 ABI 语义无变化 |
 
 ## 七、本轮停止条件
 

docs/rules/06-code-styles.md

@@ -35,7 +35,10 @@
 - Prefer `async def` for I/O-bound operations (network requests, database queries, file operations).
 - Use `await` consistently; do not mix sync and async code paths in the same function without using `run_in_threadpool` from FastAPI or `asyncio.to_thread`.
 - For CPU-bound work that must not block the event loop, submit to `ThreadHelper` (see `app/runtime/thread.py`).
-- Do not use bare `threading.Thread` in new code; use `ThreadHelper.submit()`.
+- Use `ThreadHelper.submit()` for finite background work. A long-lived protocol loop may use a
+  dedicated `threading.Thread` only when its existing lifecycle owner signals and joins that child,
+  while the external caller bounded-waits and retains the parent owner on non-convergence; do not
+  create a dedicated thread pool for that exception.
 
 ---
 

tests/fixtures/architecture/mypy-baseline.json

@@ -2580,7 +2580,7 @@
     "type-arg": 13
   },
   "app/modules/qqbot/gateway.py": {
-    "no-untyped-def": 4,
+    "no-untyped-def": 3,
     "type-arg": 2
   },
   "app/modules/qqbot/module.py": {

tests/test_module_lifecycle.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import threading
 import time
 from unittest.mock import Mock, patch
@@ -13,6 +14,7 @@ from app.modules.feishu import FeishuModule
 from app.modules.feishu.feishu import Feishu
 from app.modules.filter import FilterModule
 from app.modules.plex import PlexModule
+from app.modules.qqbot import gateway as qq_gateway
 from app.modules.qqbot.module import QQBotModule
 from app.modules.qqbot.qqbot import QQBot
 from app.modules.slack import SlackModule
@@ -328,6 +330,335 @@ def test_qqbot_stop_retains_gateway_thread_until_retry() -> None:
     assert client.stop() is True
 
 
+def test_qqbot_stop_retains_gateway_until_inflight_heartbeat_finishes(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """心跳发送仍阻塞时 Gateway 必须保持存活，释放后才允许关闭成功。"""
+    heartbeat_started = threading.Event()
+    release_heartbeat = threading.Event()
+    connection_closed = threading.Event()
+    callbacks = {}
+    fake_ws = Mock()
+    hello_payload = json.dumps({"op": 10, "d": {"heartbeat_interval": 1}})
+
+    def build_websocket(_url, **kwargs):
+        """保存 Gateway 回调并返回隔离的 WebSocket 桩。"""
+        callbacks.update(kwargs)
+        return fake_ws
+
+    def send(payload: str) -> None:
+        """Identify 立即完成，首个心跳保持在发送中的故障状态。"""
+        if json.loads(payload).get("op") == 1:
+            heartbeat_started.set()
+            release_heartbeat.wait()
+
+    def close() -> None:
+        """同步触发 close 回调，复现 QQBot.stop() 的真实调用线程。"""
+        callbacks["on_close"](fake_ws, 1000, "test close")
+        connection_closed.set()
+
+    def run_forever(**_kwargs) -> None:
+        """发送 Hello 后保持连接，直到 stop() 主动关闭。"""
+        callbacks["on_message"](fake_ws, hello_payload)
+        connection_closed.wait()
+
+    fake_ws.send.side_effect = send
+    fake_ws.close.side_effect = close
+    fake_ws.run_forever.side_effect = run_forever
+    monkeypatch.setattr(qq_gateway.websocket, "WebSocketApp", build_websocket)
+
+    client = QQBot.__new__(QQBot)
+    client._gateway_stop = threading.Event()
+    client._gateway_ws_holder = []
+    client._gateway_join_timeout_seconds = 0.02
+    gateway_thread = threading.Thread(
+        target=qq_gateway.run_gateway,
+        kwargs={
+            "app_id": "app-id",
+            "app_secret": "secret",
+            "config_name": "test",
+            "get_token_fn": lambda _app_id, _secret: "token",
+            "get_gateway_url_fn": lambda _token: "wss://gateway.test",
+            "on_message_fn": lambda _payload: None,
+            "stop_event": client._gateway_stop,
+            "ws_holder": client._gateway_ws_holder,
+        },
+        daemon=True,
+    )
+    client._gateway_thread = gateway_thread
+    gateway_thread.start()
+
+    try:
+        assert heartbeat_started.wait(0.5)
+        started_at = time.monotonic()
+        assert client.stop() is False
+        assert time.monotonic() - started_at < 0.2
+        assert gateway_thread.is_alive()
+
+        release_heartbeat.set()
+        gateway_thread.join(timeout=1.0)
+        assert client.stop() is True
+        assert not gateway_thread.is_alive()
+    finally:
+        client._gateway_stop.set()
+        release_heartbeat.set()
+        connection_closed.set()
+        gateway_thread.join(timeout=1.0)
+
+
+def test_qq_gateway_replaces_heartbeat_generation_without_overlap(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """重复 Hello 必须等待旧心跳终止，任何时刻只能有一个发送 generation。"""
+    first_heartbeat_started = threading.Event()
+    release_first_heartbeat = threading.Event()
+    second_heartbeat_started = threading.Event()
+    release_second_heartbeat = threading.Event()
+    third_heartbeat_started = threading.Event()
+    second_hello_entered = threading.Event()
+    third_hello_entered = threading.Event()
+    connection_ready = threading.Event()
+    connection_closed = threading.Event()
+    counters_lock = threading.Lock()
+    callbacks = {}
+    fake_ws = Mock()
+    hello_payload = json.dumps({"op": 10, "d": {"heartbeat_interval": 1}})
+    heartbeat_count = 0
+    active_heartbeats = 0
+    max_active_heartbeats = 0
+
+    def build_websocket(_url, **kwargs):
+        """保存 Gateway 回调并返回隔离的 WebSocket 桩。"""
+        callbacks.update(kwargs)
+        return fake_ws
+
+    def send(payload: str) -> None:
+        """阻塞第一代心跳，并记录是否出现跨 generation 并发发送。"""
+        nonlocal heartbeat_count, active_heartbeats, max_active_heartbeats
+        if json.loads(payload).get("op") != 1:
+            return
+        with counters_lock:
+            heartbeat_count += 1
+            current_heartbeat = heartbeat_count
+            active_heartbeats += 1
+            max_active_heartbeats = max(
+                max_active_heartbeats,
+                active_heartbeats,
+            )
+        try:
+            if current_heartbeat == 1:
+                first_heartbeat_started.set()
+                release_first_heartbeat.wait()
+            elif current_heartbeat == 2:
+                second_heartbeat_started.set()
+                release_second_heartbeat.wait()
+            elif current_heartbeat == 3:
+                third_heartbeat_started.set()
+        finally:
+            with counters_lock:
+                active_heartbeats -= 1
+
+    def run_forever(**_kwargs) -> None:
+        """发送首个 Hello 后保持连接，第二个 Hello 由测试线程注入。"""
+        callbacks["on_message"](fake_ws, hello_payload)
+        connection_ready.set()
+        connection_closed.wait()
+
+    def send_second_hello() -> None:
+        """从独立调用线程注入重复 Hello，以观测 generation 屏障。"""
+        second_hello_entered.set()
+        callbacks["on_message"](fake_ws, hello_payload)
+
+    def send_third_hello() -> None:
+        """在第二代发送中再次注入 Hello，供 close 失效待发布 generation。"""
+        third_hello_entered.set()
+        callbacks["on_message"](fake_ws, hello_payload)
+
+    fake_ws.send.side_effect = send
+    fake_ws.run_forever.side_effect = run_forever
+    monkeypatch.setattr(qq_gateway.websocket, "WebSocketApp", build_websocket)
+
+    stop_event = threading.Event()
+    ws_holder = []
+    gateway_thread = threading.Thread(
+        target=qq_gateway.run_gateway,
+        kwargs={
+            "app_id": "app-id",
+            "app_secret": "secret",
+            "config_name": "generation-test",
+            "get_token_fn": lambda _app_id, _secret: "token",
+            "get_gateway_url_fn": lambda _token: "wss://gateway.test",
+            "on_message_fn": lambda _payload: None,
+            "stop_event": stop_event,
+            "ws_holder": ws_holder,
+        },
+        daemon=True,
+    )
+    gateway_thread.start()
+    hello_thread = threading.Thread(target=send_second_hello, daemon=True)
+    hello_thread_started = False
+    third_hello_thread = threading.Thread(target=send_third_hello, daemon=True)
+    third_hello_thread_started = False
+
+    try:
+        assert connection_ready.wait(0.5)
+        assert first_heartbeat_started.wait(0.5)
+        hello_thread.start()
+        hello_thread_started = True
+        assert second_hello_entered.wait(0.2)
+        assert not second_heartbeat_started.wait(0.03)
+        assert hello_thread.is_alive()
+        assert max_active_heartbeats == 1
+
+        release_first_heartbeat.set()
+        hello_thread.join(timeout=1.0)
+        assert not hello_thread.is_alive()
+        assert second_heartbeat_started.wait(0.5)
+        assert max_active_heartbeats == 1
+
+        third_hello_thread.start()
+        third_hello_thread_started = True
+        assert third_hello_entered.wait(0.2)
+        assert third_hello_thread.is_alive()
+        close_started_at = time.monotonic()
+        callbacks["on_close"](fake_ws, 1000, "close during replacement")
+        assert time.monotonic() - close_started_at < 0.2
+
+        release_second_heartbeat.set()
+        third_hello_thread.join(timeout=1.0)
+        assert not third_hello_thread.is_alive()
+        assert not third_heartbeat_started.wait(0.05)
+        assert max_active_heartbeats == 1
+
+        stop_event.set()
+        connection_closed.set()
+        gateway_thread.join(timeout=1.0)
+        assert not gateway_thread.is_alive()
+    finally:
+        stop_event.set()
+        release_first_heartbeat.set()
+        release_second_heartbeat.set()
+        connection_closed.set()
+        if hello_thread_started:
+            hello_thread.join(timeout=1.0)
+        if third_hello_thread_started:
+            third_hello_thread.join(timeout=1.0)
+        gateway_thread.join(timeout=1.0)
+
+
+def test_qq_gateway_reconnect_joins_old_heartbeat_and_ignores_stale_close(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """重连必须先回收旧心跳，迟到的旧连接 close 不得终止新 generation。"""
+    first_callbacks = {}
+    second_callbacks = {}
+    first_ws = Mock()
+    second_ws = Mock()
+    first_heartbeat_started = threading.Event()
+    release_first_heartbeat = threading.Event()
+    return_first_connection = threading.Event()
+    second_connection_created = threading.Event()
+    second_heartbeat_started = threading.Event()
+    release_second_heartbeat = threading.Event()
+    second_followup_started = threading.Event()
+    return_second_connection = threading.Event()
+    hello_payload = json.dumps({"op": 10, "d": {"heartbeat_interval": 1}})
+    websocket_count = 0
+    second_heartbeat_count = 0
+
+    def build_websocket(_url, **kwargs):
+        """依次创建两条连接，并保留各自回调以注入迟到 close。"""
+        nonlocal websocket_count
+        websocket_count += 1
+        if websocket_count == 1:
+            first_callbacks.update(kwargs)
+            return first_ws
+        if websocket_count == 2:
+            second_callbacks.update(kwargs)
+            second_connection_created.set()
+            return second_ws
+        raise AssertionError("Gateway 在停止后不应建立第三条测试连接")
+
+    def send_first(payload: str) -> None:
+        """阻塞第一条连接的心跳，供重连屏障检查。"""
+        if json.loads(payload).get("op") == 1:
+            first_heartbeat_started.set()
+            release_first_heartbeat.wait()
+
+    def send_second(payload: str) -> None:
+        """阻塞新连接首个心跳，并记录迟到 close 后是否继续工作。"""
+        nonlocal second_heartbeat_count
+        if json.loads(payload).get("op") != 1:
+            return
+        second_heartbeat_count += 1
+        if second_heartbeat_count == 1:
+            second_heartbeat_started.set()
+            release_second_heartbeat.wait()
+        else:
+            second_followup_started.set()
+
+    def run_first(**_kwargs) -> None:
+        """第一条连接握手后按测试信号返回，触发真实 reconnect 路径。"""
+        first_callbacks["on_message"](first_ws, hello_payload)
+        return_first_connection.wait()
+
+    def run_second(**_kwargs) -> None:
+        """第二条连接保持运行，直到测试完成迟到 close 校验。"""
+        second_callbacks["on_message"](second_ws, hello_payload)
+        return_second_connection.wait()
+
+    first_ws.send.side_effect = send_first
+    first_ws.run_forever.side_effect = run_first
+    second_ws.send.side_effect = send_second
+    second_ws.run_forever.side_effect = run_second
+    monkeypatch.setattr(qq_gateway.websocket, "WebSocketApp", build_websocket)
+    monkeypatch.setattr(qq_gateway.time, "sleep", lambda _seconds: None)
+
+    stop_event = threading.Event()
+    ws_holder = []
+    gateway_thread = threading.Thread(
+        target=qq_gateway.run_gateway,
+        kwargs={
+            "app_id": "app-id",
+            "app_secret": "secret",
+            "config_name": "reconnect-test",
+            "get_token_fn": lambda _app_id, _secret: "token",
+            "get_gateway_url_fn": lambda _token: "wss://gateway.test",
+            "on_message_fn": lambda _payload: None,
+            "stop_event": stop_event,
+            "ws_holder": ws_holder,
+        },
+        daemon=True,
+    )
+    gateway_thread.start()
+
+    try:
+        assert first_heartbeat_started.wait(0.5)
+        return_first_connection.set()
+        assert not second_connection_created.wait(0.05)
+
+        release_first_heartbeat.set()
+        assert second_connection_created.wait(0.5)
+        assert second_heartbeat_started.wait(0.5)
+
+        first_callbacks["on_close"](first_ws, 1000, "stale close")
+        release_second_heartbeat.set()
+        assert second_followup_started.wait(0.5)
+
+        stop_event.set()
+        second_callbacks["on_close"](second_ws, 1000, "current close")
+        return_second_connection.set()
+        gateway_thread.join(timeout=1.0)
+        assert not gateway_thread.is_alive()
+    finally:
+        stop_event.set()
+        release_first_heartbeat.set()
+        release_second_heartbeat.set()
+        return_first_connection.set()
+        return_second_connection.set()
+        gateway_thread.join(timeout=1.0)
+
+
 def test_wechat_bot_stop_reports_each_live_thread_until_retry() -> None:
     """企业微信网关或心跳任一存活时都必须返回未收敛。"""
     client = WeChatBot.__new__(WeChatBot)