refactor!: 协议帧与业务事件分流，各接入方式配专有协议处理器

BaseProtocol 从 emitter.py 移入新模块 protocol.py，接口由 register(emitter)
改为 async on_frame(payload)（返回需回送的响应体）；协议处理器不再注册进
emitter，而由 run_loop 注入对应的 connecter，由适配器逐帧调用。

破坏性变更：只有 op=0 的业务事件进事件队列交给 EventEmitter 分发，
HELLO/READY/INVALID_SESSION/op=13 等协议帧在连接器内就地处理——用户 handler
不再收到协议帧（此前 HELLO/INVALID_SESSION 等会路由到 handler）。
EventEmitter.register_protocol()/handle() 与 emitter 持有协议层的机制删除，
协议类不再用 register() 注册 handler。

附带修掉一处时序耦合：INVALID_SESSION 的会话重置此前依赖分发泵在重连前
消费掉该事件（慢 handler 会推迟它，导致带着已死 session_id 去 RESUME），
现在连接器在 break 重连前同步完成重置。

同时做的事件流优化：
- emit() 先取事件名、无 listener 时不构造 Event（未订阅事件省一次对象与拷贝）
- Event.data 改为惰性拷贝（只用 .typed/.raw 的 handler 不再触发 dict(d)）
- ws 重连收尾 await 被取消的 reply/heartbeat 任务，避免异常无人取用
- EventQueue.put_event/put_reply 去掉多余的 return await

测试 80 全绿（新增协议帧处理测试组、ws 集成断言协议帧不触达 handler），
ruff/pyright 干净。文档 ARCHITECTURE/API/GUIDE/README/CHANGELOG 同步。

CHANGELOG.md

@@ -26,15 +26,21 @@
 - 组件构造签名瘦身：`WebsocketProtocol(config, session, token)` 移除未使用的 `queue`；`WebsocketConnecter` 改收 `(config, http, token, session, queue)`；`WebhookConnecter` 改收 `(config, queue, handle_event)`，op=13 应答经注入的回调取得，不再依赖具体 `EventEmitter`；`Session` 合并 `sequence_id`/`seq` 双属性为 `seq`（心跳序列号无历史时发 0 而非 null）；移除服务端不会推送的 HEARTBEAT 死代码 handler；`ApiError`/`TokenError` 的错误响应解析合并为 `model.error_parts`
 - **intents.toml 改为只按大类订阅**：配置项从"`[分组]` 表内逐事件开关"简化为"`分组名 = true/false`"（如 `GROUP_AND_C2C_EVENT = true`）——网关只接受分组位掩码，组内事件本就无法单独订阅，此前的逐事件开关只是"组内任一为 true 即订阅整组"的装饰。破坏性变更：旧格式（表内逐事件开关）不再兼容，与拼错的分组名、非布尔值一样在启动时直接抛 `ValueError`，而非静默少订阅
 - 形参注入缓存化：handler 的签名与类型标注解析结果（此前每条事件都重新跑 `inspect.signature`/`get_type_hints`）合并缓存，`services` 命中改为每条事件实时判定——先注册 handler、后登记组件的既有装配顺序不受影响
+- **协议帧与业务事件分流，各接入方式配专有协议处理器**：`BaseProtocol` 从 `emitter.py` 移入新模块 `protocol.py`，接口由 `register(emitter)` 改为 `async on_frame(payload)`（返回需回送的响应体）；协议处理器不再注册进 emitter，而是由 `run_loop` 构造后**注入对应的 connecter**（`WebsocketConnecter(..., protocol)` / `WebhookConnecter(config, queue, protocol)`），由适配器逐帧调用。破坏性变更：只有 op=0 的业务事件进事件队列交给 `EventEmitter`，HELLO/READY/INVALID_SESSION/op=13 等协议帧在连接器内就地处理——用户 handler 不再收到协议帧；`EventEmitter.register_protocol()`、`EventEmitter.handle()`、`emitter._protocol`/`on_payload` 机制随之删除，协议类不再需要 `register()` 注册 handler。附带修掉一处时序耦合：`INVALID_SESSION` 的会话重置此前依赖分发泵在重连前消费掉该事件，现在在 connecter `break` 重连前同步完成，不会误 RESUME 已死会话
+- 事件分发热路径减负：`emit()` 先取事件名、无 listener 时不再构造 `Event`（每条未订阅事件/协议帧省一次对象与 dict 拷贝）；`Event.data` 改为惰性拷贝（只用 `.typed`/`.raw` 的 handler 不再触发 `dict(d)`）
+- ws 重连收尾补 `await`：取消 `reply`/`heartbeat` 任务后 `gather(..., return_exceptions=True)` 取回结果，避免 "Task exception was never retrieved" 并确保它们在下一轮重连前真正停下
+- `EventQueue.put_event`/`put_reply` 去掉多余的 `return await`
 
 ### 移除
 
+- `EventEmitter.register_protocol()`、`EventEmitter.handle()` 与 `emitter` 持有协议层的机制（协议处理器改为由 connecter 持有并逐帧调用，见上文"变更"）
 - apscheduler 依赖（心跳迁入 connecter 后不再需要调度器）
 - 音乐插件示例（`plugins/`、根目录 `main.py`）已拆分为独立的点歌 bot 项目（含 Markdown 卡片、按钮回调与内置短网址多线程下载代理），不再随 SDK 仓库分发
 
 ### 修复
 
-- **INVALID_SESSION 可续传标记失效**：op=9 的 `d` 是布尔值（true=可 RESUME），此前参数注入回退把 d 字典化成 `{}`（恒 falsy），任何 INVALID_SESSION 都会清空会话——RESUME 续传永远走不到，断线期间本可回放的消息丢失；注入回退改为直传原始 d，`async def on_invalid(resumable: bool)` 现在能拿到真实布尔
+- **INVALID_SESSION 可续传标记失效**：op=9 的 `d` 是布尔值（true=可 RESUME），此前参数注入回退把 d 字典化成 `{}`（恒 falsy），任何 INVALID_SESSION 都会清空会话——RESUME 续传永远走不到，断线期间本可回放的消息丢失。现协议帧不再经参数注入，`WebsocketProtocol` 直接读取 `d` 的真值，`resumable=true` 正确保留会话（协议帧处理测试覆盖）
+- **会话重置时序**：`INVALID_SESSION` 的会话清空此前依赖分发泵在重连前消费掉该事件（慢 handler 会推迟它，导致带着已死 session_id 去 RESUME）；现在连接器在 `break` 重连前同步调用协议处理器完成重置
 - token 请求失败时抛出 `TokenError`（携带 QQ 返回的 `code`/`message`），不再抛出掩盖真实原因的 `KeyError: 'access_token'`
 - 修正被动回复 `event_id` 的指引：官方要求取网关帧最外层的 id（`Event.event_id`），事件体里的裸 UUID（`Interaction.id` 等）不能作 `event_id`，此前文档与 docstring 的说法有误
 - 响应 `Content-Type` 非 JSON 时不再抛 `ContentTypeError`（个别接口如回应互动返回 200 text/plain + `{}` body）；JSON 解析统一切换 ujson

README.md

@@ -100,8 +100,9 @@ async def handle(event: Event, api: BotApi):
         msg_id=event.message_id,
     )   # 失败抛 ApiError；富媒体先 upload_group_file 拿 file_info
 
-# 业务事件（op=0）处理器的返回值不会回流；
-# 协议层（op≠0）由 ws_protocol / webhook_protocol 的类处理，经 reply 队列发回。
+# 业务事件（op=0）才进分发，返回值不回流，回复请显式调 BotApi；
+# 协议帧（HELLO/READY/INVALID_SESSION/op=13）由各接入方式的协议处理器
+# 在连接器内就地处理，不进队列、也不会触达 handler。
 # 单个 handler 抛异常不会影响其他 handler 和主循环，异常会被记录。
 
 if __name__ == "__main__":

docs/API.md

@@ -7,7 +7,8 @@
 ```python
 from qqbotsdk import (
     main, run_loop,          # 入口（组件在 run_loop 显式装配）
-    EventEmitter, BaseProtocol, EventQueue,   # 分发层
+    EventEmitter, EventQueue,   # 分发层
+    BaseProtocol,            # 协议处理器接口（自定义接入方式时继承）
     Config, Connecter,       # 配置与连接器接口
 )
 ```
@@ -21,7 +22,7 @@ from qqbotsdk import (
 | `EventQueue()` | 事件/应答队列；`EventEmitter` 默认自建，需多处共享时手动传入（`ee.queue` 取实例） |
 | `Config` | frozen dataclass，环境变量一次性读齐（见下文） |
 | `Connecter` | 连接适配器 Protocol（接口），按 `CONNECTER` 自动选择实现 |
-| `BaseProtocol` | 协议层抽象基类，自定义协议处理器时继承并实现 `register(emitter)` |
+| `BaseProtocol` | 协议处理器抽象基类（`protocol.py`），自定义接入方式时继承并实现 `on_frame(payload)` |
 
 子包按需导入：
 
@@ -36,18 +37,17 @@ from qqbotsdk.events import Event
 ```python
 ee = EventEmitter()
 
-@ee.on("GROUP_AT_MESSAGE_CREATE")     # DISPATCH 事件用 t 名
-@ee.on("INVALID_SESSION")             # 协议事件用 Opcode 名（IDENTIFY/RESUME/HEARTBEAT/...）
+@ee.on("GROUP_AT_MESSAGE_CREATE")     # 业务事件用 t 名（仅 op=0 会分发到这里）
 async def handler(...): ...
 
 @ee.on("GROUP_AT_MESSAGE_CREATE", background=True)   # 后台并发执行，不阻塞事件流
 async def slow(...): ...
 ```
 
-- **事件命名**：op=0（DISPATCH）按 payload 的 `t` 名路由；其余按 `Opcode` 名路由。协议事件（op≠0）已由内置协议层处理，一般只需监听业务事件；handler 返回 dict 时协议事件的返回值会回流给 ws 发送（业务事件不会）。
+- **只分发业务事件**：`EventEmitter` 只处理 op=0（DISPATCH）事件，按 payload 的 `t` 名路由。协议帧（HELLO/READY/INVALID_SESSION/op=13 等）由各接入方式的协议处理器（`WebsocketProtocol`/`WebhookProtocol`）在连接器内就地处理，不进事件队列、也不会触达你的 handler。
 - **异常隔离**：handler 异常就地记录（loguru），不影响同事件其他 handler 与主循环。
-- **后台并发**：`background=True` 的 handler 进后台任务，emit 不等它；同一事件内失去先后保证，返回值不回流，仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务（`ee.close()`）。
-- **`ee.handle(payload)`**：同步分发并返回首个 handler 的非空返回值，供协议层应答 op=13 等场景，业务代码不用。
+- **后台并发**：`background=True` 的 handler 进后台任务，emit 不等它；同一事件内失去先后保证，仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务（`ee.close()`）。
+- 业务 handler 的返回值不回流；回复消息请在 handler 里显式调 `BotApi`。
 
 ### handler 参数注入约定
 

docs/ARCHITECTURE.md

@@ -11,14 +11,14 @@ flowchart TB
         WH["WebhookConnecter\naiohttp server\n验签 + TTL去重"]
     end
 
-    subgraph 分发层["分发层 emitter.py"]
-        Q[("EventQueue\nmsg_queue / reply_queue")]
-        EE["EventEmitter\nop/t 路由 + 参数注入\nhandler 异常隔离\nbackground 后台并发"]
+    subgraph 协议层["协议层 protocol.py"]
+        WSP["WebsocketProtocol\nHELLO→IDENTIFY/RESUME\nREADY/INVALID_SESSION + s 序列号"]
+        WHP["WebhookProtocol\nop=13 验证应答"]
     end
 
-    subgraph 协议层["协议层 BaseProtocol"]
-        WSP["WebsocketProtocol\nIDENTIFY/RESUME\ns 序列号跟踪"]
-        WHP["WebhookProtocol\nop=13 验证应答"]
+    subgraph 分发层["分发层 emitter.py"]
+        Q[("EventQueue\nmsg_queue / reply_queue")]
+        EE["EventEmitter\nop=0/t 路由 + 参数注入\nhandler 异常隔离\nbackground 后台并发"]
     end
 
     subgraph 业务层["业务层"]
@@ -35,12 +35,9 @@ flowchart TB
 
     QQ["QQ 开放平台"]
 
-    WS -- "事件 payload" --> Q
-    WH -- "事件 payload" --> Q
+    WS -- "业务事件 op=0" --> Q
+    WH -- "业务事件 op=0" --> Q
     Q -- "dispatch 泵" --> EE
-    EE -- "on_payload" --> WSP
-    WSP -- "IDENTIFY/RESUME 等\npayload 返回值" --> Q
-    WS -- "HEARTBEAT（心跳任务）" --> Q
     EE -- "Event 注入" --> USER
     USER -- "api: BotApi" --> API
     API --> QQ
@@ -48,6 +45,9 @@ flowchart TB
     TK --> HTTP
     API --> TK
     WSP --> TK
+    WS -. "逐帧 on_frame\n（协议帧就地处理，不入队）" .-> WSP
+    WH -. "op=13 逐帧 on_frame" .-> WHP
+    WSP -. "IDENTIFY/RESUME 应答" .-> WS
     WHP --> CRY
     WH --> CRY
     CFG -. "构造传参" .-> WS & WH & API & TK & WSP & WHP
@@ -59,30 +59,28 @@ flowchart TB
 sequenceDiagram
     participant QQ as QQ 开放平台
     participant WS as WebsocketConnecter
+    participant P as WebsocketProtocol
     participant Q as EventQueue
     participant EE as EventEmitter
-    participant P as WebsocketProtocol
     participant H as 用户 handler
 
     QQ->>WS: HELLO (op=10)
-    WS->>Q: put_event
-    Q->>EE: dispatch 泵
-    EE->>P: on_payload（记 s 序列号）
-    EE->>P: 路由到 HELLO handler
-    P-->>Q: IDENTIFY/RESUME payload → reply_queue
+    WS->>P: on_frame（握手判定）
+    P-->>WS: IDENTIFY / RESUME payload
     WS->>QQ: send（过滤 NOT_SET 字段）
 
     loop 每次心跳间隔（心跳任务随连接生灭）
         WS->>Q: HEARTBEAT payload
         WS->>QQ: send
-        QQ->>WS: HEARTBEAT_ACK（刷新看门狗时钟）
+        QQ->>WS: HEARTBEAT_ACK（刷新看门狗时钟，不进队列）
     end
 
     QQ->>WS: DISPATCH 事件 (op=0, t=GROUP_AT_MESSAGE_CREATE)
-    WS->>Q: put_event
+    WS->>P: on_frame（记 s 序列号）
+    WS->>Q: put_event（仅 op=0）
     Q->>EE: dispatch 泵
     EE->>H: 按形参标注注入 Event/BotApi
-    H->>EE: 返回值不回流（op=0）
+    H->>EE: 返回值不回流
 ```
 
 ## 分层规则
@@ -91,33 +89,33 @@ sequenceDiagram
 
 | 层 | 模块 | 职责 | 不允许出现 |
 |---|---|---|---|
-| 接入层 | `connecter.py` | 传输细节：握手、心跳、重连、验签、去重 | 事件业务语义、协议状态机 |
-| 分发层 | `emitter.py` | 路由、参数注入、异常隔离、后台并发、应答回流 | 传输细节、具体协议状态机 |
-| 协议层 | `ws_protocol.py` `webhook_protocol.py` | 鉴权/验证等协议状态机（会话语义） | 传输细节（只经队列收发） |
+| 接入层 | `connecter.py` | 传输细节：握手、心跳、重连、验签、去重；逐帧驱动协议处理器 | 事件业务语义、协议状态机 |
+| 协议层 | `protocol.py` `ws_protocol.py` `webhook_protocol.py` | 适配器专有的协议处理（`BaseProtocol.on_frame`）：鉴权/验证/会话语义 | 传输细节、业务语义 |
+| 分发层 | `emitter.py` | 只分发业务事件（op=0）：路由、参数注入、异常隔离、后台并发 | 传输细节、协议帧、协议状态机 |
 | 业务层 | `api/` `payloads.py` + 用户代码 | OpenAPI 调用、payload 解析与机器人逻辑 | 协议细节 |
 
-跨层共享的最小公共物：`model.py`（Opcode/Payload/Intent）、`queue.py`（事件通道）、`config.py`、`session.py`（ws 会话状态）、`payloads.py`（事件 dataclass 与解析，被 `events.py`/`emitter.py` 使用）。
+跨层共享的最小公共物：`model.py`（Opcode/Payload/Intent）、`protocol.py`（协议处理器接口，被接入层与协议层共同依赖）、`queue.py`（事件通道）、`config.py`、`session.py`（ws 会话状态）、`payloads.py`（事件 dataclass 与解析，被 `events.py`/`emitter.py` 使用）。
 
 ## 生命周期与依赖装配
 
 - 全部组件在 `run_loop`（`__init__.py`）显式构造装配，无 DI 框架，也无任何全局单例。
 - 组件按类型登记进 `emitter.services`，handler 形参标注类型即得实例；用户自定义组件同样登记即可注入。
 - 唯一的 `ClientSession` 在 `run_loop` 构造，`finally` 中先 `emitter.close()` 收尾后台 handler、再统一释放连接池（全 SDK 共享一个连接池）。
-- 接入方式的选择收敛在 `run_loop` 的 `match config.connecter` 装配分支，主体代码只依赖 `BaseProtocol`/`Connecter` 抽象；两个 connecter 的依赖面一致（config + queue + 各自所需的窄依赖），`WebhookConnecter` 经 `handle_event` 回调取得 op=13 应答，不感知 `EventEmitter`。
+- 接入方式的选择收敛在 `run_loop` 的 `match config.connecter` 装配分支：每种接入方式构造**自己的协议处理器**并注入对应的 connecter（主体不再持有协议层）；`WebhookConnecter` 经协议处理器的 `on_frame` 就地取得 op=13 应答，不感知 `EventEmitter`。两个 connecter 的依赖面一致（config + queue + 各自所需的窄依赖 + `BaseProtocol`）。
 - 用户在 `main(ee)` 之前构造的 `EventEmitter` 会被复用（连同它的队列），保证 handler 注册在主体实际使用的实例上。
 
 ## 关键设计决策
 
-1. **管线而非回调直调**：事件一律经 `EventQueue` 异步流转，接入与分发两侧解耦——Webhook handler 里不需要知道 ws 的存在，反之亦然。代价是多一跳队列延迟（可忽略）。
-2. **协议层只管会话语义，心跳归传输层**：`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号（说什么），`WebsocketConnecter` 负责连接与心跳（怎么保活）——心跳任务随连接生灭（HELLO 给出间隔后启动，断线即取消），断线期间不在 reply 队列堆积过期心跳，也免去了调度器的生命周期管理。协议类通过返回 payload dict 经 reply 队列外发。
+1. **业务事件走管线，协议帧走专有处理器**：只有 op=0 的业务事件经 `EventQueue` 异步流转给 `EventEmitter` 分发；HELLO/READY/INVALID_SESSION/op=13 等协议帧由 connecter 逐帧调用其专有协议处理器（`BaseProtocol.on_frame`）就地处理。这样用户 handler 不会看到协议帧，且"断线前重置会话"这类时序不再取决于分发泵的进度——`INVALID_SESSION` 的会话清空在 connecter `break` 重连前已同步完成，不会误 RESUME 一个已死会话。代价是多一跳队列延迟（可忽略）。
+2. **协议层只管会话语义，心跳归传输层**：`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号（说什么），`WebsocketConnecter` 负责连接与心跳（怎么保活）——心跳任务随连接生灭（HELLO 给出间隔后启动，断线即取消），断线期间不在 reply 队列堆积过期心跳，也免去了调度器的生命周期管理。协议处理器的应答经 `on_frame` 返回值由 connecter 写回（ws 走 reply 队列，webhook 走 HTTP 响应体）。
 3. **看门狗两层互补**（见 `connecter.py` 模块注释）：静默超时抓连接彻底失联，ACK 期限检查抓"有事件流但心跳已死"。阈值 `_SILENCE_FACTOR=1.1`、`_ACK_FACTOR=1.2` 是按 ACK 自然节奏校准的，**不能压到 1× 整**，否则正常抖动会周期性误杀（误杀由 RESUME 兜底，不丢消息）。重连退避只在连接稳定存活（`_STABLE_SECONDS`）后才归一，抖动的服务端不会被 1s 间隔反复探测。
 4. **handler 异常就地隔离**：`_invoke` 捕获记日志，`dispatch` 主循环另有兜底；单条事件的单个 handler 失败不影响进程。`background=True` 的 handler 进后台任务（emit 不等它），任务由 emitter 持强引用、`close()` 统一取消。
-5. **注入约定**：handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象；`Event` → 事件封装；登记进 `emitter.services` 的组件类型（如 `BotApi`）→ 实例；未标注或未登记 → 回退传**原始 d**（协议事件的 d 可能是标量，如 INVALID_SESSION 的 resumable 布尔；dict 化会丢真值——这里曾出过 bug）。解析是纯同步查表（`get_type_hints` + dict），无框架、无运行时魔法。
+5. **注入约定**：handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象；`Event` → 事件封装；登记进 `emitter.services` 的组件类型（如 `BotApi`）→ 实例；未标注或未登记 → 回退传**原始 d**。解析是纯同步查表（`get_type_hints` + 签名缓存），无框架、无运行时魔法。签名与类型标注每个 handler 只解析一次并缓存，但 `services` 命中在每条事件时实时判定，保证"先注册 handler、后登记组件"的装配顺序可用。
 6. **OpenAPI 请求内核自治**：`api/core.py` 统一处理鉴权头、429 限流重试（优先 `Retry-After`，脏值容错，退避封顶）与 401 刷新 token 重试一次；v2/channel/guild 三个域方法集继承内核组合成 `BotApi`，新增域加模块即可，方法对用户始终平铺在 `BotApi` 上。
 
 ## 扩展点
 
-- **新增接入方式**（如轮询/其他平台）：实现 `Connecter.run()` 与 `BaseProtocol.register()`，在 `run_loop` 的 `match` 装配分支加一个 `case`，`Config.load` 的白名单加一个名字。
+- **新增接入方式**（如轮询/其他平台）：实现 `Connecter.run()` 与 `BaseProtocol.on_frame()`，在 `run_loop` 的 `match` 装配分支加一个 `case`（构造该方式的协议处理器并注入 connecter），`Config.load` 的白名单加一个名字。
 - **新增业务 API**：在 `api/` 对应域模块（`v2`/`channel`/`guild`）加方法即自动平铺到 `BotApi`，全部获得鉴权、限流重试与错误处理；新增域则新建模块并组合进 `api/__init__.py` 的 `BotApi`。群/单聊消息经 `_post_message` 统一处理 msg_id/msg_seq/media 语义；已有方法覆盖：单聊/群聊消息与富媒体上传、频道消息/撤回/表态、公告、精华消息、日程、禁言、频道信息查询（guild/channels/roles/member）。
 - **新增事件类型 dataclass**：在 `payloads.py` 定义 dataclass 并登记 `EVENT_TYPES`，handler 即可按类型标注拿解析结果；未收录事件回退传 dict。当前已覆盖 intents.toml 的全部事件组（消息、成员变动、审核、表态、互动、论坛、音频）。
 
@@ -125,9 +123,9 @@ sequenceDiagram
 
 `tests/` 覆盖：
 
-- 纯逻辑：crypto 往返、intents 解析、Config 校验、`_route` 路由边界、webhook 去重、序列号单调性、handler 异常隔离与 reply 回流、payload dataclass 解析、BotApi 各业务方法的拼参（Recording 子类截获 request）。
+- 纯逻辑：crypto 往返、intents 解析、Config 校验、协议帧处理（`test_route.py`：IDENTIFY/RESUME 选路、READY 捕获、s 序列号、INVALID_SESSION 重置、业务事件不产生应答）、webhook 去重、序列号单调性、handler 异常隔离、payload dataclass 解析、BotApi 各业务方法的拼参（Recording 子类截获 request）。
 - **网络集成**（`test_ws_integration.py` / `test_webhook_integration.py`）：
-  - WebSocket：aiohttp 假网关（token 接口 + gateway + ws 协议行为）端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK（心跳任务随连接生灭）→ 业务事件 dataclass 分发；
+  - WebSocket：aiohttp 假网关（token 接口 + gateway + ws 协议行为）端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK（心跳任务随连接生灭）→ 业务事件 dataclass 分发，并断言协议帧（HELLO/HEARTBEAT_ACK）不会触达 handler；
   - Webhook：真实起 `WebhookConnecter` 的 http server（TestServer），覆盖 op=13 验证应答（签名内容正确性）、合法签名放行分发、坏签名 401、同 id 去重。
 
-另有 429/401 重试、token 并发锁、op=9 resumable 布尔注入、background 后台并发与 `close()` 收尾的专项单测。两套集成测试验证的都是真实连接路径，改接入层/协议层后跑 `uv run pytest` 即可回归。
+另有 429/401 重试、token 并发锁、background 后台并发与 `close()` 收尾的专项单测。两套集成测试验证的都是真实连接路径，改接入层/协议层后跑 `uv run pytest` 即可回归。

docs/GUIDE.md

@@ -116,11 +116,12 @@ handler 参数按标注解析的完整规则：
 | payload dataclass（如 `GroupAtMessage`） | 解析后的 dataclass 对象；未收录事件回退原始 dict |
 | `Event` | 事件封装对象（`.raw`/`.data`/`.typed`/`.user_id`/`.group_id`/`.content`/`.message_id`/`.event_id`） |
 | 组件类型（`BotApi`/`Session`/`Config` 等） | `emitter.services` 中按类型登记的实例；未登记回退传原始 d |
-| 无标注 | 原始 d（业务事件即事件 dict；协议事件可能是标量，如 `INVALID_SESSION` 的 `resumable: bool`） |
+| 无标注 | 原始 d（即事件 dict） |
 
 注意：
 
-- 业务事件（op=0）处理器的**返回值不会回流**，回复消息请在 handler 里显式调 API。
+- 只有**业务事件（op=0）**会分发给你的 handler；协议帧（HELLO/READY/INVALID_SESSION/op=13）由 SDK 的协议处理器就地处理，不触达 handler（`Ready` 是例外，它既是握手结果也是可监听的业务事件）。
+- 业务事件处理器的**返回值不会回流**，回复消息请在 handler 里显式调 API。
 - 单个 handler 抛异常只记录日志，不影响同事件其他 handler 和主循环。
 - 所在分组未订阅（intents.toml 中为 `false`）的事件不会到达 handler。
 - 自定义组件也能注入：`ee.services[MyService] = MyService(...)` 登记后，handler 形参标注 `MyService` 即可拿到实例。

src/qqbotsdk/__init__.py

@@ -15,8 +15,8 @@ from .api import BotApi
 from .config import Config
 from .connecter import Connecter as Connecter
 from .connecter import WebhookConnecter, WebsocketConnecter
-from .emitter import BaseProtocol as BaseProtocol
 from .emitter import EventEmitter as EventEmitter
+from .protocol import BaseProtocol as BaseProtocol
 from .queue import EventQueue as EventQueue
 from .session import Session
 from .token import AccessToken
@@ -52,17 +52,20 @@ async def run_loop(emitter: EventEmitter | None = None) -> None:
         }
     )
 
+    # 各接入方式配自己的协议处理器：协议帧由适配器就近处理，只有业务事件
+    # （op=0）入队交给 emitter 分发（见 protocol.py / connecter.py）
     match config.connecter:
         case "websocket":
             protocol = WebsocketProtocol(config, session, token)
-            connecter = WebsocketConnecter(config, http, token, session, emitter.queue)
+            connecter = WebsocketConnecter(
+                config, http, token, session, emitter.queue, protocol
+            )
         case "webhook":
             protocol = WebhookProtocol(config)
-            connecter = WebhookConnecter(config, emitter.queue, emitter.handle)
+            connecter = WebhookConnecter(config, emitter.queue, protocol)
         case unknown:
             raise ValueError(f"未知的 CONNECTER: {unknown}")
 
-    emitter.register_protocol(protocol)
     logger.info(f"qqbotsdk 启动，连接方式: {config.connecter}")
     try:
         await asyncio.gather(emitter.dispatch(), connecter.run())

src/qqbotsdk/connecter.py

@@ -1,5 +1,9 @@
 """连接适配器：把两种接入方式（WebSocket / Webhook）归一为事件流入、reply 队列流出。
 
+每条帧先交给本适配器注入的协议处理器（`BaseProtocol.on_frame`）处理协议语义，
+只有业务事件（op=0）才入队交给 emitter 分发给用户 handler——协议帧与业务事件
+因此不共享分发路径。
+
 - WebsocketConnecter：gateway 握手 + ws 长连接，指数退避重连（连接稳定
   存活才把退避归一，抖动的服务端不会被 1s 间隔反复探测）；心跳任务随
   连接生灭——HELLO 给出间隔后周期投递，断线即取消，不在 reply 队列堆积
@@ -7,13 +11,12 @@
   收到消息时的 ACK 期限检查抓"有事件流但心跳已死"；
 - WebhookConnecter：aiohttp server 接收回调，验签失败 401，
   按 payload id 做 60s TTL 去重，正常推送进事件队列；op=13 验证请求
-  经注入的 handle_event 同步取得应答（HTTP 响应不走 reply 队列）。
+  由协议处理器就地应答（HTTP 响应不走 reply 队列）。
 """
 
 import asyncio
-from collections.abc import Awaitable, Callable
 from time import monotonic
-from typing import Any, Protocol
+from typing import Protocol, cast
 
 import aiohttp
 import ujson
@@ -23,6 +26,7 @@ from loguru import logger
 from .config import Config
 from .crypto import verify_sig
 from .model import NOT_SET, Opcode, Payload, payload_of
+from .protocol import BaseProtocol
 from .queue import EventQueue
 from .session import Session
 from .token import AccessToken
@@ -51,12 +55,14 @@ class WebsocketConnecter:
         token: AccessToken,
         session: Session,
         queue: EventQueue,
+        protocol: BaseProtocol,
     ) -> None:
         self._config = config
         self._http = http
         self._token = token
         self._session = session
         self._queue = queue
+        self._protocol = protocol
 
     async def run(self) -> None:
         backoff: float = 1
@@ -105,6 +111,9 @@ class WebsocketConnecter:
             finally:
                 reply.cancel()
                 heartbeat.cancel()
+                # 取回被取消任务的结果，避免 "Task exception was never retrieved"，
+                # 也确保它们在下一轮重连前真正停下
+                await asyncio.gather(reply, heartbeat, return_exceptions=True)
             logger.info("WebSocket 已断开")
         return monotonic() - started
 
@@ -129,7 +138,17 @@ class WebsocketConnecter:
                         interval_fut.set_result(interval)
             elif op == Opcode.HEARTBEAT_ACK:
                 last_ack = monotonic()
-            await self._queue.put_event(data)
+
+            # 每帧都交给本适配器的协议处理器（HELLO→IDENTIFY/RESUME、
+            # READY→捕获会话、INVALID_SESSION→清会话、s→序列号）；只有业务
+            # 事件（op=0）入队分发。协议帧不入队，所以 INVALID_SESSION 的
+            # 会话重置在下面 break 之前已同步完成，重连不会误 RESUME 死会话。
+            response = await self._protocol.on_frame(data)
+            if response is not None:
+                await self._queue.put_reply(cast(Payload, response))
+            if op == Opcode.DISPATCH:
+                await self._queue.put_event(data)
+
             if op in (Opcode.RECONNECT, Opcode.INVALID_SESSION):
                 break
             if monotonic() - last_ack > interval * self._ACK_FACTOR:
@@ -160,13 +179,13 @@ class WebhookConnecter:
         self,
         config: Config,
         queue: EventQueue,
-        handle_event: Callable[[Payload], Awaitable[Any]],
+        protocol: BaseProtocol,
     ) -> None:
-        """handle_event 供 op=13 验证请求同步取应答（通常传 emitter.handle）；
-        与 ws 侧对齐，本类只依赖队列与该回调，不感知 EventEmitter。"""
+        """protocol 为本接入方式的协议处理器（WebhookProtocol），op=13 验证
+        请求由它就地应答；本类只依赖队列与该处理器，不感知 EventEmitter。"""
         self._config = config
         self._queue = queue
-        self._handle_event = handle_event
+        self._protocol = protocol
         self._seen: dict[str, float] = {}
 
     def _make_app(self) -> web.Application:
@@ -206,7 +225,7 @@ class WebhookConnecter:
 
         if payload.get("op") == Opcode.VALIDATION:
             logger.info("收到 Webhook 验证请求")
-            ret = await self._handle_event(payload)
+            ret = await self._protocol.on_frame(payload)
             if ret is None:
                 return web.json_response(
                     {"error": "missing validation handler"}, status=500
@@ -224,5 +243,8 @@ class WebhookConnecter:
             logger.info(f"Webhook 重复推送已跳过: {event_id}")
             return web.json_response({"opcode": 12})
 
-        await self._queue.put_event(payload)
+        if payload.get("op") == Opcode.DISPATCH:
+            await self._queue.put_event(payload)
+        else:
+            logger.warning(f"Webhook 收到非业务帧，已忽略: op={payload.get('op')}")
         return web.json_response({"opcode": 12})

src/qqbotsdk/emitter.py

@@ -1,22 +1,21 @@
-"""事件分发主体：路由 payload → 触发 handler → 协议层应答回流。
+"""事件分发主体：路由业务事件 → 触发 handler → 异常隔离与后台并发。
 
-- `dispatch()` 泵循环消费 EventQueue.msg_queue，`emit()` 按
-  op（DISPATCH 用 t 名、其余用 Opcode 名）路由到 `@on` 注册的 handler；
+- `dispatch()` 泵循环消费 EventQueue.msg_queue，`emit()` 把 op=0 的业务事件
+  按 t 名路由到 `@on` 注册的 handler；协议帧（HELLO/INVALID_SESSION/op=13 等）
+  不进队列，由连接适配器的协议处理器就地处理（见 protocol.py）；
 - handler 参数按形参标注注入：payload dataclass → 解析对象，`Event` →
   事件封装，`services` 里登记的组件类型 → 实例，其余直传原始 d
-  （可能是 dict、标量甚至 None，协议事件的 d 不保证是 dict）；
-- 业务 handler（op=0）返回值不回流；协议层（op≠0）返回的 payload
-  进 reply 队列由 ws 发出。handler 异常就地隔离，不影响其他 handler。
+  （可能是 dict、标量甚至 None）；
+- 业务 handler 返回值不回流；handler 异常就地隔离，不影响其他 handler。
 - `on(..., background=True)` 把 handler 放进后台任务并发执行：
-  不阻塞事件流，但同事件 handler 间失去先后保证，返回值不回流；
+  不阻塞事件流，但同事件 handler 间失去先后保证；
   进程退出经 `close()` 统一取消在飞任务。
 """
 
 import asyncio
 import inspect
-from abc import ABC, abstractmethod
 from collections.abc import Awaitable, Callable
-from typing import Any, TypeVar, cast, get_type_hints, overload
+from typing import Any, TypeVar, get_type_hints, overload
 
 from loguru import logger
 from pyee.asyncio import AsyncIOEventEmitter
@@ -29,22 +28,11 @@ from .queue import EventQueue
 Handler = TypeVar("Handler", bound=Callable[..., Awaitable[Any]])
 
 
-class BaseProtocol(ABC):
-    """协议层处理器接口：具体接入方式（websocket/webhook）实现后注册到主体。"""
-
-    @abstractmethod
-    def register(self, emitter: "EventEmitter") -> None: ...
-
-    def on_payload(self, payload: Payload) -> None:  # noqa: B027
-        """路由后的每条 payload 回调，供协议类跟踪协议级状态（如 s 序列号）。"""
-
-
 class EventEmitter:
     def __init__(self, queue: EventQueue | None = None) -> None:
         self._ee = AsyncIOEventEmitter()
         self._queue = queue if queue is not None else EventQueue()
         self.services: dict[type, Any] = {}
-        self._protocol: BaseProtocol | None = None
         # handler → (类型标注, 形参名)：签名解析较贵，缓存后每条事件只查表
         self._signatures: dict[
             Callable[..., Any], tuple[dict[str, Any], tuple[str, ...]]
@@ -57,10 +45,6 @@ class EventEmitter:
         """本 emitter 消费的事件队列；未显式传入时构造即自建，run_loop 复用同一实例。"""
         return self._queue
 
-    def register_protocol(self, protocol: BaseProtocol) -> None:
-        self._protocol = protocol
-        protocol.register(self)
-
     @overload
     def on(
         self, event: Opcode | str, *, background: bool = False
@@ -108,71 +92,34 @@ class EventEmitter:
                 # dispatch 是主循环，单条事件路由失败不能带崩整个进程
                 logger.exception(f"事件分发异常，已跳过: {data} ({e})")
 
-    def _route(self, data: Payload) -> tuple[int, str, Event] | None:
-        op = data.get("op")
-        if not isinstance(op, int):
-            return None
-
-        # 线上 payload 反序列化出来是 int，Opcode(9) 取名供非 DISPATCH 事件路由
-        event_name: str | None
-        if op == Opcode.DISPATCH:
-            t = data.get("t")
-            event_name = t if isinstance(t, str) else None
-        else:
-            try:
-                event_name = Opcode(op).name
-            except ValueError:
-                return None
-        if not isinstance(event_name, str):
-            return None
-
-        return op, event_name, Event(data)
-
     async def emit(self, data: Payload) -> None:
-        routed = self._route(data)
-        if routed is None:
-            logger.warning(f"无法确定事件名，已跳过: {data}")
+        # 先取事件名，无 listener 就不必构造 Event（每条未订阅/协议帧都省一次拷贝）
+        op = data.get("op")
+        if op != Opcode.DISPATCH:
+            logger.warning(f"非业务事件不应进分发队列，已跳过: {data}")
+            return
+        t = data.get("t")
+        if not isinstance(t, str):
+            logger.warning(f"业务事件缺少 t 字段，已跳过: {data}")
             return
-        op, event_name, event = routed
 
-        if self._protocol is not None:
-            self._protocol.on_payload(data)
-
-        listeners = self._ee.listeners(event_name)
+        listeners = self._ee.listeners(t)
         if not listeners:
             return
 
-        reply = op != Opcode.DISPATCH
+        event = Event(data)
         for fn in listeners:
             if fn in self._background:
                 self._spawn(fn, event)
             else:
-                await self._invoke(fn, event, reply=reply)
+                await self._invoke(fn, event)
 
-    async def handle(self, data: Payload) -> Any:
-        """同步分发事件并返回首个处理器的非空返回值，用于 op=13 等需要应答的协议事件。"""
-        routed = self._route(data)
-        if routed is None:
-            return None
-        _, event_name, event = routed
-
-        for fn in self._ee.listeners(event_name):
-            ret = await fn(*self._resolve_params(fn, event))
-            if ret is not None:
-                return ret
-        return None
-
-    async def _invoke(
-        self, fn: Callable[..., Awaitable[Any]], event: Event, reply: bool
-    ) -> None:
+    async def _invoke(self, fn: Callable[..., Awaitable[Any]], event: Event) -> None:
         # 用户 handler 的异常在这里就地隔离，不影响同一事件的其他 handler
         try:
-            ret: Any = await fn(*self._resolve_params(fn, event))
+            await fn(*self._resolve_params(fn, event))
         except Exception as e:
             logger.exception(f"事件处理器 {getattr(fn, '__qualname__', fn)} 异常: {e}")
-            return
-        if reply and isinstance(ret, dict):
-            await self._queue.put_reply(cast(Payload, ret))
 
     def _spawn(self, fn: Callable[..., Awaitable[Any]], event: Event) -> None:
         """handler 进后台任务，emit 不等它；任务由 _bg_tasks 持强引用防 GC。"""

src/qqbotsdk/events.py

@@ -7,14 +7,13 @@ from .payloads import parse_event
 
 
 class Event:
-    """一条 DISPATCH 事件；属性取值兼容群（openid）与频道两套字段名。"""
+    """一条业务事件（op=0）；属性取值兼容群（openid）与频道两套字段名。"""
 
     def __init__(self, payload: Payload) -> None:
         self._payload = payload
         t = payload.get("t")
         self.type: str | None = t if isinstance(t, str) else None
-        d = payload.get("d")
-        self._d: dict[str, Any] = dict(d) if isinstance(d, dict) else {}
+        self._d: dict[str, Any] | None = None
         self._typed: object | None = None
 
     @property
@@ -23,30 +22,34 @@ class Event:
 
     @property
     def data(self) -> dict[str, Any]:
+        """d 的只读副本；仅用到 `.typed`/`.raw` 的 handler 不会触发这次拷贝。"""
+        if self._d is None:
+            d = self._payload.get("d")
+            self._d = dict(d) if isinstance(d, dict) else {}
         return self._d
 
     @property
     def typed(self) -> object:
         """d 的 dataclass 解析结果（见 payloads.py）；未知事件类型为 dict。"""
         if self._typed is None:
-            self._typed = parse_event(self.type, self._d)
+            self._typed = parse_event(self.type, self.data)
         return self._typed
 
     @property
     def user_id(self) -> str | None:
-        return self._d.get("user_id") or self._d.get("from_user_id")
+        return self.data.get("user_id") or self.data.get("from_user_id")
 
     @property
     def group_id(self) -> str | None:
-        return self._d.get("group_openid")
+        return self.data.get("group_openid")
 
     @property
     def content(self) -> str | None:
-        return self._d.get("content")
+        return self.data.get("content")
 
     @property
     def message_id(self) -> str | None:
-        return self._d.get("id")
+        return self.data.get("id")
 
     @property
     def event_id(self) -> str | None:

src/qqbotsdk/protocol.py

@@ -0,0 +1,25 @@
+"""协议帧处理器接口：由连接适配器逐帧调用，不走事件队列。
+
+只有业务事件（op=0）进入事件队列交给 `EventEmitter` 分发给用户 handler；
+HELLO/READY/INVALID_SESSION/op=13 这类协议帧在适配器内就地处理，因为它们是
+握手续传语义——混进业务分发既让用户 handler 看到不该看的协议帧，也让"断线
+前重置会话"这类时序取决于分发泵的进度。各接入方式的实现见 `ws_protocol.py`
+与 `webhook_protocol.py`。
+"""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+from .model import Payload
+
+
+class BaseProtocol(ABC):
+    """适配器专有的协议处理器：实现方返回需要回送的响应体。"""
+
+    @abstractmethod
+    async def on_frame(self, payload: Payload) -> Payload | dict[str, Any] | None:
+        """处理一条帧，返回需回送的响应体（无需回送则 None）。
+
+        IDENTIFY/RESUME 等握手应答、op=13 验证应答都经返回值回送；业务事件
+        （op=0）不必处理，连接器会照常入队分发。
+        """

src/qqbotsdk/webhook_protocol.py

@@ -1,11 +1,15 @@
-"""Webhook 接入的协议层：op=13 验证请求应答（签名 event_ts + plain_token）。"""
+"""Webhook 接入的协议层：op=13 验证请求应答（签名 event_ts + plain_token）。
 
-from typing import override
+由 `WebhookConnecter` 在验签前就地调用；返回的应答体由连接器直接写回 HTTP
+响应，不走 reply 队列。
+"""
+
+from typing import Any, cast, override
 
 from .config import Config
 from .crypto import sign_msg
-from .emitter import BaseProtocol, EventEmitter
-from .model import Opcode, ValidationData
+from .model import Opcode, Payload, ValidationData
+from .protocol import BaseProtocol
 
 
 class WebhookProtocol(BaseProtocol):
@@ -13,10 +17,13 @@ class WebhookProtocol(BaseProtocol):
         self._config = config
 
     @override
-    def register(self, emitter: EventEmitter) -> None:
-        emitter.on(Opcode.VALIDATION)(self.validation)
-
-    async def validation(self, data: ValidationData) -> dict[str, str]:
+    async def on_frame(self, payload: Payload) -> dict[str, Any] | None:
+        if payload.get("op") != Opcode.VALIDATION:
+            return None
+        d = payload.get("d")
+        if not isinstance(d, dict):
+            return None
+        data = cast(ValidationData, d)
         return {
             "plain_token": data["plain_token"],
             "signature": sign_msg(

src/qqbotsdk/ws_protocol.py

@@ -1,8 +1,10 @@
-"""WebSocket 接入的协议层：IDENTIFY/RESUME 鉴权、序列号跟踪与会话保持。
+"""WebSocket 接入的协议层：握手鉴权、会话捕获与失效重置。
 
-心跳与连接生命周期归 WebsocketConnecter；本层只负责会话语义——
-识别 HELLO 后按有无 session_id 决定 RESUME 续传或 IDENTIFY 新建，
-收到不可恢复的 INVALID_SESSION（d=false）才清空会话重新 IDENTIFY。
+由 `WebsocketConnecter` 逐帧调用，不经过事件队列：HELLO → 按有无
+session_id 决定 RESUME 续传或 IDENTIFY 新建并返回应答；READY → 捕获
+session_id（该帧同时是 op=0，仍会入队分发给用户的 READY handler）；
+INVALID_SESSION → d=false 时清空会话，使重连走 IDENTIFY；每条帧的 s
+在此更新，供心跳与 RESUME 取用。
 """
 
 from tomllib import load
@@ -11,16 +13,8 @@ from typing import override
 from loguru import logger
 
 from .config import Config
-from .emitter import BaseProtocol, EventEmitter
-from .model import (
-    IdentifyData,
-    Intent,
-    Opcode,
-    Payload,
-    ResumeData,
-    payload_of,
-)
-from .payloads import Ready
+from .model import IdentifyData, Intent, Opcode, Payload, ResumeData, payload_of
+from .protocol import BaseProtocol
 from .session import Session
 from .token import AccessToken
 
@@ -67,18 +61,21 @@ class WebsocketProtocol(BaseProtocol):
         self._token = token
 
     @override
-    def register(self, emitter: EventEmitter) -> None:
-        emitter.on(Opcode.HELLO)(self.identify)
-        emitter.on("READY")(self.ready)
-        emitter.on(Opcode.INVALID_SESSION)(self.invalid_session)
-
-    @override
-    def on_payload(self, payload: Payload) -> None:
+    async def on_frame(self, payload: Payload) -> Payload | None:
         seq = payload.get("s")
         if isinstance(seq, int):
             self._session.update_sequence(seq)
 
-    async def identify(self) -> Payload:
+        op = payload.get("op")
+        if op == Opcode.HELLO:
+            return await self._handshake()
+        if op == Opcode.INVALID_SESSION:
+            self._invalid_session(payload.get("d"))
+        elif op == Opcode.DISPATCH and payload.get("t") == "READY":
+            self._ready(payload.get("d"))
+        return None
+
+    async def _handshake(self) -> Payload:
         access_token = await self._token.get_access_token()
 
         d: ResumeData | IdentifyData
@@ -101,11 +98,15 @@ class WebsocketProtocol(BaseProtocol):
 
         return payload_of(op, d)
 
-    async def ready(self, ready: Ready) -> None:
-        self._session.session_id = ready.session_id
-        logger.info("会话已建立（READY）")
+    def _ready(self, d: object) -> None:
+        if isinstance(d, dict):
+            session_id = d.get("session_id")
+            if isinstance(session_id, str):
+                self._session.session_id = session_id
+                logger.info("会话已建立（READY）")
 
-    async def invalid_session(self, resumable: bool) -> None:
-        if not resumable:
+    def _invalid_session(self, d: object) -> None:
+        # 官方语义：d 为布尔，true=可 RESUME 续传
+        if not d:
             self._session.session_id = None
             logger.warning("会话已失效，下次连接将重新 IDENTIFY")

tests/test_emitter.py

@@ -2,14 +2,12 @@ import asyncio
 
 import pytest
 
-from qqbotsdk.config import Config
 from qqbotsdk.emitter import EventEmitter
+from qqbotsdk.events import Event
 from qqbotsdk.model import Opcode, payload_of
 from qqbotsdk.payloads import GroupAtMessage
 from qqbotsdk.queue import EventQueue
 from qqbotsdk.session import Session
-from qqbotsdk.token import AccessToken
-from qqbotsdk.ws_protocol import WebsocketProtocol
 
 
 def test_sequence_only_moves_forward():
@@ -41,7 +39,8 @@ async def test_handler_exception_is_isolated():
 
 
 @pytest.mark.asyncio
-async def test_reply_payload_is_enqueued():
+async def test_protocol_frame_never_reaches_business_handler():
+    """协议帧改由适配器的协议处理器就近处理，不再经 emitter 回流应答。"""
     queue = EventQueue()
     emitter = EventEmitter(queue)
 
@@ -50,9 +49,7 @@ async def test_reply_payload_is_enqueued():
         return payload_of(Opcode.HEARTBEAT, 42)
 
     await emitter.emit({"op": Opcode.HEARTBEAT, "d": 0})
-    reply = await queue.get_reply()
-    assert reply["op"] == Opcode.HEARTBEAT
-    assert reply["d"] == 42
+    assert queue.reply_queue.empty()
 
 
 @pytest.mark.asyncio
@@ -126,40 +123,23 @@ async def test_service_registered_after_handler_still_injects():
 
 
 @pytest.mark.asyncio
-async def test_scalar_d_falls_back_to_raw_value():
-    """回归：op=9 INVALID_SESSION 的 d 是布尔值，DI 回退必须传原始 d。
-    此前实现把 d dict 化成 {}（恒 falsy），resumable=true 也被当成不可续传。
-    走真实协议层 handler 验证路由 → 注入 → 会话状态的全链路。"""
-    config = Config(
-        app_id="a",
-        app_secret="s",
-        base_url="https://x",
-        timeout=None,  # type: ignore[arg-type]
-        connecter="websocket",
-        intents_file="no-such.toml",
-        webhook_host="0.0.0.0",
-        webhook_port=8080,
-        webhook_path="/x",
-    )
+async def test_typed_and_data_see_same_content():
+    """`.typed` 与 `.data` 均可用且一致；只取 `.typed` 的 handler 不触发 data 拷贝。"""
     emitter = EventEmitter()
-    session = Session()
-    session.session_id = "S1"
-    emitter.register_protocol(
-        WebsocketProtocol(config, session, AccessToken(config, None))  # type: ignore[arg-type]
+    seen: list[tuple[str | None, str | None]] = []
+
+    @emitter.on("GROUP_AT_MESSAGE_CREATE")
+    async def handler(msg: GroupAtMessage, event: Event):
+        seen.append((msg.content, event.data["content"]))
+
+    await emitter.emit(
+        {
+            "op": Opcode.DISPATCH,
+            "t": "GROUP_AT_MESSAGE_CREATE",
+            "d": {"content": "hi", "group_openid": "G1"},
+        }
     )
-    seen: list[bool] = []
-
-    @emitter.on(Opcode.INVALID_SESSION)
-    async def spy(resumable: bool):
-        seen.append(resumable)
-
-    await emitter.emit({"op": Opcode.INVALID_SESSION, "d": True})
-    assert seen == [True]
-    assert session.session_id == "S1"  # 可续传，会话保留
-
-    await emitter.emit({"op": Opcode.INVALID_SESSION, "d": False})
-    assert seen == [True, False]
-    assert session.session_id is None  # 不可续传，会话清空
+    assert seen == [("hi", "hi")]
 
 
 @pytest.mark.asyncio

tests/test_route.py

@@ -1,36 +1,128 @@
+"""协议帧处理测试：协议处理器（WebsocketProtocol）就近处理握手续传语义，
+不进事件分发；只有业务事件（op=0）由 emitter 分发给用户 handler。"""
+
 import pytest
 
+from qqbotsdk.config import Config
 from qqbotsdk.emitter import EventEmitter
 from qqbotsdk.model import Opcode
+from qqbotsdk.session import Session
+from qqbotsdk.ws_protocol import WebsocketProtocol
 
 
-@pytest.fixture
-def emitter() -> EventEmitter:
-    return EventEmitter()  # 无参构造，覆盖 queue 默认自建路径
+class FakeToken:
+    """手搓 token 源，避免构造真实 AccessToken 去发网络请求。"""
+
+    async def get_access_token(self) -> str:
+        return "tok"
 
 
-def test_route_dispatch_uses_t(emitter: EventEmitter):
-    routed = emitter._route({"op": Opcode.DISPATCH, "t": "READY", "d": {}})
-    assert routed is not None
-    op, name, event = routed
-    assert op == Opcode.DISPATCH
-    assert name == "READY"
-    assert event.type == "READY"
+def make_protocol(session: Session | None = None) -> tuple[WebsocketProtocol, Session]:
+    session = session or Session()
+    config = Config(
+        app_id="app",
+        app_secret="secret",
+        base_url="https://api.bot.qq.com",
+        timeout=None,  # type: ignore[arg-type]
+        connecter="websocket",
+        intents_file="no-such.toml",
+        webhook_host="0.0.0.0",
+        webhook_port=8080,
+        webhook_path="/x",
+    )
+    return WebsocketProtocol(config, session, FakeToken()), session  # type: ignore[arg-type]
 
 
-def test_route_non_dispatch_uses_opcode_name(emitter: EventEmitter):
-    routed = emitter._route({"op": Opcode.HELLO, "d": {"heartbeat_interval": 1}})
-    assert routed is not None
-    assert routed[1] == "HELLO"
+@pytest.mark.asyncio
+async def test_hello_without_session_identifies():
+    protocol, _ = make_protocol()
+    reply = await protocol.on_frame(
+        {"op": Opcode.HELLO, "d": {"heartbeat_interval": 1}}
+    )
+    assert reply is not None
+    assert reply["op"] == Opcode.IDENTIFY
+    assert reply["d"]["token"] == "QQBot tok"
+    assert reply["d"]["shard"] == (0, 1)
 
 
-def test_route_rejects_missing_op(emitter: EventEmitter):
-    assert emitter._route({"d": {}}) is None
+@pytest.mark.asyncio
+async def test_hello_with_session_resumes():
+    protocol, session = make_protocol()
+    session.session_id = "S1"
+    session.update_sequence(7)
+    reply = await protocol.on_frame({"op": Opcode.HELLO, "d": {}})
+    assert reply is not None
+    assert reply["op"] == Opcode.RESUME
+    assert reply["d"]["session_id"] == "S1"
+    assert reply["d"]["seq"] == 7
 
 
-def test_route_rejects_unknown_opcode(emitter: EventEmitter):
-    assert emitter._route({"op": 999}) is None
+@pytest.mark.asyncio
+async def test_ready_captures_session_without_reply():
+    protocol, session = make_protocol()
+    reply = await protocol.on_frame(
+        {"op": Opcode.DISPATCH, "t": "READY", "d": {"session_id": "S9"}}
+    )
+    assert reply is None
+    assert session.session_id == "S9"
 
 
-def test_route_rejects_dispatch_without_t(emitter: EventEmitter):
-    assert emitter._route({"op": Opcode.DISPATCH, "d": {}}) is None
+@pytest.mark.asyncio
+async def test_sequence_is_tracked_from_frames():
+    protocol, session = make_protocol()
+    await protocol.on_frame({"op": Opcode.DISPATCH, "t": "X", "s": 5, "d": {}})
+    await protocol.on_frame(
+        {"op": Opcode.DISPATCH, "t": "X", "s": 3, "d": {}}
+    )  # 回退被忽略
+    assert session.seq == 5
+
+
+@pytest.mark.asyncio
+async def test_invalid_session_resets_only_when_not_resumable():
+    protocol, session = make_protocol()
+    session.session_id = "S1"
+    await protocol.on_frame({"op": Opcode.INVALID_SESSION, "d": True})
+    assert session.session_id == "S1"  # 可续传，会话保留
+    await protocol.on_frame({"op": Opcode.INVALID_SESSION, "d": False})
+    assert session.session_id is None  # 不可续传，下次重连走 IDENTIFY
+
+
+@pytest.mark.asyncio
+async def test_business_event_returns_no_reply():
+    """业务事件（op=0 非 READY）协议层不产生应答，只留给 emitter 分发。"""
+    protocol, _ = make_protocol()
+    reply = await protocol.on_frame(
+        {"op": Opcode.DISPATCH, "t": "GROUP_AT_MESSAGE_CREATE", "d": {"content": "hi"}}
+    )
+    assert reply is None
+
+
+@pytest.mark.asyncio
+async def test_emit_drops_protocol_frames():
+    """协议帧不进分发队列；万一进来也不得触达业务 handler。"""
+    emitter = EventEmitter()
+    called: list[str] = []
+
+    @emitter.on(Opcode.HELLO)
+    async def on_hello(_):
+        called.append("hello")
+
+    await emitter.emit({"op": Opcode.HELLO, "d": {"heartbeat_interval": 1}})
+    assert called == []
+
+
+@pytest.mark.asyncio
+async def test_emit_drops_dispatch_without_t():
+    emitter = EventEmitter()
+
+    @emitter.on("READY")
+    async def on_ready(_):
+        raise AssertionError("不应被调用")
+
+    await emitter.emit({"op": Opcode.DISPATCH, "d": {}})
+
+
+@pytest.mark.asyncio
+async def test_emit_ignores_missing_op():
+    emitter = EventEmitter()
+    await emitter.emit({"d": {}})  # 无 op，静默跳过

tests/test_webhook_integration.py

@@ -51,8 +51,7 @@ def event_body(event_id: str, content: str) -> dict[str, Any]:
 async def test_webhook_end_to_end():
     config = make_config(0)
     emitter = EventEmitter()
-    emitter.register_protocol(WebhookProtocol(config))
-    connecter = WebhookConnecter(config, emitter.queue, emitter.handle)
+    connecter = WebhookConnecter(config, emitter.queue, WebhookProtocol(config))
 
     received: list[GroupAtMessage] = []
 

tests/test_ws_integration.py

@@ -11,6 +11,7 @@ from aiohttp import ClientSession, WSMsgType, web
 from qqbotsdk.config import Config
 from qqbotsdk.connecter import WebsocketConnecter
 from qqbotsdk.emitter import EventEmitter
+from qqbotsdk.model import Opcode
 from qqbotsdk.payloads import GroupAtMessage
 from qqbotsdk.queue import EventQueue
 from qqbotsdk.session import Session
@@ -115,16 +116,25 @@ async def test_websocket_end_to_end():
         session = Session()
         emitter = EventEmitter(queue)
         protocol = WebsocketProtocol(config, session, token)
-        emitter.register_protocol(protocol)
-        connecter = WebsocketConnecter(config, http, token, session, queue)
+        connecter = WebsocketConnecter(config, http, token, session, queue, protocol)
 
         received: list[GroupAtMessage] = []
+        # 协议帧（HELLO/HEARTBEAT_ACK）绝不该触达 handler——只有 op=0 业务事件入队
+        protocol_frames: list[object] = []
 
         @emitter.on("GROUP_AT_MESSAGE_CREATE")
         async def on_msg(msg: GroupAtMessage) -> None:
             received.append(msg)
             gw.got_dispatch.set()
 
+        @emitter.on(Opcode.HELLO)
+        async def on_hello(_) -> None:
+            protocol_frames.append("HELLO")
+
+        @emitter.on(Opcode.HEARTBEAT_ACK)
+        async def on_ack(_) -> None:
+            protocol_frames.append("HEARTBEAT_ACK")
+
         dispatch_task = asyncio.create_task(emitter.dispatch())
         connect_task = asyncio.create_task(connecter.run())
         try:
@@ -146,6 +156,8 @@ async def test_websocket_end_to_end():
             assert len(received) == 1
             assert received[0].content == "ping"
             assert received[0].group_openid == "G1"
+            # 协议帧（HELLO/HEARTBEAT_ACK）没进分发路径
+            assert protocol_frames == []
         finally:
             dispatch_task.cancel()
             connect_task.cancel()