merge: FB —— 可编程假后端测试台（devtools/fake-backend/，纯 Python 标准库），解锁 after 补洞 / 打断帧 / stream_chunk 三条端到端验证

2026-10-05 08:03:03 +08:00
commit 27d5680d21
@@ -0,0 +1,3 @@
 out/
 __pycache__/
 *.pyc
@@ -0,0 +1,262 @@
 # Cyrene 假后端（`devtools/fake-backend/`）
 一个**可编程的假后端**：用来把「需要特定服务端行为才验得到」的客户端行为变成可复现的测试。
 纯 Python 3 标准库，零第三方依赖，不需要 Gradle、不需要模拟器里的真后端。
 它**不是**产品的一部分，也**不是**后端行为的权威来源 —— 所有响应形状都是从**客户端代码**里读出来的
 （出处逐条列在下面「形状出处」一节）。它照我们读出来的形状回话，不代表真后端真的这么回。
 **它不做什么**（这条比它能做什么更重要）：
 - **不校验任何鉴权**。`Authorization` / `X-API-Version` 照收不误、一律放行；收收到什么头都会记进 `out/http.jsonl`。
 - **不落盘**成任何持久数据（只有 `out/*.jsonl` 的流水日志）。进程一停，消息全没了。
 - **不调模型**。回复正文是你在控制端点里给的字符串按字符切片切出来的，没有任何生成。
 - **不实现契约的全集**。只实现「启动到渲染」要用到的那一段（见端点表），其余一律 404。
 - **不要拿它当后端行为的事实来源**。要判断真后端怎么回，去看 `Cyrene_v2/Docs/api-client.md` 与真后端本身；
  这个工具的价值是「造场景 + 记录客户端发了什么」，不是「回答后端是什么」。
 ---
 ## 1. 起服务（Windows + Git Bash）
 ```bash
 cd /d/Project/Code/Uni/Cyrene-Android-FB/devtools/fake-backend
 python fake_backend.py --port 18080
 ```
 启动后会打印监听地址、回放条数、回复形状、记录目录，然后原地等连接（Ctrl-C 退出）。
 > 端口随便挑，**别用 10942** —— 那是真后端的端口，占用或干扰它会打断另一个会话的联调。
 常用开关：
 | 开关 | 默认 | 作用 |
 |---|---|---|
 | `--port N` | `18080` | 监听端口 |
 | `--host H` | `127.0.0.1` | 监听地址 |
 | `--history-replay N` | `50` | WS 握手 `history` 帧**只**回放最近 N 条（比库里的条数小就是那个洞） |
 | `--stream-chunk` | 关 | 回复走旧形状 `stream_chunk` + `stream_end`（验客户端的回退分支） |
 | `--auto-reply` | 关 | 收到上行 `chat` 帧后自动回一轮（默认关，保持确定性） |
 | `--interrupt-ack` | 关 | 收到上行 `interrupt` 帧后回 `session_interrupted{mode:"idle",dropped:0}` |
 | `--out-dir DIR` | `<脚本目录>/out` | 记录目录 |
 ## 2. 客户端怎么指过来
 ```bash
 adb reverse tcp:18080 tcp:18080
 ```
 然后在 App 的登录页（「服务器地址」一栏，`LoginViewModel.kt:57`）填：
 ```
 http://127.0.0.1:18080
 ```
 登录随便填（用户名 / 密码都不校验）。`adb reverse` 把模拟器内的 `127.0.0.1:18080` 转到 PC 上的假后端，
 所以 REST 与 `ws://…/ws/chat?token=…` 走的是同一个地址，App 侧不用做任何特殊配置。
 想从干净状态重来：设置里退出登录 / 清 App 数据，再打一次 `adb reverse`（它只跟着设备会话活）。
 ## 3. 场景一：`after` 正向补洞
 **要验什么**：WS 重连只回放最近 50 条，断线期间漏掉的那一段由 `GET /messages?after=` 补回来
 （客户端 `ChatViewModel.catchUp` → `MessageRepository.loadAfter`，契约 §1 行 6）。
 用真后端要灌 51 条消息、每条跑一次模型回合；这里灌 120 条是**免费且瞬时**的。
 ```bash
 # 1) 库里灌 120 条（不调模型，直接进内存）
 curl -s -X POST 'http://127.0.0.1:18080/__control/messages?session=web&count=120'
 # 2) 看握手到底回放了什么：只剩最近 50 条（msg-000071..msg-000120），前 70 条是洞
 python ws_probe.py --port 18080 --listen 1
 # 3) 断线前「本端最后看到的那条」——它才是客户端 gapCursor 的来源
 curl -s 'http://127.0.0.1:18080/__control/message?session=web&index=20'
 # 4) 用它的 createdAt 作 after 游标补洞：第 1 页给回来的正好是那段洞
 curl -s 'http://127.0.0.1:18080/api/v1/sessions/web/messages?limit=50&after=<上一步的 createdAt>'
 # 5) 用「本页最后一条的 createdAt」继续推进，直到返回空数组（客户端就是这么推进的）
 ```
 **在 App 上跑同一条路**（真正的端到端）：
 ```bash
 # App 已经连着假后端时，把它踢下线；客户端会重连，重连触发 catchUp
 curl -s -X POST 'http://127.0.0.1:18080/__control/ws-close?code=1001'
 # 客户端到底发了哪些 after 请求？看请求流水：
 grep -o '"after=[0-9]*"' devtools/fake-backend/out/http.jsonl
 ```
 注意客户端只在「曾经 Connected 过、然后断开、再连上」时才做这次正向补拉
 （`ChatViewModel.kt:634-646`），冷启动走的是首屏 REST，不动 `after`。
 ## 4. 场景二：打断帧（`interrupt`）
 **要验什么**：设置档位 = 「打断」且会话**正忙**时，客户端在发消息之前会先发
 `interrupt{scope:"current"}`（`InterruptScope.kt:27`、`ChatViewModel.kt:1058-1063`）。
 真后端不给你看上行帧，这里**看得到**，而且这是唯一的第一现场。
 ```bash
 # 1) 起服务（回执可选：开了之后打断会立刻回到 idle，方便看界面状态）
 python fake_backend.py --port 18080 --interrupt-ack
 # 2) 让主会话忙起来 —— 客户端收到 session_busy 才会认为「正忙」
 curl -s -X POST 'http://127.0.0.1:18080/__control/busy?session=web&mode=thinking'
 # 3) App 里：设置 → 打断档；然后在聊天页再发一条消息
 # 4) 证据（每条上行帧一行，含时间戳）
 tail -n 2 out/upstream.jsonl
 # 也可以用一个不依赖 App 的等价动作，证明记录链路本身是通的：
 python ws_probe.py --port 18080 --listen 2 \
  --send '{"type":"interrupt","data":{"sessionId":"web","scope":"current"}}'
 ```
 ## 5. 场景三：`stream_chunk` 回退路径
 **要验什么**：契约 §3.1 的「非流式回退路径」——整段正文只在 `stream_chunk` + `stream_end` 里，
 没有 `message.start/delta/end`。真后端可能永远不走那条路，这里可以**指定**走。
 ```bash
 # 1) 起服务，指定回退形状
 python fake_backend.py --port 18080 --stream-chunk
 # 2) 触发一轮回复（不需要在 App 里打字；text 用 URL 编码传中文）
 curl -s -X POST 'http://127.0.0.1:18080/__control/reply?session=web&text=hello&chunks=3'
 # 3) 想在连接上原样看到这几帧：
 python ws_probe.py --port 18080 --listen 4
 ```
 想临时切回结构化形状，不改启动参数也行：`…/__control/reply?session=web&text=hi&frame=structured`。
 ## 6. 其它可编排场景（都试过）
 ```bash
 B=http://127.0.0.1:18080
 # 会话在服务端没了 → 404 session_not_found（客户端据此置 sessionGone，ChatViewModel.kt:1808）
 curl -s -X POST "$B/__control/forget?session=web:5eed0000000000000000000000000001"
 curl -s "$B/api/v1/sessions/web:5eed0000000000000000000000000001/messages?limit=1"
 # 队列满：广播 pending_full（可选带 clientMsgId / messageId，两个键都能缺席）
 curl -s -X POST "$B/__control/pending-full?session=web&mode=thinking&max=5&clientMsgId=local-zz"
 # 定向错误帧（与 REST 的 409 chat_busy 同码）
 curl -s -X POST "$B/__control/error?session=web&code=chat_busy&clientMsgId=local-zz"
 # 忙碌 / 空闲 / 已打断
 curl -s -X POST "$B/__control/busy?session=web&mode=executing"
 curl -s -X POST "$B/__control/idle?session=web&mode=idle"
 curl -s -X POST "$B/__control/interrupted?session=web&mode=idle&dropped=0"
 # 任意下行帧（整帧原样广播，形状不对时客户端会安全忽略 —— 正好用来验「坏帧不崩」）
 curl -s -X POST "$B/__control/frame?json=%7B%22type%22%3A%22session_idle%22%2C%22sessionId%22%3A%22web%22%2C%22data%22%3A%7B%22mode%22%3A%22idle%22%7D%7D"
 # 编辑 / 删除单条消息（REST 回执 + 广播帧都会来）
 curl -s -X PATCH "$B/api/v1/messages/msg-000001" -H 'Content-Type: application/json' -d '{"content":"edited"}'
 curl -s -X DELETE "$B/api/v1/messages/msg-000002"
 # 已读游标（推进后 unreadCount 归零，并广播 session_read）
 curl -s -X POST "$B/api/v1/sessions/web/read" -H 'Content-Type: application/json' -d '{}'
 # 会话列表 / 改名 / 置顶 / 新建 / 清空 / 复位
 curl -s "$B/api/v1/sessions"
 curl -s -X PATCH "$B/api/v1/sessions/web" -H 'Content-Type: application/json' --data-binary @body.json
 curl -s -X POST "$B/__control/create?session=web:mine&type=sub&pinned=1"
 curl -s -X POST "$B/__control/clear?session=web"
 curl -s -X POST "$B/__control/reset"          # 会话/消息/上行帧记录回到种子状态（out/*.jsonl 不动）
 # 看状态 / 看上行帧 / 看 HTTP 请求流水
 curl -s "$B/__control/state"
 curl -s "$B/__control/upstream?limit=20"
 curl -s "$B/__control/requests?limit=20"
 ```
 > 控制端点响应里的 `delivered` = **这一刻**连着 WS 的客户端数。它是**广播**（没有离线队列、没有补发），
 > 所以「App 还没连上 / 已经断开」时是 `0` 而帧不会留到下次。想让 App 收到，先确认 `GET /__control/state`
 > 里的 `wsClients` 不为 0。
 > Git Bash 的坑：命令行里的中文会被转成本地代码页（GBK）再交给子进程，服务端按 UTF-8 解 ⇒ 400 `invalid json body`。
 > 所以**带中文的请求体写进文件再用 `--data-binary @file`**，控制端点的参数用 URL 编码（`%E4%BD%A0%E5%A5%BD`）。
 ## 7. 端点速查
 `/api/v1/**`（一律包信封：成功 `{data}` / 失败 `{error:{code,message}}`，`ApiEnvelope.kt:8`）：
 | 端点 | 备注 |
 |---|---|
 | `POST /api/v1/auth/login` `/refresh` `/logout` | 回假 token（`role:"owner"`，`expiresAt` 是 **Unix 秒**） |
 | `GET /api/v1/sessions` | 排序照真契约：`pinned DESC, type='main' DESC, created_at ASC`；`unreadCount` 按已读游标算 |
 | `POST /api/v1/sessions` | 可绑 `personaId`（未知角色 → 400）；支持**完全不带 body** |
 | `PATCH /api/v1/sessions/{id}` | 局部响应；`{}` → 400；`personaId` → 400；title trim 后空 → 400；>80 字 → 400 |
 | `DELETE /api/v1/sessions/{id}` | `web` → 403 |
 | `GET /api/v1/sessions/{id}/messages` | 支持 `limit`（默认 50 / 上限 200）/ `before`（倒序）/ `after`（**升序**）/ `q` / `platform`；`after×before`、`after×q` → 400 |
 | `POST /api/v1/sessions/{id}/read` | 游标只前进；成功后广播 `session_read` |
 | `GET /api/v1/meta/enums` | 载荷逐字取自客户端仓库的线上夹具 |
 | `GET /api/v1/personas` | 用户侧角色目录（不含 admin 专属 `thinking`） |
 | `PATCH` / `DELETE /api/v1/messages/{id}` | 编辑 / 删除单条，并广播 `message_edited` / `message_deleted` |
 | `POST /api/v1/chat` | **额外**实现的 REST 写路径（长文 / 图片 / 附件走这条），回 202 + `ChatAckDto` 形状 |
 | `GET /ws/chat?token=…` | 手写 WebSocket；握手后立刻发 `history`（回放条数可配） |
 `/__control/**`（同样包信封，但载荷是本工具自己的状态，不是产品契约）：
 | 端点 | 作用 |
 |---|---|
 | `POST /__control/messages?session=&count=&role=&text=&prefix=&platform=` | 免费灌 N 条（`text` 支持 `{n}` 占位） |
 | `GET /__control/message?session=&index=` | 取第 n 条（1 基），脚本里用来**精确**拿游标 |
 | `POST /__control/busy` / `idle` / `interrupted` | 下发 `session_busy` / `session_idle` / `session_interrupted` |
 | `POST /__control/reply?session=&text=&chunks=&frame=&persist=` | 下发一轮助手回复（`frame=default\|chunk\|structured`） |
 | `POST /__control/pending-full` / `error` / `frame` | 队列满、定向错误帧、任意帧 |
 | `POST /__control/ws-close?code=&reason=` | 踢掉所有客户端 WS（用来触发客户端的重连 / 补洞路径） |
 | `POST /__control/forget` / `create` / `clear` / `reset` | 会话级造场景 |
 | `GET /__control/state` / `upstream` / `requests` | 当前状态、上行帧、HTTP 请求流水 |
 ## 8. 记录文件（`out/`，每次启动追加，删掉即清空）
 - `out/upstream.jsonl` —— **每条上行帧一行**：`{ts, epochMs, client, type, sessionId, raw, frame}`。
  `raw` 是客户端发来的**原样文本**。「客户端到底发没发那一帧」以这里为准。
 - `out/downstream.jsonl` —— 每条下发帧：`{ts, to, origin, raw}`。`origin` 会写清是 `history`、`control:reply` 还是某次广播。
 - `out/http.jsonl` —— 每条 HTTP 请求：`{method, path, query, headers}`。
  `Authorization` 只留 sha256 指纹、WS 的 `?token=` 同样只留指纹 —— 就算它是假 token，也没理由把凭据原文写进可能被 commit 的文件。
 ## 9. 形状出处（它照谁回话）
 | 实现 | 依据 |
 |---|---|
 | 信封 `{data}` / `{error:{code,message}}` | `app/.../data/api/ApiEnvelope.kt:8,18` |
 | 鉴权头 `X-API-Version: 1` + `Authorization: Bearer …` | `app/.../data/api/ApiClientFactory.kt:115-118`（**只记录，不校验**） |
 | WS 地址 `GET /ws/chat?token=<jwt>` | `app/.../data/ws/ChatSocket.kt:287-295` |
 | `LoginResponse`（`expiresAt` 是 Unix **秒**、`refreshToken` 可缺） | `app/.../data/api/dto/LoginDto.kt:12-19` |
 | `SessionSummaryDto`（`title` / `lastMessage` / `personaId` 可整键缺席） | `app/.../data/api/dto/SessionDto.kt:16-50` |
 | 会话排序 `pinned DESC, main DESC, created_at ASC` | `SessionDto.kt:8-9` + 线上夹具 `SessionDtoRealPayloadTest.kt:36-38` |
 | `HistoryMessageDto` | `app/.../data/api/dto/MessageDto.kt:13-46` |
 | `after` 升序 / `before` 倒序 / `q` 与 `after` 互斥 | `app/.../data/api/SessionsApi.kt:71-101` + `MessageRepository.kt:79-101` |
 | limit 默认 50 / 上限 200 | `app/.../data/repository/MessageRepository.kt:149-150` |
 | `EnumTableDto` 的 6/5/9 张表与 `unknown:"display-raw"` | 线上夹具 `MetaDtoRealPayloadTest.kt:24`（逐字） |
 | `UserPersonaDto`（用户侧不含 `thinking`） | `PersonaDto.kt:16-26` + `PersonaDtoDecodingTest.kt:32` |
 | `404 {code:"session_not_found"}` | `ChatViewModel.kt:1802-1808` |
 | `session_read` 载荷只有 `readAt` | `WsEvent.kt:338-356` |
 | `message_edited` 的 `data` **就是**消息本体（没有 `message` 包一层） | `WsEvent.kt:676-680` |
 | `message_persisted` / `channel_persisted` 的 `data:{message}` | `WsEvent.kt:671-674` |
 | `chat_duplicate` / `pending_full` 的可选键整键缺席 | `WsEvent.kt:291-336, 798-809` |
 | `interrupt{scope}` 取值与「只取消在跑回合」 | `WsOutbound.kt:59-78` + `InterruptScope.kt:27` |
 | `stream_chunk` + `stream_end` 回退形状 | `WsEvent.kt:205-228` + `ChatViewModel.kt:1451-1472` |
 ## 10. 已知不确定 / 没做
 - **`history` 帧壳**：我发了 `{type, sessionId:"web", seq, data:{messages}}`。契约说 v0.36 起每个结构化流帧的帧壳
  都带 `sessionId`（`WsEvent.kt:21-27`），但客户端**对 `history` 不读帧壳**（`ChatViewModel.kt:1410-1412`
  明确绕过 `isForSession`，改用每条消息自己的 `sessionId`）。所以这一处就算和真后端不一致，也不影响本工具能验的东西。
 - **`platform` 过滤**是「相等即收」的近似，不是契约里那套 LIKE 转义语义。客户端当前所有调用点都传 `null`
  （`MessageRepository.kt:42,99`），所以这条路径没被真正走过。
 - **`q` 搜索**是 ASCII 大小写不敏感的子串匹配；`%` / `_` / `\` 不做通配（我根本不解析通配符）。
 - **不实现**：admin 面全部端点、`documents` / `upload`、`voice_input` 的落库、`mode_switch` 的语义、
  `observability_*` 推送、`subagent_ask` / `task_output_query`。上行收到这些帧**只记录、不处置**。因此 App 里
  管理员页面、附件上传这类路径在这个假后端上跑不通。
 - **`message_persisted` 与流帧的先后顺序**是我按「正文流完再落库」安排的，没有对着真后端核过。
 - **不做 TLS**（只有 `http://`），也不支持 WebSocket 扩展 / permessage-deflate。
 - 控制端点的响应形状（`{"data":{...}}` 里的键）**是自造的**，不要照它写客户端代码。
@@ -0,0 +1,215 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """最小 WebSocket 探针（标准库手写握手 + 帧收发），配合 fake_backend.py 用。
 用途：
 - 证明 `/ws/chat` 能握手（打印 101 与响应头），并且握手后能收到 `history` 帧；
 - 往连接上发任意上行帧（`--send`），再去 out/upstream.jsonl 里对账——这是「客户端到底发没发那一帧」的独立第二现场；
 - 监听 N 秒，把服务端下发的每一帧原样打出来。
    python ws_probe.py --port 18080 --listen 3
    python ws_probe.py --port 18080 --send '{"type":"interrupt","data":{"sessionId":"web","scope":"current"}}' --listen 2
 """
 import argparse
 import base64
 import hashlib
 import json
 import os
 import socket
 import struct
 import sys
 import time
 for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass
 WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
 def encode_text_frame(text: str, mask: bytes = b"\x37\xfa\x21\x3d") -> bytes:
    payload = text.encode("utf-8")
    header = bytearray([0x80 | 0x1])
    length = len(payload)
    if length < 126:
        header.append(0x80 | length)
    elif length < 65536:
        header.append(0x80 | 126)
        header += struct.pack(">H", length)
    else:
        header.append(0x80 | 127)
        header += struct.pack(">Q", length)
    masked = bytes(byte ^ mask[i % 4] for i, byte in enumerate(payload))
    return bytes(header) + mask + masked
 class RawReader:
    """裸 socket 上的定长读。
    刻意**不用** `socket.makefile()`：给一个 BufferedReader 设过超时之后它会永久变成
    「cannot read from timed out object」，于是「监听 2 秒」这种用法第二帧就读不到了。
    """
    def __init__(self, sock: socket.socket):
        self.sock = sock
        self.buffer = b""
    def read_exact(self, count: int):
        while len(self.buffer) < count:
            chunk = self.sock.recv(65536)
            if not chunk:
                return None
            self.buffer += chunk
        head, self.buffer = self.buffer[:count], self.buffer[count:]
        return head
    def read_line(self):
        while b"\r\n" not in self.buffer:
            chunk = self.sock.recv(65536)
            if not chunk:
                return None
            self.buffer += chunk
        line, self.buffer = self.buffer.split(b"\r\n", 1)
        return line
 class Probe:
    def __init__(self, host: str, port: int, path: str):
        self.sock = socket.create_connection((host, port), timeout=10)
        key = base64.b64encode(os.urandom(16)).decode("ascii")
        request = (
            "GET {} HTTP/1.1\r\n"
            "Host: {host}:{port}\r\n"
            "Upgrade: websocket\r\n"
            "Connection: Upgrade\r\n"
            "Sec-WebSocket-Key: {key}\r\n"
            "Sec-WebSocket-Version: 13\r\n"
            "\r\n"
        ).format(path, host=host, port=port, key=key)
        self.sock.sendall(request.encode("ascii"))
        self.reader = RawReader(self.sock)
        status = self.reader.read_line().decode("latin-1").strip()
        self.sock.settimeout(None)
        print("<<< " + status)
        headers = {}
        while True:
            line = self.reader.read_line().decode("latin-1")
            if line == "":
                break
            name, _, value = line.partition(":")
            headers[name.strip()] = value.strip()
            print("<<< " + line.strip())
        expected = base64.b64encode(hashlib.sha1((key + WS_GUID).encode("ascii")).digest()).decode("ascii")
        code = status.split()[1] if len(status.split()) > 1 else ""
        if code != "101":
            raise SystemExit("handshake failed: " + status)
        if headers.get("Sec-WebSocket-Accept") != expected:
            raise SystemExit("bad Sec-WebSocket-Accept: " + str(headers.get("Sec-WebSocket-Accept")))
        print("<<< handshake OK (101, Sec-WebSocket-Accept matches)")
        self.closed = False
    def send_json(self, payload: str) -> None:
        self.sock.sendall(encode_text_frame(payload))
        print(">>> " + payload)
    def send_ping(self, payload: bytes = b"probe-ping") -> None:
        header = bytearray([0x80 | 0x9])
        length = len(payload)
        if length < 126:
            header.append(0x80 | length)
        else:
            header.append(0x80 | 126)
            header += struct.pack(">H", length)
        mask = b"\x11\x22\x33\x44"
        masked = bytes(byte ^ mask[i % 4] for i, byte in enumerate(payload))
        self.sock.sendall(bytes(header) + mask + masked)
        print(">>> ping (payload={})".format(payload.decode("latin-1")))
    def _read_exact(self, count: int):
        return self.reader.read_exact(count)
    def listen(self, seconds: float, label: str = "downstream") -> int:
        deadline = time.time() + seconds
        frames = 0
        self.sock.settimeout(0.25)
        try:
            while time.time() < deadline:
                try:
                    head = self._read_exact(2)
                except socket.timeout:
                    continue
                if not head or len(head) < 2:
                    print("<<< connection closed by peer")
                    self.closed = True
                    break
                opcode = head[0] & 0x0F
                length = head[1] & 0x7F
                if length == 126:
                    length = struct.unpack(">H", self._read_exact(2))[0]
                elif length == 127:
                    length = struct.unpack(">Q", self._read_exact(8))[0]
                payload = self._read_exact(length) if length else b""
                if payload is None:
                    print("<<< connection closed by peer")
                    self.closed = True
                    break
                if opcode == 0x1:
                    frames += 1
                    print("<<< frame {} [{}] {}".format(frames, label, payload.decode("utf-8", "replace")))
                elif opcode == 0x8:
                    code = struct.unpack(">H", payload[:2])[0] if len(payload) >= 2 else 0
                    reason = payload[2:].decode("utf-8", "replace")
                    print("<<< close from server code={} reason={}".format(code, reason))
                    self.closed = True
                    break
                elif opcode == 0x9:
                    print("<<< ping")
                elif opcode == 0xA:
                    print("<<< pong")
        finally:
            self.sock.settimeout(None)
        if frames == 0:
            print("<<< （这 {} 秒里没有收到任何文本帧）".format(seconds))
        return frames
    def close(self) -> None:
        try:
            self.sock.sendall(bytes([0x80 | 0x8, 0x02]) + struct.pack(">H", 1000))
        except OSError:
            pass
        try:
            self.sock.close()
        except OSError:
            pass
 def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description="最小 WebSocket 探针")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=18080)
    parser.add_argument("--path", default="/ws/chat?token=probe-token")
    parser.add_argument("--send", action="append", default=[], help="连接建立后发送的帧（JSON 字符串，可重复）")
    parser.add_argument("--ping", action="store_true", help="发一个 ping 帧，验证服务端回 pong（OkHttp 30s 心跳走的就是它）")
    parser.add_argument("--listen", type=float, default=3.0, help="监听秒数（默认 3）")
    args = parser.parse_args(argv)
    probe = Probe(args.host, args.port, args.path)
    print("--- history 帧应当就在这里 ---")
    probe.listen(min(args.listen, 1.0), "initial")
    for payload in args.send:
        json.loads(payload)  # 先自己校验一次，免得把坏 JSON 发出去
        probe.send_json(payload)
    if args.ping:
        probe.send_ping()
    if args.send or args.ping:
        print("--- 上行帧已发出，等待服务端回帧 ---")
        probe.listen(args.listen, "after-send")
    probe.close()
    return 0
 if __name__ == "__main__":
    sys.exit(main())