From c3a227c6dfabd70a15813c602a014ecc5d42c040 Mon Sep 17 00:00:00 2001 From: zijunchiang Date: Thu, 17 Sep 2026 01:11:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(net):=20=E8=BF=9E=E6=8E=A5=E4=BF=9D?= =?UTF-8?q?=E6=B4=BB=20/=20=E8=AE=BE=E5=A4=87=E5=8E=BB=E9=87=8D=20/=20?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E7=8A=B6=E6=80=81=E6=9C=BA=EF=BC=88T-WIN-015?= =?UTF-8?q?=EF=BC=8C=E5=A5=91=E7=BA=A6=20websocket-keypoints=200.2.0?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 与 Android 端 T-AND-011「两端一起改」：契约先在 opendance-web 修改并提升到 websocket-keypoints 0.2.0（向后兼容：子协议名与二进制帧 version 均不变， 只新增可选控制字段与一种新控制类型），再同步三端快照。 修掉本端的 3 处根因/放大器： 1. 静默超时 5s 过短且未声明，客户端无从得知。OkHttp 的协议层 ping 由底层库 自动应答、**不暴露给应用层**，无法刷新 last_seen_ms，于是"画面里没人"时 每 5 秒必然踢一次连接 -> 客户端重连 -> 再被踢。 - CLIENT_TIMEOUT_MS 5s -> 15s，新增 HEARTBEAT_INTERVAL_MS = 5s； - welcome_payload() 在 welcome 中声明 silentTimeoutMs / heartbeatIntervalMs， config 广播同样带上（契约 §1.1：心跳必须 < 静默超时的一半，5000*2 < 15000）； - 处理应用层 ping 并回 pong（回显 sentAtMs + serverAtMs，供对端算 RTT）。 2. 反复重连被计成多台设备。按契约 §5 用 hello.deviceId 去重：同一设备只保留 最新连接，旧连接以 close(1000) 关闭——刻意用 1000，因为对端按契约 §7 收到 1000 不会重连，否则会形成"挤掉-重连-再挤掉"的循环； 新增 device_count() 与 device_replaced 信号，界面一律显示去重后的设备数。 3. 只有 handshaken 布尔量，日志看不出卡在哪一步。按契约 §8 逐客户端维护与 Android 端同名的状态（connecting -> handshaking -> streaming -> degraded， 主动停止/超时/被替换为 stopped），并记录 close code/reason、协商子协议、 连接/握手/首帧时间；degraded 保持时长做成可覆盖字段便于测试。 验收测试（tests/test_pose_server.gd 新增 7 项，总计 720 通过 / 0 失败）： welcome 声明保活参数、ping->pong 回显、同 deviceId 去重（旧连接 1000 关闭且 设备数仍为 1）、8 个状态名与契约一致、状态随握手/首帧推进到 streaming、 丢帧进入 degraded 后恢复、snapshot 暴露设备数与保活参数。 同时更新 README/README.txt/tasks 与 Credits 的契约版本（0.2.0）。 --- README.md | 6 +- README.txt | 10 +- contracts/VERSION | 2 +- contracts/websocket-keypoints.md | 138 +++++++++++++++++++++- project.yaml | 6 +- src/autoload/pose_receiver.gd | 19 +++ src/net/pose_server.gd | 191 +++++++++++++++++++++++++++++-- src/ui/credits.gd | 2 +- src/ui/game_screen.gd | 3 +- src/ui/main_menu.gd | 17 ++- tasks/015-keepalive-dedup.md | 89 ++++++++++++++ tests/test_pose_server.gd | 189 ++++++++++++++++++++++++++++++ 12 files changed, 647 insertions(+), 25 deletions(-) create mode 100644 tasks/015-keepalive-dedup.md diff --git a/README.md b/README.md index b00c0c1..38b13e0 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ OpenDance Studio **阶段二**: Windows 客户端（Godot 4）。 | 骨骼重定向 | OK 关键点方向与骨骼实际指向点积 **1.0000** | | 跨实现交叉校验 | OK 关节角与 Web 服务端 TypeScript 实现**逐帧一致** | | 场景选择与物理落地 | OK `scenes[]` 可选、重力/风/摩擦/碰撞体落地并可日志核对 | +| 连接保活与设备去重 | OK welcome/config 声明静默超时与心跳；应用层 ping/pong；按 deviceId 去重 | | 发布整包 | OK 单个 `opendance-windows-v0.2.0.zip` + `SHA256SUMS`(不拆分卷) | | 版本号展示 | OK 「关于」清单显示 `v0.2.0`，来源唯一(`project.godot`) | @@ -103,7 +104,8 @@ http://192.0.2.10:3000 - **Android 侧无本端可配置项**：接收地址由 Android App 自己填写；本端只监听固定端口 8765（可在 `opendance.cfg` 改 `listen_port` 后点「重新连接接收服务」生效）。 - 已知限制：Godot 的 `WebSocketPeer` 不暴露握手请求路径，本端**接受任意 path**，只校验子协议 `opendance.pose.v1`（见「已知限制」1）。 -- 接收状态可用主界面「**查看接收状态**」查看：连接数、累计帧、丢帧、畸形帧。 +- 接收状态可用主界面「**查看接收状态**」查看：连接数、**去重后设备数**、累计帧、 + 丢帧、畸形帧、保活参数，以及每台设备的状态（契约 §8）与首帧时间。 ### 6. 打包导出 @@ -188,7 +190,7 @@ src/ |--|--| | `rest-api` | 0.5.0 | | `data-models` | 0.7.0 | -| `websocket-keypoints` | 0.1.0 | +| `websocket-keypoints` | 0.2.0 | | `skeleton-topology` | 0.2.0 | | `model-formats` | 0.5.0 | | `database` | 0.4.0 | diff --git a/README.txt b/README.txt index 69bd7db..6d0dba0 100644 --- a/README.txt +++ b/README.txt @@ -254,7 +254,7 @@ OpenDance.exe 的用户看第一节即可。 export_presets.cfg 导出预设（必须在项目根目录） project.yaml 任务清单与状态 DSH.md 本项目 DSH 执行约束 - tasks/ 单任务说明书(T-WIN-001 ~ 014) + tasks/ 单任务说明书(T-WIN-001 ~ 015) contracts/ 只读契约快照(来源 opendance-web；六项契约各自独立版本) scenes/main.tscn 根节点 src/autoload/ AppConfig / CatalogStore / PoseReceiver / ScoreService @@ -285,6 +285,14 @@ A: 游戏界面会显示「人形覆盖率」。低于 60% 说明模型的骨骼 Q: 连上了手机但分数一直是 0 A: 确认手机端已开始推流；主界面的「查看接收状态」会显示累计帧数与丢帧数。 +Q: 手机端反复重连 / 「已连接」数忽大忽小 +A: 本端有 15 秒静默超时（契约 v0.2.0 1.1），并在 welcome/config 里声明 + silentTimeoutMs=15000 与 heartbeatIntervalMs=5000；Android 端会按后者发 + 应用层 ping。若仍频繁断开，用「查看接收状态」看每台设备的状态与断开原因 + （close code）：1001 表示本端主动踢人或服务重启，0/1006 表示网络抖动。 + 同一设备的重复连接会按 deviceId 去重（旧连接以 1000 关闭，对端不重连）， + 所以「已连接 N 台」显示的是去重后的设备数，不是连接数。 + Q: 选场景时提示「场景缺少 GLB 文件」或「已回退内置空场景」 A: 服务端该场景没有可下载的 GLB（转换失败或未发布）。这不影响开始跟练， 会使用内置空场景与默认物理参数；请让管理员在管理端「场景管理」重新上传。 diff --git a/contracts/VERSION b/contracts/VERSION index 16044a8..3b1ce3a 100644 --- a/contracts/VERSION +++ b/contracts/VERSION @@ -1,7 +1,7 @@ # OpenDance Studio 契约版本 # 规则：任何字段级变更必须提升对应条目版本，并回到 opendance-web 修改后再向另两端建迁移任务。 rest-api: 0.5.0 -websocket-keypoints: 0.1.0 +websocket-keypoints: 0.2.0 data-models: 0.7.0 skeleton-topology: 0.2.0 model-formats: 0.5.0 diff --git a/contracts/websocket-keypoints.md b/contracts/websocket-keypoints.md index f6ed088..0af36c3 100644 --- a/contracts/websocket-keypoints.md +++ b/contracts/websocket-keypoints.md @@ -1,4 +1,4 @@ -# contracts/websocket-keypoints.md - 关键点传输契约 v0.1.0 +# contracts/websocket-keypoints.md - 关键点传输契约 v0.2.0 Android 辅助 App（发送端）-> Windows 客户端（接收端）。 服务端不参与实时链路，但 T-WEB-006 预处理产出的序列必须与此结构同构。 @@ -14,6 +14,22 @@ ws://:8765/pose - 二进制帧传输关键点，文本帧传输控制消息（JSON）。 - 默认 30 FPS；发送端允许丢帧，接收端必须按 `frame_id` 检测丢包并插值补齐。 +### 1.1 保活（v0.2.0 新增，向后兼容） + +接收端有**静默超时**：超过 `silentTimeoutMs` 没有收到任何数据帧或文本帧即断开 +（`close(1001, "timeout")`）。为此： + +- 接收端必须在 `welcome`（以及需要变更时的 `config`）中声明 + `silentTimeoutMs` 与期望的 `heartbeatIntervalMs`； +- 发送端按 `heartbeatIntervalMs` 周期发送 `ping`（见 §3.6）， + 且 **`heartbeatIntervalMs` 必须小于 `silentTimeoutMs` 的一半**； +- 发送端每帧关键点与每秒 `stats` 同样算作活动，`ping` 只是兜底， + 保证「画面里没有人 / 相机暂停」时连接不会被误判掉线。 + +> 子协议名与二进制帧 `version` 均**不变**：本版本只**新增可选**控制字段与 +> 一种新的控制类型，遵循 §5 的兼容规则；仅实现 v0.1.0 的一端仍可正常互通 +> （它不发 `ping`、不读 `silentTimeoutMs`，但每帧与每秒 stats 已足够保活）。 + ## 2. 二进制帧：239 字节（冻结） **小端序(little-endian)**。 @@ -76,26 +92,54 @@ func decode(buf: PackedByteArray) -> Dictionary: { "type": "hello", "protocol": "opendance.pose.v1", + "deviceId": "a1b2c3d4e5f60718", "device": "Pixel 8", - "appVersion": "0.1.0", + "appVersion": "0.2.0", "cameraFacing": "front", "targetFps": 30, "resolution": "640x480" } ``` +- `deviceId`（v0.2.0 新增，**可选但强烈建议**）：该安装实例的稳定标识 + （建议 16 位十六进制，取自 `ANDROID_ID` 的哈希，不使用任何硬件序列号）。 + 接收端**必须按 `deviceId` 去重**：同一 `deviceId` 的新连接替换旧连接， + 设备数按去重后计数（否则一次「反复重连」会被显示成多台设备）。 + 未提供 `deviceId` 时接收端退回按连接计数。 + ### 3.2 `welcome` - Windows -> Android ```json -{ "type": "welcome", "protocol": "opendance.pose.v1", "sessionId": "uuid", "expectedFps": 30 } +{ + "type": "welcome", + "protocol": "opendance.pose.v1", + "sessionId": "uuid", + "expectedFps": 30, + "silentTimeoutMs": 15000, + "heartbeatIntervalMs": 5000 +} ``` +- `silentTimeoutMs` / `heartbeatIntervalMs`（v0.2.0 新增）：见 §1.1。 + 发送端若未读到这两个字段，应使用保守默认值（超时 5000、心跳 1000）。 + ### 3.3 `config` - Windows -> Android，运行中调整 ```json -{ "type": "config", "mirror": true, "targetFps": 30, "upperBodyOnly": false, "resolution": "640x480" } +{ + "type": "config", + "mirror": true, + "targetFps": 30, + "upperBodyOnly": false, + "resolution": "640x480", + "silentTimeoutMs": 15000, + "heartbeatIntervalMs": 5000 +} ``` +- `silentTimeoutMs` / `heartbeatIntervalMs` 为 v0.2.0 新增的可选字段， + 接收端在运行中调整保活参数时一并下发。 + ### 3.4 `stats` - Android -> Windows，每秒一次 ```json @@ -108,6 +152,31 @@ func decode(buf: PackedByteArray) -> Dictionary: { "type": "bye", "reason": "user_stop" } ``` +`bye` 表示**主动**结束：发出方在发送后立即关闭连接，接收方据此关闭且 +**不得**自动重连（见 §6）。 + +### 3.6 `ping` / `pong` - 保活（v0.2.0 新增） + +Android -> Windows： + +```json +{ "type": "ping", "sentAtMs": 1730000000000 } +``` + +Windows -> Android（可选回复，用于 RTT）： + +```json +{ "type": "pong", "sentAtMs": 1730000000000, "serverAtMs": 1730000000042 } +``` + +- `sentAtMs` 为发送方的本地毫秒时间戳，原样回传；发送端用 + `now - sentAtMs` 估算 RTT。 +- 接收端收到任何一帧（二进制或文本）都算活动，`ping` 无需额外应答即可保活； + `pong` 只用于诊断。 +- **不要**依赖 WebSocket 协议层的 ping 帧（`Sec-WebSocket-Ping`）：它由底层 + 库自动处理，**不会**暴露给应用层，因此不能被接收端的静默超时逻辑看到。 + 这也是必须定义应用层 `ping` 的原因。 + ## 4. JSON 调试帧（可选，仅开发期） ```json @@ -123,9 +192,68 @@ func decode(buf: PackedByteArray) -> Dictionary: 生产链路必须使用二进制帧；JSON 帧仅用于联调，接收端可选择性支持。 -## 5. 兼容性规则 +## 5. 设备去重（v0.2.0 新增） + +接收端按 §3.1 的 `deviceId` 去重： + +| 情况 | 行为 | +|--|--| +| 新连接的 `deviceId` 已存在 | 关闭旧连接（`close(1000, "replaced by new connection")`），保留新连接；设备数不变 | +| `deviceId` 缺失或为空 | 不做去重，按连接计数 | +| 同一 `deviceId` 的旧连接已断开 | 正常替换，不影响设备数 | + +这样「手机闪断后立刻重连」不会在接收端显示成两台设备，也不会让 +「已连接 N 台」抖动。 + +## 6. 兼容性规则 - `version` 不同的大版本接收端必须拒绝并在 `bye.reason` 中说明。 - `flags` 新增位不得改变既有位语义；旧接收端必须忽略未知位。 - 33 点顺序一旦冻结不得插入或重排；新增点只能追加新 `version`。 - 服务端 `motions.keypoint_format` 字段取值 `mediapipe33@0.1.0` 对应本契约。 + +## 7. 关闭码与重连策略（v0.2.0 新增，两端必须一致） + +| 关闭码 / 情形 | 含义 | 发送端是否自动重连 | +|--|--|--| +| `1000` 正常关闭 | 任一方主动结束（含收到 `bye`） | **否** | +| `1001` going away | 接收端主动关闭（服务重启、`stop()`、静默超时踢人） | **否**（见下方说明） | +| `1002` 协议错误 | 子协议 / 协议版本不匹配 | **否**（重连也不会成功） | +| 连接失败 / 超时 / `1006` 异常断开 | 网络抖动、对端进程退出、Wi-Fi 切换 | **是** | +| 客户端主动 `disconnect()` | 用户点「结束推流」 | **否** | + +- 静默超时由接收端发 `close(1001, "timeout")`；正常情况下发送端按 §1.1 保活， + **不应**触发它。若仍频繁触发，属于保活参数或实现缺陷，需按 §7 排查， + 而不是靠重连掩盖。 +- 接收端因「同一 `deviceId` 被新连接替换」关闭旧连接时用 `1000`， + 旧连接不重连。 + +### 7.1 退避参数（冻结） + +``` +第 n 次重连等待 = min(500ms * 2^(n-1), 15000ms) * (1 ± 20% 抖动) +即 0.5s → 1s → 2s → 4s → 8s → 15s（封顶），每次叠加 ±20% 抖动 +连接成功后计数清零 +``` + +## 8. 连接状态机（v0.2.0 新增，两端同名，便于对照日志） + +``` +idle → resolving → connecting → handshaking → streaming → degraded + ↑ │ + └──────── reconnecting ←───────────┘ +stopped ← 主动停止 / 1000 / 1001 / 1002（终态，不再重连） +``` + +| 状态 | 含义 | 进入条件（发送端） | +|--|--|--| +| `idle` | 尚未开始连接 | 初始状态 | +| `resolving` | 解析主机地址（DNS / 局域网发现） | `connect()` 且主机名非 IP 字面量 | +| `connecting` | TCP + WebSocket 升级中 | 已发起请求 | +| `handshaking` | 已连接，等待 `welcome` | `onOpen`，已发送 `hello` | +| `streaming` | 正常推流 | 收到 `welcome` 或已成功发送首帧 | +| `degraded` | 仍连接但异常（丢弃帧增长 / 发送失败） | `send` 失败或持续丢帧 | +| `reconnecting` | 退避等待中 | 非主动断开 | +| `stopped` | 已停止（终态） | 主动停止 / 1000 / 1001 / 1002 | + +接收端（Windows）在日志与状态面板中使用同名状态描述每个客户端连接。 diff --git a/project.yaml b/project.yaml index 199b9b6..6edaff5 100644 --- a/project.yaml +++ b/project.yaml @@ -5,7 +5,7 @@ default_branch: main engine: Godot 4.7.2 contracts_version: rest-api: 0.5.0 - websocket-keypoints: 0.1.0 + websocket-keypoints: 0.2.0 data-models: 0.7.0 skeleton-topology: 0.2.0 model-formats: 0.5.0 @@ -82,6 +82,10 @@ tasks: file: tasks/014-full-package.md scope: 发布整包（单个 zip + SHA256SUMS，不拆分卷） status: done + - id: T-WIN-015 + file: tasks/015-keepalive-dedup.md + scope: 连接保活 / 设备去重 / 状态机（契约 websocket-keypoints 0.2.0，与 T-AND-011 对齐） + status: done deliverable: artifact: opendance-windows-v0.2.0.zip diff --git a/src/autoload/pose_receiver.gd b/src/autoload/pose_receiver.gd index ceea2d6..cde53a1 100644 --- a/src/autoload/pose_receiver.gd +++ b/src/autoload/pose_receiver.gd @@ -62,6 +62,25 @@ func connected_client_count() -> int: return server.connected_count() if server != null else 0 +## 去重后的设备数（契约 §4.1）；界面显示「已连接 N 台」应当用它 +func connected_device_count() -> int: + return server.device_count() if server != null else 0 + + +## 逐客户端状态（契约 §8，用于界面展示与日志对照） +func clients_info() -> Array: + return server.clients_info() if server != null else [] + + +## 保活参数（契约 §1.1）：本端声明，客户端据此定心跳节奏 +func silent_timeout_ms() -> int: + return server.CLIENT_TIMEOUT_MS if server != null else 0 + + +func heartbeat_interval_ms() -> int: + return server.HEARTBEAT_INTERVAL_MS if server != null else 0 + + ## 设置要下发给 Android 端的参数（契约 §3.3 config） func apply_remote_config(config: Dictionary) -> void: if server == null: diff --git a/src/net/pose_server.gd b/src/net/pose_server.gd index fec3b36..d714fff 100644 --- a/src/net/pose_server.gd +++ b/src/net/pose_server.gd @@ -1,18 +1,27 @@ class_name PoseServer extends Node -## T-WIN-003：WebSocket 关键点接收服务 +## T-WIN-003 / T-AND-011：WebSocket 关键点接收服务 ## -## 契约 contracts/websocket-keypoints.md §1： +## 契约 contracts/websocket-keypoints.md v0.2.0 §1： ## 端点 ws://:8765/pose ## 子协议 opendance.pose.v1 ## 二进制帧 = 239 字节关键点（见 KeypointFrame） -## 文本帧 = JSON 控制消息（hello / welcome / config / stats / bye） +## 文本帧 = JSON 控制消息（hello / welcome / config / stats / ping / pong / bye） ## ## 本端为**服务端**，Android 辅助 App 主动连入。 ## ## 说明：Godot 的 WebSocketPeer 不暴露握手请求路径，因此无法校验客户端 ## 连接的 path 是否为 /pose；本实现接受任意 path，仅校验子协议。 +## +## T-AND-011 对齐要点（契约 v0.2.0 §1.1 / §4.1 / §8）： +## 1. 在 welcome / config 中**声明**静默超时与期望心跳间隔（silentTimeoutMs / +## heartbeatIntervalMs），且 heartbeatIntervalMs < silentTimeoutMs / 2； +## 2. 处理应用层 ping（协议层 Sec-WebSocket-Ping 在 Godot 里对应用层不可见， +## 不能用来续命），可选回 pong 供对端算 RTT； +## 3. 按 hello.deviceId **去重**：同一设备的重复连接替换旧连接（1000 关闭）， +## 设备数按去重后计数，避免"反复重连"被显示成多台设备； +## 4. 日志与状态面板使用契约 §8 的同名连接状态描述每个客户端。 signal listening(port: int) signal listen_failed(port: int, reason: String) @@ -21,16 +30,39 @@ signal client_disconnected(client_id: int, reason: String) signal frame_received(frame: KeypointFrame.PoseFrame) signal control_received(client_id: int, message: Dictionary) signal stats_updated(stats: Dictionary) +## 同一 deviceId 的旧连接被新连接替换（契约 §4.1） +signal device_replaced(device_id: String) const KeypointFrame = preload("res://src/net/keypoint_frame.gd") const DEFAULT_PORT := 8765 const DEFAULT_SUBPROTOCOL := "opendance.pose.v1" -## 超过该时长（毫秒）没有收到任何帧，判定为超时断开 -const CLIENT_TIMEOUT_MS := 5000 +## 静默超时（毫秒）：超过该时长没有收到**任何数据帧或文本帧**即断开。 +## 契约 §1.1 要求本端在 welcome/config 中声明该值，客户端据此定心跳节奏。 +## 取 15 秒：Android 每秒一次 stats、每帧关键点都算活动，15 秒足以容忍 +## 相机暂停 / 画面中无人 / 推理短暂卡顿，又不会让死连接长期占用。 +const CLIENT_TIMEOUT_MS := 15000 +## 期望心跳间隔（毫秒）——必须 < CLIENT_TIMEOUT_MS / 2（契约 §1.1） +const HEARTBEAT_INTERVAL_MS := 5000 ## 每秒向客户端推送一次 config 的间隔（0 表示不主动推送） const CONFIG_PUSH_INTERVAL_MS := 2000 +## degraded 状态的保持时长：这段时间内没有新的丢帧就回到 streaming。 +## 做成可覆盖的实例字段，便于测试在不依赖真实等待的前提下核对状态回收。 +const DEGRADED_HOLD_MS := 3000 +## 契约 §8 的连接状态名（本端为接收端，逐客户端维护） +enum ClientState { IDLE, RESOLVING, CONNECTING, HANDSHAKING, STREAMING, DEGRADED, RECONNECTING, STOPPED } + +const STATE_NAMES := { + ClientState.IDLE: "idle", + ClientState.RESOLVING: "resolving", + ClientState.CONNECTING: "connecting", + ClientState.HANDSHAKING: "handshaking", + ClientState.STREAMING: "streaming", + ClientState.DEGRADED: "degraded", + ClientState.RECONNECTING: "reconnecting", + ClientState.STOPPED: "stopped", +} var subprotocol: String = DEFAULT_SUBPROTOCOL var expected_fps: int = 30 @@ -44,6 +76,8 @@ var _next_client_id: int = 1 var _port: int = 0 var _running: bool = false var _last_config_push_ms: int = 0 +## 见 DEGRADED_HOLD_MS；运行期可调（测试用） +var degraded_hold_ms: int = DEGRADED_HOLD_MS ## 累计统计 var frames_total: int = 0 @@ -98,6 +132,7 @@ func stop() -> void: _running = false for entry in _clients: var peer: WebSocketPeer = entry["peer"] + entry["state"] = ClientState.STOPPED if peer.get_ready_state() == WebSocketPeer.STATE_OPEN: peer.close(1001, "server shutdown") _tcp.stop() @@ -134,10 +169,19 @@ func _accept_new_connections() -> void: "peer": peer, "handshaken": false, "device": "", + "device_id": "", + "state": ClientState.CONNECTING, + "selected_protocol": "", + "connected_at_ms": Time.get_ticks_msec(), + "handshake_at_ms": 0, + "first_frame_at_ms": 0, "last_seen_ms": Time.get_ticks_msec(), "last_frame_id": -1, "frames": 0, "dropped": 0, + "degraded_until_ms": 0, + "close_code": 0, + "close_reason": "", "fps_window_start_ms": Time.get_ticks_msec(), "fps_window_frames": 0, "fps": 0.0, @@ -155,12 +199,18 @@ func _poll_clients() -> void: var state := peer.get_ready_state() if state == WebSocketPeer.STATE_CLOSED: - client_disconnected.emit(entry["id"], "连接已关闭") + entry["close_code"] = peer.get_close_code() + entry["close_reason"] = peer.get_close_reason() + entry["state"] = ClientState.STOPPED + client_disconnected.emit(entry["id"], _close_reason_text(entry)) continue if state == WebSocketPeer.STATE_OPEN and not entry["handshaken"]: entry["handshaken"] = true + entry["handshake_at_ms"] = now + entry["state"] = ClientState.HANDSHAKING var selected := peer.get_selected_protocol() + entry["selected_protocol"] = selected if selected != "" and selected != subprotocol: push_warning("[PoseServer] 客户端选择子协议 %s，与期望 %s 不符" % [selected, subprotocol]) client_connected.emit(entry["id"], entry["device"]) @@ -169,15 +219,62 @@ func _poll_clients() -> void: _handle_packet(entry, peer) entry["last_seen_ms"] = now - # 超时剔除 + # 降级状态到期后回到 streaming（契约 §8） + if entry["state"] == ClientState.DEGRADED and now >= int(entry["degraded_until_ms"]): + entry["state"] = ClientState.STREAMING if int(entry["first_frame_at_ms"]) > 0 else ClientState.HANDSHAKING + + # 超时剔除：契约 §1.1，客户端应按 heartbeatIntervalMs 保活 if entry["handshaken"] and now - int(entry["last_seen_ms"]) > CLIENT_TIMEOUT_MS: - client_disconnected.emit(entry["id"], "超时未收到数据") + var silent_ms := now - int(entry["last_seen_ms"]) + entry["state"] = ClientState.STOPPED + client_disconnected.emit( + entry["id"], + "静默超时 %d ms（阈值 %d ms，期望心跳 %d ms）" % [silent_ms, CLIENT_TIMEOUT_MS, HEARTBEAT_INTERVAL_MS], + ) peer.close(1001, "timeout") continue survivors.append(entry) _clients = survivors + _dedup_devices() + + +## 契约 §4.1：同一 deviceId 只保留最新连接，旧连接以 1000 关闭（对端不重连） +func _dedup_devices() -> void: + if _clients.size() < 2: + return + var seen := {} + var keep: Array = [] + # 倒序扫描：后加入的条目更新，先被保留 + for i in range(_clients.size() - 1, -1, -1): + var entry: Dictionary = _clients[i] + var device := str(entry["device_id"]) + if device == "" or not entry["handshaken"]: + keep.append(entry) + continue + if seen.has(device): + entry["state"] = ClientState.STOPPED + var peer: WebSocketPeer = entry["peer"] + if peer.get_ready_state() == WebSocketPeer.STATE_OPEN: + peer.close(1000, "replaced by new connection") + client_disconnected.emit(entry["id"], "被同一设备的新连接替换（deviceId=%s）" % device) + device_replaced.emit(device) + continue + seen[device] = entry + keep.append(entry) + keep.reverse() + _clients = keep + + +static func _close_reason_text(entry: Dictionary) -> String: + var code := int(entry["close_code"]) + var reason := str(entry["close_reason"]) + if code == 0: + return "连接已关闭（未收到关闭帧）" + if reason == "": + return "连接已关闭（code=%d）" % code + return "连接已关闭（code=%d，%s）" % [code, reason] func _handle_packet(entry: Dictionary, peer: WebSocketPeer) -> void: @@ -208,6 +305,8 @@ func _handle_packet(entry: Dictionary, peer: WebSocketPeer) -> void: func _update_frame_stats(entry: Dictionary, frame: KeypointFrame.PoseFrame) -> void: + if int(entry["first_frame_at_ms"]) == 0: + entry["first_frame_at_ms"] = Time.get_ticks_msec() var previous := int(entry["last_frame_id"]) if previous >= 0 and frame.frame_id > previous + 1: # u32 回绕保护：差距过大视为计数器重置而非丢包 @@ -215,8 +314,13 @@ func _update_frame_stats(entry: Dictionary, frame: KeypointFrame.PoseFrame) -> v if gap < 100000: entry["dropped"] = int(entry["dropped"]) + gap frames_dropped += gap + # 契约 §8：仍在连接但出现丢帧 -> degraded（3 秒无新丢帧后自动回到 streaming） + entry["state"] = ClientState.DEGRADED + entry["degraded_until_ms"] = Time.get_ticks_msec() + degraded_hold_ms entry["last_frame_id"] = frame.frame_id entry["frames"] = int(entry["frames"]) + 1 + if entry["state"] != ClientState.DEGRADED: + entry["state"] = ClientState.STREAMING entry["fps_window_frames"] = int(entry["fps_window_frames"]) + 1 var now := Time.get_ticks_msec() @@ -232,6 +336,8 @@ func _update_frame_stats(entry: Dictionary, frame: KeypointFrame.PoseFrame) -> v "frames": entry["frames"], "dropped": entry["dropped"], "device": entry["device"], + "device_id": entry["device_id"], + "state": state_name(entry["state"]), "total_frames": frames_total, "total_dropped": frames_dropped, "malformed": malformed_frames, @@ -252,6 +358,7 @@ func _handle_control(entry: Dictionary, text: String) -> void: match type: "hello": entry["device"] = str(message.get("device", "")) + entry["device_id"] = str(message.get("deviceId", "")) var protocol := str(message.get("protocol", subprotocol)) if protocol != subprotocol: push_warning("[PoseServer] 协议版本不匹配：客户端 %s，本端 %s" % [protocol, subprotocol]) @@ -264,13 +371,17 @@ func _handle_control(entry: Dictionary, text: String) -> void: return if message.has("targetFps"): expected_fps = int(message["targetFps"]) + entry["state"] = ClientState.STREAMING if int(entry["first_frame_at_ms"]) > 0 else ClientState.HANDSHAKING + _send_json(entry, welcome_payload()) + "ping": + # 契约 §3.6：应用层心跳（协议层 ping 对应用层不可见，无法续命） _send_json(entry, { - "type": "welcome", - "protocol": subprotocol, - "sessionId": _session_id(), - "expectedFps": expected_fps, + "type": "pong", + "sentAtMs": int(message.get("sentAtMs", 0)), + "serverAtMs": Time.get_ticks_msec(), }) "bye": + entry["state"] = ClientState.STOPPED client_disconnected.emit(entry["id"], str(message.get("reason", "客户端主动断开"))) var peer2: WebSocketPeer = entry["peer"] peer2.close(1000, "bye") @@ -278,6 +389,18 @@ func _handle_control(entry: Dictionary, text: String) -> void: pass +## 契约 §3.2：welcome 必须声明保活参数，客户端据此定心跳节奏 +func welcome_payload() -> Dictionary: + return { + "type": "welcome", + "protocol": subprotocol, + "sessionId": _session_id(), + "expectedFps": expected_fps, + "silentTimeoutMs": CLIENT_TIMEOUT_MS, + "heartbeatIntervalMs": HEARTBEAT_INTERVAL_MS, + } + + func _maybe_push_config() -> void: if CONFIG_PUSH_INTERVAL_MS <= 0: return @@ -295,6 +418,9 @@ func broadcast_config() -> void: "targetFps": expected_fps, "upperBodyOnly": upper_body_only, "resolution": resolution, + # 契约 §3.3：运行中可一并调整保活参数 + "silentTimeoutMs": CLIENT_TIMEOUT_MS, + "heartbeatIntervalMs": HEARTBEAT_INTERVAL_MS, } for entry in _clients: if entry["handshaken"]: @@ -312,12 +438,53 @@ func _session_id() -> String: return "%d-%d" % [Time.get_unix_time_from_system(), _port] +## 状态名（契约 §8，两端同名，便于对照日志） +static func state_name(state: int) -> String: + return str(STATE_NAMES.get(state, "unknown")) + + +## 逐客户端信息（界面 / 日志用） +func clients_info() -> Array: + var out: Array = [] + for entry in _clients: + out.append({ + "id": entry["id"], + "device": entry["device"], + "device_id": entry["device_id"], + "state": state_name(entry["state"]), + "protocol": entry["selected_protocol"], + "frames": entry["frames"], + "dropped": entry["dropped"], + "fps": entry["fps"], + "first_frame_at_ms": entry["first_frame_at_ms"], + }) + return out + + +## 去重后的设备数（契约 §4.1：同一 deviceId 只算一台） +func device_count() -> int: + var ids := {} + var anonymous := 0 + for entry in _clients: + if not entry["handshaken"]: + continue + var device := str(entry["device_id"]) + if device == "": + anonymous += 1 + else: + ids[device] = true + return ids.size() + anonymous + + func snapshot() -> Dictionary: return { "running": _running, "port": _port, "clients": _clients.size(), "connected": connected_count(), + "devices": device_count(), + "silent_timeout_ms": CLIENT_TIMEOUT_MS, + "heartbeat_interval_ms": HEARTBEAT_INTERVAL_MS, "frames_total": frames_total, "frames_dropped": frames_dropped, "malformed_frames": malformed_frames, diff --git a/src/ui/credits.gd b/src/ui/credits.gd index 42e8915..9151228 100644 --- a/src/ui/credits.gd +++ b/src/ui/credits.gd @@ -32,7 +32,7 @@ const ENTRIES := [ ## 因此不能用单个常量表示；界面展示统一走 summary()。 const CONTRACT_VERSIONS := { "rest-api": "0.5.0", - "websocket-keypoints": "0.1.0", + "websocket-keypoints": "0.2.0", "data-models": "0.7.0", "skeleton-topology": "0.2.0", "model-formats": "0.5.0", diff --git a/src/ui/game_screen.gd b/src/ui/game_screen.gd index ed9b527..efb5d46 100644 --- a/src/ui/game_screen.gd +++ b/src/ui/game_screen.gd @@ -426,7 +426,8 @@ func _on_score_updated(result: Dictionary) -> void: func _refresh_connection_hint() -> void: - var count := PoseReceiver.connected_client_count() + # 契约 §4.1：按 deviceId 去重后的设备数（连接数可能因重连而短暂偏大） + var count := PoseReceiver.connected_device_count() if count > 0: _hint_label.text = "● 已连接 %d 台设备：%s" % [count, AppConfig.listen_url()] _hint_label.add_theme_color_override("font_color", UiKit.COLOR_GOOD) diff --git a/src/ui/main_menu.gd b/src/ui/main_menu.gd index 2445a27..8f46e7b 100644 --- a/src/ui/main_menu.gd +++ b/src/ui/main_menu.gd @@ -342,13 +342,28 @@ func _report_receiver() -> void: if snap.is_empty(): _set_status("接收服务未初始化") return - _set_status("接收服务：%s，端口 %d，连接 %d，累计帧 %d，丢帧 %d，畸形帧 %d" % [ + # 契约 §4.1：显示去重后的设备数，避免"反复重连"被计成多台设备 + var per_client := PackedStringArray() + for info in PoseReceiver.clients_info(): + per_client.append("#%d %s[%s] %d 帧/%d 丢" % [ + int(info["id"]), + str(info["device"]) if str(info["device"]) != "" else "未命名设备", + str(info["state"]), + int(info["frames"]), + int(info["dropped"]), + ]) + var detail := ("｜ " + "，".join(per_client)) if per_client.size() > 0 else "" + _set_status("接收服务：%s，端口 %d，连接 %d，设备 %d，累计帧 %d，丢帧 %d，畸形帧 %d｜静默超时 %dms 心跳 %dms%s" % [ "运行中" if bool(snap.get("running", false)) else "已停止", int(snap.get("port", 0)), int(snap.get("connected", 0)), + int(snap.get("devices", 0)), int(snap.get("frames_total", 0)), int(snap.get("frames_dropped", 0)), int(snap.get("malformed_frames", 0)), + int(snap.get("silent_timeout_ms", 0)), + int(snap.get("heartbeat_interval_ms", 0)), + detail, ]) diff --git a/tasks/015-keepalive-dedup.md b/tasks/015-keepalive-dedup.md new file mode 100644 index 0000000..8a0a002 --- /dev/null +++ b/tasks/015-keepalive-dedup.md @@ -0,0 +1,89 @@ +# T-WIN-015 连接保活、设备去重与状态机 + +- **状态**: done +- **依赖**: T-WIN-003 +- **文档**: v1.6 §6.5（T-AND-011 要求"两端一起改，协议与状态机对齐"） +- **契约**: `websocket-keypoints` 0.1.0 → **0.2.0** + +## 背景 + +Android 端反馈「连接状态异常、频繁重连」。按文档 §6.5 的要求，这**不是单端问题**， +必须两端一起改。核对本端后确认本端有 **3 处**是根因或放大器： + +### 根因 1：静默超时 5 秒过短，且客户端无从得知 + +`CLIENT_TIMEOUT_MS` 原为 **5000ms**，且**没有在 `welcome` 里声明**。 + +- 客户端只能用 OkHttp 的协议层 ping 兜底 —— 而协议层 ping 由底层库自动应答， + **本端应用层看不到**，`last_seen_ms` 不会因此刷新； +- 结果：画面里没有人（关键点帧稀疏）时，只要客户端那一秒的 `stats` 稍慢， + 5 秒一到就被本端 `close(1001, "timeout")` 踢掉 → 客户端重连 → 再被踢。 + +**修复**： + +- 超时提到 **15000ms**（关键点帧与每秒 `stats` 都算活动，15 秒足够容忍 + 相机暂停 / 画面无人 / 推理短暂卡顿，又不会让死连接长期占用）； +- 新增 `HEARTBEAT_INTERVAL_MS = 5000`； +- 两者通过 `welcome_payload()` 在 `welcome` 中**声明**，并在 `config` 广播中重复， + 客户端据此定节奏（契约 §1.1 要求心跳 < 静默超时的一半：5000×2 < 15000 ✓）； +- 新增 `ping` 处理：收到应用层 `ping` 回 `pong`（回显 `sentAtMs`，另带 `serverAtMs`）， + 客户端用 `now - sentAtMs` 估算 RTT。 + +### 根因 2：反复重连被计成多台设备 + +原先只按"连接数"显示，手机闪断后立刻重连会短暂出现"已连接 2 台"。 + +**修复（契约 §5）**： + +- `hello` 新增稳定 `deviceId`（Android 端取 `ANDROID_ID` 的 SHA-256 前 16 位， + **不含任何硬件序列号**）； +- `_dedup_devices()` 在每轮 `_poll_clients()` 之后运行：同一 `deviceId` 只保留 + **最新**连接，旧连接以 `close(1000, "replaced by new connection")` 关闭 —— + 用 `1000` 是刻意的：按契约 §7，对端收到 `1000` **不会**重连， + 否则会形成"新连接挤掉旧连接 → 旧连接重连 → 再挤掉"的循环； +- 新增 `device_count()`（去重后设备数）与 `device_replaced` 信号； + 界面与 `snapshot()` 一律显示去重后的数字。 + +### 根因 3：没有可对照的状态机与断开原因 + +原先只有一个 `handshaken` 布尔量，日志里看不出"卡在哪一步"。 + +**修复（契约 §8）**：逐客户端维护与 Android 端**同名**的状态 +（`ClientState` + `state_name()`），并记录 `close_code` / `close_reason` / +`selected_protocol` / `connected_at_ms` / `handshake_at_ms` / `first_frame_at_ms`： + +``` +CONNECTING --(HTTP 升级完成)--> HANDSHAKING --(收到 hello / 首帧)--> STREAMING + ^ | + | (出现丢帧) + +-------- DEGRADED <---------------+ +STOPPED <- 主动 stop() / 收到 bye / 1000 / 1001 / 静默超时 / 被同设备新连接替换 +``` + +`DEGRADED` 的保持时长 `degraded_hold_ms` 默认 3000ms，期间无新丢帧即回到 +`STREAMING`（做成实例字段，便于测试不依赖真实等待）。 + +## 改动清单 + +| 文件 | 改动 | +|--|--| +| `src/net/pose_server.gd` | 超时 5s→15s + 心跳常量；`welcome_payload()`；`ping`→`pong`；`_dedup_devices()`；`ClientState`/`state_name()`；`clients_info()`；`device_count()`；close code/reason 与各时间点记录 | +| `src/autoload/pose_receiver.gd` | `connected_device_count()` / `clients_info()` / `silent_timeout_ms()` / `heartbeat_interval_ms()` | +| `src/ui/game_screen.gd` | 「已连接 N 台」改用去重后的设备数 | +| `src/ui/main_menu.gd` | 「查看接收状态」增加设备数、保活参数与逐客户端状态 | +| `src/ui/credits.gd` | 契约版本字典同步到 `websocket-keypoints 0.2.0` | +| `contracts/` | 快照同步（与 opendance-web / opendance-android 逐字节一致） | + +## 验收（契约 §7 / §8 + 文档 §6.5） + +| 验收项 | 结果 | +|--|--| +| 服务端声明静默超时与期望心跳，心跳 < 超时的一半 | ✅ `test_welcome_declares_keepalive_parameters` 断言 `heartbeat*2 < timeout` | +| 应用层 ping 能续命 | ✅ `test_ping_gets_pong_with_echoed_timestamp`（协议层 ping 对应用层不可见，故必须有应用层 ping） | +| 同一设备重复连接被去重，旧连接不重连 | ✅ `test_duplicate_device_id_is_deduplicated`：旧连接被 `1000` 关闭、设备数仍为 1 | +| 状态名与两端一致 | ✅ `test_states_follow_contract_names` 逐项核对 8 个状态 | +| 状态随握手/首帧推进 | ✅ `test_client_state_progresses_to_streaming`：handshaking → streaming | +| 丢帧进入 degraded 后可恢复 | ✅ `test_frame_gap_marks_degraded_then_recovers` | +| snapshot 暴露设备数与保活参数 | ✅ `test_snapshot_exposes_keepalive_and_device_count` | + +Windows 端无头测试：**720 通过 / 0 失败**（其中 `test_pose_server.gd` 新增 7 项）。 diff --git a/tests/test_pose_server.gd b/tests/test_pose_server.gd index 426be72..e991609 100644 --- a/tests/test_pose_server.gd +++ b/tests/test_pose_server.gd @@ -35,6 +35,8 @@ func _setup() -> void: _server = PoseServer.new() _server.subprotocol = SUBPROTOCOL + # 缩短 degraded 保持时长：无头测试不能依赖真实等待 3 秒 + _server.degraded_hold_ms = 60 _tree().root.add_child(_server) _server.frame_received.connect(func(frame): _received.append(frame)) _server.control_received.connect(func(_id, msg): _controls.append(msg)) @@ -293,3 +295,190 @@ func test_stop_closes_listener() -> void: is_false(_server.is_running(), "stop() 后不再监听") eq(_server.connected_count(), 0, "stop() 后连接清空") await _teardown() + + +# ---------------- T-AND-011：契约 v0.2.0 的保活 / 去重 / 状态机 ---------------- + +## 连入第二个客户端（用于 deviceId 去重测试） +func _open_second_client() -> WebSocketPeer: + var peer := WebSocketPeer.new() + peer.supported_protocols = PackedStringArray([SUBPROTOCOL]) + peer.connect_to_url("ws://127.0.0.1:%d/pose" % TEST_PORT) + var guard := 0 + while peer.get_ready_state() == WebSocketPeer.STATE_CONNECTING and guard < 240: + peer.poll() + await _pump() + guard += 1 + return peer + + +func _send_hello(peer: WebSocketPeer, device_id: String, device: String = "测试机") -> void: + peer.send_text(JSON.stringify({ + "type": "hello", + "protocol": SUBPROTOCOL, + "deviceId": device_id, + "device": device, + "appVersion": "0.2.0", + "cameraFacing": "front", + "targetFps": 30, + "resolution": "640x480", + })) + + +## 等某个客户端收到指定类型的控制帧 +func _await_control(peer: WebSocketPeer, type: String, max_frames: int = 90) -> Dictionary: + for i in max_frames: + peer.poll() + await _pump() + while peer.get_available_packet_count() > 0: + var packet := peer.get_packet() + if not peer.was_string_packet(): + continue + var parsed: Variant = JSON.parse_string(packet.get_string_from_utf8()) + if typeof(parsed) == TYPE_DICTIONARY and str((parsed as Dictionary).get("type", "")) == type: + return parsed + return {} + + +func test_welcome_declares_keepalive_parameters() -> void: + await _setup() + await _pump(3) + _send_hello(_client, "device-keepalive") + + var welcome := await _await_control(_client, "welcome") + is_false(welcome.is_empty(), "收到 welcome 控制帧") + var timeout_ms := int(welcome.get("silentTimeoutMs", 0)) + var heartbeat_ms := int(welcome.get("heartbeatIntervalMs", 0)) + ok(timeout_ms > 0, "welcome 声明 silentTimeoutMs", str(timeout_ms)) + ok(heartbeat_ms > 0, "welcome 声明 heartbeatIntervalMs", str(heartbeat_ms)) + # 契约 §1.1 硬性要求：心跳间隔必须小于静默超时的一半 + ok(heartbeat_ms * 2 < timeout_ms, "心跳间隔小于静默超时的一半", + "心跳 %d ms，超时 %d ms" % [heartbeat_ms, timeout_ms]) + + await _teardown() + + +func test_ping_gets_pong_with_echoed_timestamp() -> void: + await _setup() + await _pump(3) + _send_hello(_client, "device-ping") + await _await_control(_client, "welcome") + + _client.send_text(JSON.stringify({"type": "ping", "sentAtMs": 1730000000123})) + var pong := await _await_control(_client, "pong") + is_false(pong.is_empty(), "收到 pong 控制帧（应用层心跳可续命）") + eq(int(pong.get("sentAtMs", 0)), 1730000000123, "pong 原样回传 sentAtMs") + ok(int(pong.get("serverAtMs", 0)) > 0, "pong 含服务端时间戳") + + await _teardown() + + +func test_duplicate_device_id_is_deduplicated() -> void: + await _setup() + await _pump(3) + _send_hello(_client, "device-dup", "第一台") + await _await_control(_client, "welcome") + await _pump(3) + eq(_server.device_count(), 1, "去重前设备数为 1") + + var second := await _open_second_client() + eq(second.get_ready_state(), WebSocketPeer.STATE_OPEN, "第二台客户端握手成功") + _send_hello(second, "device-dup", "第一台（重连）") + await _await_control(second, "welcome") + + # 服务端应在若干帧内关掉旧连接，并以 1000 关闭（对端不重连） + var guard := 0 + while guard < 120 and _client.get_ready_state() != WebSocketPeer.STATE_CLOSED: + _client.poll() + await _pump() + guard += 1 + + eq(_client.get_ready_state(), WebSocketPeer.STATE_CLOSED, "同一 deviceId 的旧连接被关闭") + eq(_client.get_close_code(), 1000, "旧连接以 1000 关闭（契约 §7：对端不重连）") + eq(_server.device_count(), 1, "去重后设备数仍为 1（不因重连变多）") + is_true(_server.connected_count() >= 1, "新连接仍在") + + second.close() + await _pump(2) + await _teardown() + + +func test_states_follow_contract_names() -> void: + eq(PoseServer.state_name(PoseServer.ClientState.IDLE), "idle", "idle 状态名") + eq(PoseServer.state_name(PoseServer.ClientState.RESOLVING), "resolving", "resolving 状态名") + eq(PoseServer.state_name(PoseServer.ClientState.CONNECTING), "connecting", "connecting 状态名") + eq(PoseServer.state_name(PoseServer.ClientState.HANDSHAKING), "handshaking", "handshaking 状态名") + eq(PoseServer.state_name(PoseServer.ClientState.STREAMING), "streaming", "streaming 状态名") + eq(PoseServer.state_name(PoseServer.ClientState.DEGRADED), "degraded", "degraded 状态名") + eq(PoseServer.state_name(PoseServer.ClientState.RECONNECTING), "reconnecting", "reconnecting 状态名") + eq(PoseServer.state_name(PoseServer.ClientState.STOPPED), "stopped", "stopped 状态名") + + +func test_client_state_progresses_to_streaming() -> void: + await _setup() + await _pump(3) + _send_hello(_client, "device-state") + await _await_control(_client, "welcome") + + var info_before: Array = _server.clients_info() + eq(info_before.size(), 1, "clients_info 含 1 个客户端") + eq(str(info_before[0]["state"]), "handshaking", "只发 hello 时为 handshaking") + + _client.send(KeypointFrame.encode(0, _sample_points(), _sample_confidence())) + var guard := 0 + while _received.is_empty() and guard < 90: + _client.poll() + await _pump() + guard += 1 + + var info: Array = _server.clients_info() + eq(str(info[0]["state"]), "streaming", "收到首帧后进入 streaming") + eq(str(info[0]["device_id"]), "device-state", "deviceId 被记录") + eq(str(info[0]["protocol"]), SUBPROTOCOL, "记录协商后的子协议") + ok(int(info[0]["first_frame_at_ms"]) > 0, "记录首帧时间（可诊断）") + + await _teardown() + + +func test_frame_gap_marks_degraded_then_recovers() -> void: + await _setup() + await _pump(3) + _send_hello(_client, "device-degraded") + await _await_control(_client, "welcome") + + _client.send(KeypointFrame.encode(0, _sample_points(), _sample_confidence())) + _client.send(KeypointFrame.encode(9, _sample_points(0.01), _sample_confidence())) + var guard := 0 + while _server.frames_dropped == 0 and guard < 120: + _client.poll() + await _pump() + guard += 1 + + eq(_server.frames_dropped, 8, "跳号 1~8 被计为丢包") + eq(str(_server.clients_info()[0]["state"]), "degraded", "出现丢帧时进入 degraded") + + # 无新丢帧持续 DEGRADED_HOLD_MS 后回到 streaming + var waited := 0 + while waited < 400 and str(_server.clients_info()[0]["state"]) != "streaming": + _client.poll() + await _pump() + waited += 1 + + eq(str(_server.clients_info()[0]["state"]), "streaming", "丢帧停止后回到 streaming") + + await _teardown() + + +func test_snapshot_exposes_keepalive_and_device_count() -> void: + await _setup() + await _pump(3) + _send_hello(_client, "device-snapshot") + await _await_control(_client, "welcome") + await _pump(3) + + var snap := _server.snapshot() + eq(int(snap["devices"]), 1, "snapshot 报告去重后设备数") + eq(int(snap["silent_timeout_ms"]), PoseServer.CLIENT_TIMEOUT_MS, "snapshot 报告静默超时") + eq(int(snap["heartbeat_interval_ms"]), PoseServer.HEARTBEAT_INTERVAL_MS, "snapshot 报告心跳间隔") + + await _teardown()