From 86a22178588f2ae934be1f149ae2244ef35f3660 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 08:01:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(devtools):=20=E6=96=B0=E5=A2=9E=E5=8F=AF?= =?UTF-8?q?=E7=BC=96=E7=A8=8B=E5=81=87=E5=90=8E=E7=AB=AF=EF=BC=88devtools/?= =?UTF-8?q?fake-backend=EF=BC=89=EF=BC=8C=E7=94=A8=E6=9D=A5=E9=80=A0?= =?UTF-8?q?=E3=80=8C=E7=89=B9=E5=AE=9A=E6=9C=8D=E5=8A=A1=E7=AB=AF=E8=A1=8C?= =?UTF-8?q?=E4=B8=BA=E3=80=8D=E7=9A=84=E7=AB=AF=E5=88=B0=E7=AB=AF=E5=9C=BA?= =?UTF-8?q?=E6=99=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么要它 客户端有三类行为只在特定服务端行为下才验得到，而真后端不可编排、造场景的代价高到不可做： 1. after 正向增量补洞：要造「WS 握手只回放最近 50 条、库里却有 120 条」这个洞，用真后端得灌 51 条消息、 每条跑一次模型回合。假后端可以精确、免费、可复现地造出来（--history-replay N）。 2. 打断帧：客户端在「设置=打断 + 会话正忙」时会发 interrupt{scope:"current"}。要验它真的发出去了， 就得看上行帧——假后端把每条上行帧原样记进 out/upstream.jsonl，这是唯一的第一现场。 3. stream_chunk 回退路径：真后端可能永远不走那条路，假后端可以用 --stream-chunk 指定走。 加了什么 - devtools/fake-backend/fake_backend.py：纯 Python 3 标准库（无第三方依赖）。实现了客户端启动到渲染 用得到的那一段 REST（login/refresh/logout、sessions 列表与增删改、messages 分页与搜索、read、 messages 编辑删除、meta/enums、personas、chat）加一个手写 WebSocket（GET /ws/chat?token=…， 握手后立刻发 history 帧）。鉴权头 X-API-Version / Authorization 一律不校验，但会记进 out/http.jsonl （Authorization 与 ?token 只留 sha256 指纹）。 - 控制端点 /__control/**（不碰 /api/v1/**）：灌消息（不调模型）、下发 session_busy/idle/interrupted、 触发一轮回复、pending_full、定向 error、任意帧、踢掉客户端 WS、404 session_not_found、复位、看状态。 目的是让「造场景」与「看结果」都不需要人手点 UI。 - devtools/fake-backend/ws_probe.py：最小 WebSocket 探针（标准库手写握手 + 帧收发 + ping）， 用来证明握手返回 101 / 收到 history 帧，并往连接上发任意上行帧对账。 - devtools/fake-backend/README.md：起法（Windows + Git Bash）、客户端怎么指过来（adb reverse + http://127.0.0.1:18080）、每个场景的可复制命令序列、端点速查、形状出处表、以及「它不做什么」。 边界（README 里也写明了） 不校验鉴权、不落盘成持久数据、不调模型、不是产品的一部分、不实现契约全集（admin 面、附件上传、 observability 推送等一律 404），并且不要拿它当后端行为的事实来源——所有响应形状都是照客户端代码 读出来的（README 第 9 节逐条列出文件:行号），不代表真后端真的这么回。 默认端口 18080，刻意避开真后端的 10942。 形状来源（举几个关键例子）：信封 ApiEnvelope.kt:8；鉴权头 ApiClientFactory.kt:115-118； WS 地址 ChatSocket.kt:287-295；SessionSummaryDto 与排序 SessionDto.kt:16-50 加线上夹具 SessionDtoRealPayloadTest.kt:36-38；after 升序 / before 倒序 / q 与 after 互斥 SessionsApi.kt:71-101； limit 默认 50 上限 200 MessageRepository.kt:149-150；枚举表逐字取自 MetaDtoRealPayloadTest.kt:24； 404 session_not_found ChatViewModel.kt:1802-1808；stream_chunk 回退 WsEvent.kt:205-228。 验证（本机实跑，输出见提交后的报告） curl 打通 login/sessions/messages（limit/before/after/q 与三个 400 互斥面）、meta/enums、personas； ws_probe 握手拿到 101 与匹配的 Sec-WebSocket-Accept，并收到 history 帧；灌 120 条后握手只回放最近 50 条 （msg-000071..msg-000120），用 after 游标正向走两页正好补回 msg-000021..msg-000120、第三页为空； --stream-chunk 下下发的是 stream_chunk×3 + stream_end（没有 message.* 帧）；打断帧原样落进 out/upstream.jsonl；ping/pong 与 HTTP keep-alive 均正常。 --- devtools/fake-backend/.gitignore | 3 + devtools/fake-backend/README.md | 262 ++++ devtools/fake-backend/fake_backend.py | 1616 +++++++++++++++++++++++++ devtools/fake-backend/ws_probe.py | 215 ++++ 4 files changed, 2096 insertions(+) create mode 100644 devtools/fake-backend/.gitignore create mode 100644 devtools/fake-backend/README.md create mode 100644 devtools/fake-backend/fake_backend.py create mode 100644 devtools/fake-backend/ws_probe.py diff --git a/devtools/fake-backend/.gitignore b/devtools/fake-backend/.gitignore new file mode 100644 index 0000000..726b971 --- /dev/null +++ b/devtools/fake-backend/.gitignore @@ -0,0 +1,3 @@ +out/ +__pycache__/ +*.pyc diff --git a/devtools/fake-backend/README.md b/devtools/fake-backend/README.md new file mode 100644 index 0000000..9892a12 --- /dev/null +++ b/devtools/fake-backend/README.md @@ -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=` | `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":{...}}` 里的键）**是自造的**，不要照它写客户端代码。 diff --git a/devtools/fake-backend/fake_backend.py b/devtools/fake-backend/fake_backend.py new file mode 100644 index 0000000..226f96c --- /dev/null +++ b/devtools/fake-backend/fake_backend.py @@ -0,0 +1,1616 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""Cyrene 假后端（测试台）。纯标准库，零第三方依赖。 + +它的存在理由只有一个：有些客户端行为**只在特定服务端行为下**才验得到，而真后端的行为不可编排、 +造场景的代价又高到不可做（例如「WS 只回放最近 50 条、库里却有 120 条」这种洞，用真后端得灌 51 条 +消息、每条跑一次模型回合）。这里可以精确、免费、可复现地造出这些场景，并把**客户端到底发了什么**记下来。 + +它**不是**产品的一部分，也**不是**后端行为的权威来源：所有形状都是照客户端代码解析出来的（每个 +字段的出处都写在下面注释与 README 里），不代表真后端真的这么回。 + + python fake_backend.py --port 18080 + +端点一览见 README.md。控制端点一律挂在 `/__control/` 前缀下，不碰 `/api/v1/**`。 +""" + +import argparse +import base64 +import hashlib +import json +import os +import re +import struct +import sys +import threading +import time +import uuid +from datetime import datetime +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, unquote, urlsplit + +# 控制台在 Windows 的默认编码（cp936 / cp1252）会把中文和上游帧原样打印时炸掉； +# 这里固定成 UTF-8，让「贴出来的日志」与 out/*.jsonl 是同一份文本。 +for _stream in (sys.stdout, sys.stderr): + try: + _stream.reconfigure(encoding="utf-8", errors="replace") + except Exception: + pass + +HERE = os.path.dirname(os.path.abspath(__file__)) + +MAIN_SESSION_ID = "web" +WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + +MAX_PAGE_SIZE = 200 # 与 MessageRepository.MAX_PAGE_SIZE 一致 +DEFAULT_PAGE_SIZE = 50 # 与 MessageRepository.DEFAULT_PAGE_SIZE 一致 +TITLE_MAX_CHARS = 80 +MESSAGE_CONTENT_MAX_RUNES = 4000 + +# GET /api/v1/meta/enums 的载荷（逐字取自客户端仓库里「线上抓下来的」夹具： +# app/src/test/kotlin/top/yeij/cyrene/data/api/dto/MetaDtoRealPayloadTest.kt:24， +# 所以这张表的事实来源是客户端仓库，与任何正在运行的实例无关）。 +ENUMS_PAYLOAD = { + "platforms": [ + {"id": "web", "label": "网页"}, + {"id": "kook", "label": "KOOK"}, + {"id": "qq-obv11", "label": "QQ"}, + {"id": "weixin-ilink", "label": "微信"}, + {"id": "telegram", "label": "Telegram"}, + {"id": "webhook", "label": "Webhook"}, + ], + "moods": [ + {"id": "calm", "label": "平静"}, + {"id": "happy", "label": "开心"}, + {"id": "excited", "label": "兴奋"}, + {"id": "sad", "label": "低落"}, + {"id": "angry", "label": "生气"}, + ], + "modes": [ + {"id": "chat", "label": "对话"}, + {"id": "code", "label": "代码"}, + {"id": "home", "label": "家居"}, + {"id": "work", "label": "工作"}, + {"id": "docs", "label": "文档"}, + {"id": "voice", "label": "语音"}, + {"id": "vision", "label": "视觉"}, + {"id": "agents", "label": "智能体"}, + {"id": "settings", "label": "设置"}, + ], + "unknown": "display-raw", +} + +# GET /api/v1/personas（UserPersonaDto，PersonaDto.kt:16）。形状取自客户端解码测试的示例载荷 +# （PersonaDtoDecodingTest.kt:32），**不是**真后端的真实角色清单。 +# 刻意省掉夹具里的 avatar / chatBackground 路径：那些是服务端文件路径，假后端不提供图片， +# 带着它们只会让客户端去要一堆 404 的图。 +PERSONAS = [ + { + "id": "xilian", + "name": "昔涟", + "character": "memokeeper", + "isDefault": True, + "mode": "chat", + "memoryNamespace": "xilian", + "mood": "calm", + }, + { + "id": "qiqi", + "name": "七七", + "character": "poker", + "isDefault": False, + "mode": "code", + "memoryNamespace": "qiqi", + }, +] + +# 会话种子的 createdAt 取自客户端夹具里的真实值（SessionDtoRealPayloadTest.kt:37）， +# 排序规则也照它：pinned DESC, type='main' DESC, created_at ASC +SEED_MAIN_CREATED_AT = 1788724031196 +SEED_SUB_CREATED_AT = 1791092728237 + + +def now_ms() -> int: + return int(time.time() * 1000) + + +def iso_now() -> str: + return datetime.now().astimezone().isoformat(timespec="milliseconds") + + +def log(tag: str, message: str) -> None: + print("[{}] {:<4} {}".format(datetime.now().strftime("%H:%M:%S"), tag, message), flush=True) + + +def redact(value, keep: int = 8) -> str: + """凭据类字段只留指纹，绝不把 token 原样写进 out/ 下的文件。""" + if not value: + return "" + digest = hashlib.sha256(str(value).encode("utf-8")).hexdigest()[:keep] + return "".format(digest) + + +class JsonlLog: + """一行一条 JSON 的追加日志。每次写都开一次文件，崩了也不丢已写部分。""" + + def __init__(self, path: str): + self.path = path + self.lock = threading.Lock() + os.makedirs(os.path.dirname(path), exist_ok=True) + + def append(self, record: dict) -> None: + line = json.dumps(record, ensure_ascii=False) + with self.lock: + with open(self.path, "a", encoding="utf-8", newline="\n") as handle: + handle.write(line + "\n") + + +def envelope(data) -> dict: + """成功信封：ApiEnvelope.kt:8 —— 成功 `{data}`，失败 `{error:{code,message}}`。""" + return {"data": data} + + +def error_envelope(code: str, message: str) -> dict: + return {"error": {"code": code, "message": message}} + + +def collapse_whitespace(text: str) -> str: + return " ".join(text.split()) + + +def truncate60(text: str) -> str: + """服务端 lastMessage.content：折叠空白 + 截断到 60 字符 + 省略号（SessionDto.kt:83）。""" + collapsed = collapse_whitespace(text) + if len(collapsed) <= 60: + return collapsed + return collapsed[:60] + "…" + + +class Session: + def __init__(self, session_id: str, kind: str, created_at: int, title=None, pinned=False, persona_id=None): + self.id = session_id + self.type = kind + self.created_at = created_at + self.title = title + self.pinned = pinned + self.persona_id = persona_id + self.read_at = 0 + self.busy_mode = None + self.messages = [] + + def summary(self) -> dict: + """SessionSummaryDto（SessionDto.kt:16）。title / lastMessage / personaId 缺席即整键省略。""" + row = { + "id": self.id, + "type": self.type, + "createdAt": self.created_at, + "pinned": self.pinned, + "unreadCount": self.unread_count(), + } + if self.title is not None: + row["title"] = self.title + if self.persona_id: + row["personaId"] = self.persona_id + if self.messages: + last = self.messages[-1] + row["lastMessage"] = { + "role": last["role"], + "content": truncate60(last.get("content", "")), + "createdAt": last["createdAt"], + } + return row + + def unread_count(self) -> int: + return sum(1 for m in self.messages if m["createdAt"] > self.read_at) + + +class Store: + """全部状态都在内存里：不落盘成数据库、不调模型、重启即回到种子状态。""" + + def __init__(self, args): + self.args = args + self.lock = threading.RLock() + self.sessions = {} + self.message_seq = 0 + self.state_seq = 0 + self.clients = [] + self.upstream_records = [] + self.seen_client_msg_ids = {} + self.max_upstream_kept = args.keep_records + self.out_dir = args.out_dir + self.upstream_log = JsonlLog(os.path.join(args.out_dir, "upstream.jsonl")) + self.downstream_log = JsonlLog(os.path.join(args.out_dir, "downstream.jsonl")) + self.http_log = JsonlLog(os.path.join(args.out_dir, "http.jsonl")) + self.downstream_count = 0 + self.http_count = 0 + self.reset() + + # —— 种子 / 复位 —— + + def reset(self) -> None: + with self.lock: + self.sessions = {} + self.message_seq = 0 + self.state_seq = 0 + self.seen_client_msg_ids = {} + self.upstream_records = [] + main = Session(MAIN_SESSION_ID, "main", SEED_MAIN_CREATED_AT) + sub = Session( + "web:5eed0000000000000000000000000001", + "sub", + SEED_SUB_CREATED_AT, + pinned=True, + ) + self.sessions[main.id] = main + self.sessions[sub.id] = sub + + def ordered_sessions(self): + with self.lock: + rows = list(self.sessions.values()) + rows.sort(key=lambda s: (0 if s.pinned else 1, 0 if s.type == "main" else 1, s.created_at)) + return rows + + # —— 消息 —— + + def next_frame_seq(self) -> int: + with self.lock: + self.state_seq += 1 + return self.state_seq + + def add_message( + self, + session_id: str, + role: str, + content: str, + client_msg_id=None, + platform=None, + mode=None, + mood=None, + documents=None, + created_at=None, + ) -> dict: + """HistoryMessageDto（MessageDto.kt:13）。可选键为空时**整键省略**，与后端 omitempty 一致。""" + with self.lock: + session = self.sessions.get(session_id) + if session is None: + raise KeyError(session_id) + self.message_seq += 1 + previous = session.messages[-1]["createdAt"] if session.messages else 0 + stamp = created_at if created_at is not None else max(now_ms(), previous + 1, session.created_at + 1) + message = { + "id": "msg-{:06d}".format(self.message_seq), + "sessionId": session_id, + "role": role, + "content": content, + "createdAt": stamp, + } + if client_msg_id: + message["clientMsgId"] = client_msg_id + if platform: + message["platform"] = platform + if mode: + message["mode"] = mode + if mood: + message["mood"] = mood + if documents: + message["documents"] = documents + session.messages.append(message) + return message + + def history_replay(self, session_id: str, limit: int) -> list: + """握手 history 帧的回放：**最近 N 条、升序**。N 与库里的条数之差就是那个洞。""" + with self.lock: + session = self.sessions.get(session_id) + if session is None: + return [] + return [dict(m) for m in session.messages[-limit:]] if limit > 0 else [] + + def find_message(self, message_id: str): + with self.lock: + for session in self.sessions.values(): + for message in session.messages: + if message["id"] == message_id: + return session, message + return None, None + + # —— 客户端连接 —— + + def add_client(self, client) -> None: + with self.lock: + self.clients.append(client) + + def remove_client(self, client) -> None: + with self.lock: + if client in self.clients: + self.clients.remove(client) + + def broadcast(self, frame: dict, origin: str) -> int: + with self.lock: + targets = list(self.clients) + sent = 0 + for client in targets: + if client.send_frame(frame, origin): + sent += 1 + return sent + + # —— 记录 —— + + def record_upstream(self, client_label: str, raw: str, frame, parse_error=None) -> dict: + record = { + "ts": iso_now(), + "epochMs": now_ms(), + "client": client_label, + "type": (frame or {}).get("type") if isinstance(frame, dict) else None, + "sessionId": None, + "raw": raw, + } + if isinstance(frame, dict): + data = frame.get("data") + if isinstance(data, dict): + record["sessionId"] = data.get("sessionId") + record["frame"] = frame + if parse_error is not None: + record["parseError"] = parse_error + self.upstream_log.append(record) + with self.lock: + self.upstream_records.append(record) + if len(self.upstream_records) > self.max_upstream_kept: + self.upstream_records = self.upstream_records[-self.max_upstream_kept:] + # 终端上实时可见：这是「客户端到底发没发那一帧」的第一现场。 + log("WS<-", json.dumps(record.get("frame", raw), ensure_ascii=False)) + return record + + def record_downstream(self, client_label: str, origin: str, raw: str) -> None: + record = { + "ts": iso_now(), + "epochMs": now_ms(), + "to": client_label, + "origin": origin, + "raw": raw, + } + self.downstream_log.append(record) + with self.lock: + self.downstream_count += 1 + + def record_http(self, handler_label: str, method: str, path: str, query: dict, headers) -> dict: + """把收到的请求头**记下来**（鉴权头不校验，但要看得见）。 + + `Authorization` 只留指纹：那是客户端当前会话的凭据，即使假后端只看 HTTP 2xx， + 也没有理由把它原文写进一个可能被 commit 的文件。 + """ + recorded_headers = {} + for key, value in headers.items(): + if key.lower() == "authorization": + recorded_headers[key] = redact(value) + else: + recorded_headers[key] = value + if "token" in query: + query = dict(query) + query["token"] = [redact(v) for v in query["token"]] + record = { + "ts": iso_now(), + "epochMs": now_ms(), + "client": handler_label, + "method": method, + "path": path, + "query": query, + "headers": recorded_headers, + } + self.http_log.append(record) + with self.lock: + self.http_count += 1 + return record + + def state(self) -> dict: + with self.lock: + sessions = [ + { + "id": s.id, + "type": s.type, + "title": s.title, + "pinned": s.pinned, + "messages": len(s.messages), + "unreadCount": s.unread_count(), + "readAt": s.read_at, + "busy": s.busy_mode, + "oldestCreatedAt": s.messages[0]["createdAt"] if s.messages else None, + "newestCreatedAt": s.messages[-1]["createdAt"] if s.messages else None, + } + for s in self.sessions.values() + ] + counts = {} + for record in self.upstream_records: + key = record.get("type") or "" + counts[key] = counts.get(key, 0) + 1 + return { + "port": self.args.port, + "host": self.args.host, + "historyReplay": self.args.history_replay, + "streamChunk": self.args.stream_chunk, + "autoReply": self.args.auto_reply, + "interruptAck": self.args.interrupt_ack, + "outDir": self.out_dir, + "sessions": sessions, + "wsClients": len(self.clients), + "wsUpstreamFrames": len(self.upstream_records), + "wsUpstreamByType": counts, + "wsDownstreamFrames": self.downstream_count, + "httpRequests": self.http_count, + } + + +class WsConnection: + """手写的 WebSocket 连接（RFC 6455 的最小集：文本帧 + close + ping/pong）。 + + 只做本工具需要的事：收上行文本帧、下发 JSON 帧。不做扩展、不做压缩、不做分片发送 + （收方向支持分片拼接，因为客户端万一发大包会分片）。 + """ + + def __init__(self, store: Store, rfile, wfile, client_label: str): + self.store = store + self.rfile = rfile + self.wfile = wfile + self.label = client_label + self.write_lock = threading.Lock() + self.closed = False + + # —— 发送 —— + + def send_frame(self, frame: dict, origin: str) -> bool: + text = json.dumps(frame, ensure_ascii=False, separators=(",", ":")) + return self.send_text(text, origin) + + def send_text(self, text: str, origin: str) -> bool: + if self.closed: + return False + payload = text.encode("utf-8") + header = bytearray() + header.append(0x80 | 0x1) # FIN + text + length = len(payload) + if length < 126: + header.append(length) + elif length < 65536: + header.append(126) + header += struct.pack(">H", length) + else: + header.append(127) + header += struct.pack(">Q", length) + try: + with self.write_lock: + self.wfile.write(bytes(header) + payload) + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, OSError, ValueError): + self.closed = True + return False + self.store.record_downstream(self.label, origin, text) + return True + + def send_close(self, code: int = 1001, reason: str = "server closing") -> None: + if self.closed: + return + payload = struct.pack(">H", code) + reason.encode("utf-8") + header = bytearray([0x80 | 0x8, len(payload)]) + try: + with self.write_lock: + self.wfile.write(bytes(header) + payload) + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, OSError, ValueError): + pass + self.closed = True + + # —— 接收 —— + + def _read_exact(self, count: int): + try: + data = self.rfile.read(count) + except (ConnectionResetError, OSError, ValueError): + return None + if data is None or len(data) < count: + return None + return data + + def run(self) -> None: + fragment = bytearray() + fragment_opcode = None + while not self.closed: + head = self._read_exact(2) + if head is None: + break + fin = bool(head[0] & 0x80) + opcode = head[0] & 0x0F + masked = bool(head[1] & 0x80) + length = head[1] & 0x7F + if length == 126: + extra = self._read_exact(2) + if extra is None: + break + length = struct.unpack(">H", extra)[0] + elif length == 127: + extra = self._read_exact(8) + if extra is None: + break + length = struct.unpack(">Q", extra)[0] + mask = self._read_exact(4) if masked else None + if masked and mask is None: + break + payload = self._read_exact(length) if length else b"" + if payload is None: + break + if mask: + payload = bytes(byte ^ mask[i % 4] for i, byte in enumerate(payload)) + + if opcode == 0x8: # close + self.send_close(1000, "ack") + break + if opcode == 0x9: # ping → pong（OkHttp pingInterval 30s 靠它活着） + self._send_control(0xA, payload) + continue + if opcode == 0xA: # pong + continue + + if opcode == 0x0: # continuation + fragment += payload + if fragment_opcode is None: + continue + if not fin: + continue + self._on_text(bytes(fragment), fragment_opcode) + fragment = bytearray() + fragment_opcode = None + continue + + if not fin: + fragment = bytearray(payload) + fragment_opcode = opcode + continue + + if opcode == 0x1: + self._on_text(bytes(payload), opcode) + # 二进制帧（0x2）本工具不处理：客户端不会发。 + + self.closed = True + + def _send_control(self, opcode: int, payload: bytes) -> None: + header = bytearray([0x80 | opcode]) + length = len(payload) + if length < 126: + header.append(length) + elif length < 65536: + header.append(126) + header += struct.pack(">H", length) + else: + header.append(127) + header += struct.pack(">Q", length) + try: + with self.write_lock: + self.wfile.write(bytes(header) + payload) + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, OSError, ValueError): + self.closed = True + + def _on_text(self, payload: bytes, opcode: int) -> None: + try: + text = payload.decode("utf-8") + except UnicodeDecodeError: + self.store.record_upstream(self.label, payload.hex(), None, "invalid utf-8") + return + try: + frame = json.loads(text) + except json.JSONDecodeError as exc: + self.store.record_upstream(self.label, text, None, "bad json: {}".format(exc)) + return + self.store.record_upstream(self.label, text, frame if isinstance(frame, dict) else None) + handle_upstream(self.store, self, frame) + + +# ============ 上行帧的处置 ============ + +def handle_upstream(store: Store, client, frame) -> None: + if not isinstance(frame, dict): + return + kind = frame.get("type") + data = frame.get("data") if isinstance(frame.get("data"), dict) else {} + + if kind == "chat": + session_id = data.get("sessionId") or MAIN_SESSION_ID + client_msg_id = data.get("clientMsgId") + if client_msg_id and client_msg_id in store.seen_client_msg_ids: + # 幂等回执（WsEvent.kt:328 chat_duplicate）：不落库、不起回合。 + client.send_frame( + { + "type": "chat_duplicate", + "sessionId": session_id, + "data": { + "sessionId": session_id, + "clientMsgId": client_msg_id, + "messageId": store.seen_client_msg_ids[client_msg_id], + }, + }, + "chat_duplicate", + ) + return + if session_id not in store.sessions: + client.send_frame( + { + "type": "error", + "data": { + "code": "bad_request", + "message": "unknown session {}".format(session_id), + "sessionId": session_id, + }, + }, + "error", + ) + return + try: + message = store.add_message( + session_id, + "user", + data.get("text", ""), + client_msg_id=client_msg_id, + ) + except KeyError: + return + if client_msg_id: + store.seen_client_msg_ids[client_msg_id] = message["id"] + store.broadcast( + {"type": "message_persisted", "sessionId": session_id, "data": {"message": message}}, + "message_persisted", + ) + if store.args.auto_reply: + send_reply_round(store, session_id, "收到：{}".format(data.get("text", "")[:20])) + return + + if kind == "interrupt": + session_id = data.get("sessionId") or MAIN_SESSION_ID + if store.args.interrupt_ack: + with store.lock: + session = store.sessions.get(session_id) + if session is not None: + session.busy_mode = None + store.broadcast( + { + "type": "session_interrupted", + "sessionId": session_id, + "data": {"sessionId": session_id, "mode": "idle", "dropped": 0}, + }, + "interrupt_ack", + ) + return + # 其余上行帧（voice_input / mode_switch / subagent_ask / task_output_query）只记录，不处置。 + + +def split_chunks(text: str, count: int): + """把整段正文切成至多 [count] 段（尽量等长；文本比 count 短时就按字符切）。""" + count = max(1, min(count, len(text))) + if count <= 1: + return [text] + size, extra = divmod(len(text), count) + parts = [] + index = 0 + for position in range(count): + step = size + (1 if position < extra else 0) + parts.append(text[index : index + step]) + index += step + return [part for part in parts if part] + + +def send_reply_round(store: Store, session_id: str, text: str, chunks: int = 3, mode: str = "default", persist: bool = True): + """下发一轮助手回复。 + + [mode] `chunk` 走**旧形状** `stream_chunk` + `stream_end`（WsEvent.kt:206）， + 用来验客户端那条回退分支（ChatViewModel.kt:1468）；`structured` 走 message.start/delta/end。 + `default` 由 `--stream-chunk` 决定。 + """ + if mode == "default": + mode = "chunk" if store.args.stream_chunk else "structured" + turn_id = "turn-{:012x}".format(uuid.uuid4().int & 0xFFFFFFFFFFFF) + seq_base = store.next_frame_seq() + deltas = split_chunks(text, chunks) + + if mode == "chunk": + for offset, delta in enumerate(deltas): + store.broadcast( + { + "type": "stream_chunk", + "turnId": turn_id, + "seq": seq_base + offset, + "sessionId": session_id, + "data": {"delta": delta}, + }, + "reply:stream_chunk", + ) + else: + store.broadcast( + { + "type": "message.start", + "turnId": turn_id, + "seq": seq_base, + "sessionId": session_id, + "data": {"index": 0}, + }, + "reply:message.start", + ) + for offset, delta in enumerate(deltas): + store.broadcast( + { + "type": "message.delta", + "turnId": turn_id, + "seq": seq_base + offset + 1, + "sessionId": session_id, + "data": {"delta": delta, "index": 0}, + }, + "reply:message.delta", + ) + store.broadcast( + { + "type": "message.end", + "turnId": turn_id, + "seq": seq_base + len(deltas) + 1, + "sessionId": session_id, + "data": {"index": 0}, + }, + "reply:message.end", + ) + + store.broadcast( + { + "type": "stream_end", + "turnId": turn_id, + "seq": seq_base + len(deltas) + 2, + "sessionId": session_id, + "data": {"stopReason": "stop", "steps": 1}, + }, + "reply:stream_end", + ) + + if persist: + try: + message = store.add_message(session_id, "assistant", text, mode="chat") + except KeyError: + return {"turnId": turn_id, "mode": mode, "chunks": len(deltas), "persisted": False} + store.broadcast( + {"type": "message_persisted", "sessionId": session_id, "data": {"message": message}}, + "message_persisted", + ) + return {"turnId": turn_id, "mode": mode, "chunks": len(deltas), "persisted": True, "messageId": message["id"]} + return {"turnId": turn_id, "mode": mode, "chunks": len(deltas), "persisted": False} + + +# ============ HTTP ============ + +MESSAGES_RE = re.compile(r"^/api/v1/sessions/([^/]+)/messages$") +SESSION_RE = re.compile(r"^/api/v1/sessions/([^/]+)$") +SESSION_READ_RE = re.compile(r"^/api/v1/sessions/([^/]+)/read$") +MESSAGE_RE = re.compile(r"^/api/v1/messages/([^/]+)$") + + +class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + server_version = "CyreneFakeBackend" + store: Store = None + + # —— 基础设施 —— + + def log_message(self, fmt, *args): + return + + def _label(self) -> str: + host, port = self.client_address[0], self.client_address[1] + return "{}:{}".format(host, port) + + def _split(self): + parts = urlsplit(self.path) + query = parse_qs(parts.query, keep_blank_values=True) + return parts.path, query + + def _read_body(self): + length = self.headers.get("Content-Length") + if not length: + return b"" + try: + size = int(length) + except ValueError: + return b"" + if size <= 0: + return b"" + return self.rfile.read(size) or b"" + + def _parse_json_body(self): + raw = self._read_body() + if not raw: + return None, None + try: + return json.loads(raw.decode("utf-8")), raw + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + return None, str(exc) + + def _send_json(self, status: int, payload) -> None: + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + try: + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except (BrokenPipeError, ConnectionResetError, OSError, ValueError): + self.close_connection = True + + def _ok(self, data, status: int = 200) -> None: + self._send_json(status, envelope(data)) + + def _fail(self, status: int, code: str, message: str) -> None: + self._send_json(status, error_envelope(code, message)) + + # —— 方法入口 —— + + def do_GET(self): + self._handle("GET") + + def do_POST(self): + self._handle("POST") + + def do_PATCH(self): + self._handle("PATCH") + + def do_DELETE(self): + self._handle("DELETE") + + def do_OPTIONS(self): + self._handle("OPTIONS") + + def _handle(self, method: str): + path, query = self._split() + self.store.record_http(self._label(), method, path, query, self.headers) + try: + if path == "/ws/chat": + self._handle_ws(method, query) + return + if path.startswith("/__control/"): + self._handle_control(method, path, query) + return + if path.startswith("/api/v1/"): + self._handle_api(method, path, query) + return + if path == "/": + self._ok({"name": "cyrene-fake-backend", "control": "/__control/state"}) + return + self._fail(404, "not_found", "no such endpoint: {}".format(path)) + except BrokenPipeError: + self.close_connection = True + except Exception as exc: # 假后端绝不该把线程炸掉；把原因如实回给调用方 + log("ERR ", "{} {} -> {}".format(method, path, exc)) + try: + self._fail(500, "fake_backend_error", repr(exc)) + except Exception: + self.close_connection = True + + # —— WebSocket 握手 —— + + def _handle_ws(self, method: str, query) -> None: + if method != "GET": + self._fail(405, "bad_request", "websocket handshake must be GET") + return + upgrade = (self.headers.get("Upgrade") or "").lower() + key = self.headers.get("Sec-WebSocket-Key") + version = self.headers.get("Sec-WebSocket-Version") + if "websocket" not in upgrade or not key: + self._fail(400, "bad_request", "not a websocket handshake") + return + if version not in (None, "13"): + self._fail(400, "bad_request", "unsupported Sec-WebSocket-Version {}".format(version)) + return + + accept = base64.b64encode(hashlib.sha1((key + WS_GUID).encode("ascii")).digest()).decode("ascii") + self.send_response(101) + self.send_header("Upgrade", "websocket") + self.send_header("Connection", "Upgrade") + self.send_header("Sec-WebSocket-Accept", accept) + self.end_headers() + self.wfile.flush() + + token = (query.get("token") or [""])[0] + client = WsConnection(self.store, self.rfile, self.wfile, self._label()) + self.store.add_client(client) + log("WS ", "handshake 101 client={} token={}".format(self._label(), redact(token) or "")) + try: + # 握手后立刻发 history，回放条数由 --history-replay 决定（这就是那个洞）。 + replay = self.store.history_replay(MAIN_SESSION_ID, self.store.args.history_replay) + client.send_frame( + { + "type": "history", + "sessionId": MAIN_SESSION_ID, + "seq": self.store.next_frame_seq(), + "data": {"messages": replay}, + }, + "history", + ) + log( + "WS->", + "history replayed={} (historyReplay={})".format(len(replay), self.store.args.history_replay), + ) + client.run() + finally: + self.store.remove_client(client) + self.close_connection = True + log("WS ", "closed client={}".format(self._label())) + + # —— /api/v1/** —— + + def _handle_api(self, method: str, path: str, query) -> None: + # 鉴权：**一律不校验**（README 里明说）。token / X-API-Version 照收不误，已经记进 http.jsonl。 + if path == "/api/v1/auth/login" and method == "POST": + body, raw = self._parse_json_body() + self._ok( + { + "token": "fake-token-{}".format(uuid.uuid4().hex[:16]), + "expiresAt": int(time.time()) + 86400, # Unix **秒**（LoginDto.kt:14） + "role": "owner", + "refreshToken": "fake-refresh-{}".format(uuid.uuid4().hex[:16]), + } + ) + return + if path == "/api/v1/auth/refresh" and method == "POST": + self._ok( + { + "token": "fake-token-{}".format(uuid.uuid4().hex[:16]), + "expiresAt": int(time.time()) + 86400, + "role": "owner", + "refreshToken": "fake-refresh-{}".format(uuid.uuid4().hex[:16]), + } + ) + return + if path == "/api/v1/auth/logout" and method == "POST": + self._parse_json_body() + self._ok({"revoked": True, "revokedCount": 1}) + return + if path == "/api/v1/meta/enums" and method == "GET": + self._ok(ENUMS_PAYLOAD) + return + if path == "/api/v1/personas" and method == "GET": + self._ok(PERSONAS) + return + if path == "/api/v1/sessions": + if method == "GET": + self._ok([s.summary() for s in self.store.ordered_sessions()]) + return + if method == "POST": + self._create_session() + return + if path == "/api/v1/chat" and method == "POST": + self._rest_chat() + return + + match = MESSAGES_RE.match(path) + if match and method == "GET": + self._list_messages(unquote(match.group(1)), query) + return + + match = SESSION_READ_RE.match(path) + if match and method == "POST": + self._mark_read(unquote(match.group(1))) + return + + match = SESSION_RE.match(path) + if match: + session_id = unquote(match.group(1)) + if method == "PATCH": + self._update_session(session_id) + return + if method == "DELETE": + self._delete_session(session_id) + return + + match = MESSAGE_RE.match(path) + if match: + message_id = unquote(match.group(1)) + if method == "PATCH": + self._update_message(message_id) + return + if method == "DELETE": + self._delete_message(message_id) + return + + self._fail(404, "not_found", "no such endpoint: {} {}".format(method, path)) + + def _create_session(self) -> None: + body, raw = self._parse_json_body() + if raw is not None and body is None: + self._fail(400, "bad_request", "invalid json body") + return + persona_id = None + if isinstance(body, dict): + persona_id = body.get("personaId") + known = {p["id"] for p in PERSONAS} + if persona_id is not None and persona_id not in known: + self._fail(400, "bad_request", "unknown personaId: {}".format(persona_id)) + return + session_id = "web:{}".format(uuid.uuid4().hex) + with self.store.lock: + self.store.sessions[session_id] = Session( + session_id, + "sub", + now_ms(), + persona_id=persona_id, + ) + payload = {"sessionId": session_id} + if persona_id: + payload["personaId"] = persona_id + log("http", "created session {} persona={}".format(session_id, persona_id)) + self._ok(payload, status=201) + + def _update_session(self, session_id: str) -> None: + body, raw = self._parse_json_body() + if not isinstance(body, dict): + self._fail(400, "bad_request", "invalid json body") + return + if "personaId" in body: + # 会话角色只在创建时写入，之后不可变（SessionsApi.kt:46）。 + self._fail(400, "bad_request", "personaId is immutable") + return + if not body: + self._fail(400, "bad_request", "title or pinned is required") + return + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + self._fail(404, "not_found", "unknown session {}".format(session_id)) + return + response = {"id": session_id} + if "title" in body: + title = body.get("title") + if not isinstance(title, str) or not title.strip(): + self._fail(400, "bad_request", "title is required") + return + if len(title.strip()) > TITLE_MAX_CHARS: + self._fail(400, "bad_request", "title too long") + return + session.title = title.strip() + response["title"] = session.title + if "pinned" in body: + session.pinned = bool(body.get("pinned")) + response["pinned"] = session.pinned + self._ok(response) + + def _delete_session(self, session_id: str) -> None: + if session_id == MAIN_SESSION_ID: + self._fail(403, "forbidden", "main session cannot be deleted") + return + with self.store.lock: + if session_id not in self.store.sessions: + self._fail(404, "not_found", "unknown session {}".format(session_id)) + return + del self.store.sessions[session_id] + self._ok({"deleted": session_id}) + + def _list_messages(self, session_id: str, query) -> None: + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + # 契约 v0.63 的会话级 404（ChatViewModel.kt:1808 按这个码判「会话没了」）。 + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + messages = [dict(m) for m in session.messages] + + limit_raw = (query.get("limit") or [None])[0] + before_raw = (query.get("before") or [None])[0] + after_raw = (query.get("after") or [None])[0] + platform = (query.get("platform") or [None])[0] + search = (query.get("q") or [None])[0] + + limit = DEFAULT_PAGE_SIZE + if limit_raw not in (None, ""): + try: + limit = int(limit_raw) + except ValueError: + self._fail(400, "bad_request", "limit must be an integer") + return + limit = max(1, min(limit, MAX_PAGE_SIZE)) + + before = self._parse_cursor(before_raw, "before") + if before is _BAD: + return + after = self._parse_cursor(after_raw, "after") + if after is _BAD: + return + if after is not None and before is not None: + self._fail(400, "bad_request", "after and before are mutually exclusive") + return + if after is not None and search: + self._fail(400, "bad_request", "after and q are mutually exclusive") + return + + if platform: + messages = [m for m in messages if m.get("platform") == platform] + if search: + needle = search.lower() + messages = [m for m in messages if needle in (m.get("content") or "").lower()] + + if after is not None: + # 正向增量：created_at > after，**升序**（契约 §1 行 6）。 + page = [m for m in sorted(messages, key=lambda m: m["createdAt"]) if m["createdAt"] > after][:limit] + elif before is not None: + # 倒序分页：created_at < before，**最新在前**。 + page = [m for m in sorted(messages, key=lambda m: m["createdAt"], reverse=True) if m["createdAt"] < before][ + :limit + ] + else: + page = sorted(messages, key=lambda m: m["createdAt"], reverse=True)[:limit] + + self._ok(page) + + def _parse_cursor(self, raw, name: str): + if raw in (None, ""): + return None + try: + return int(raw) + except ValueError: + self._fail(400, "bad_request", "{} must be a unix-millis integer".format(name)) + return _BAD + + def _mark_read(self, session_id: str) -> None: + body, raw = self._parse_json_body() + if raw is not None and body is None: + self._fail(400, "bad_request", "invalid json body") + return + before = None + if isinstance(body, dict) and body.get("before") is not None: + candidate = body.get("before") + if not isinstance(candidate, int) or candidate <= 0: + self._fail(400, "bad_request", "before must be a positive unix-millis integer") + return + before = candidate + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + # 游标只前进（SQL MAX(read_at, before)），重复 / 乱序都幂等安全。 + session.read_at = max(session.read_at, before if before is not None else now_ms()) + read_at = session.read_at + # 契约 §3.9：成功后广播 session_read，载荷里**只有 readAt**，会话归属在帧壳上。 + self.store.broadcast( + {"type": "session_read", "sessionId": session_id, "data": {"readAt": read_at}}, + "session_read", + ) + self._ok({"sessionId": session_id, "readAt": read_at}) + + def _update_message(self, message_id: str) -> None: + body, raw = self._parse_json_body() + if not isinstance(body, dict): + self._fail(400, "bad_request", "invalid json body") + return + content = body.get("content") + if not isinstance(content, str) or not content.strip(): + self._fail(400, "bad_request", "content is required") + return + if len(content) > MESSAGE_CONTENT_MAX_RUNES: + self._fail(400, "bad_request", "content too long") + return + session, message = self.store.find_message(message_id) + if message is None: + self._fail(404, "not_found", "unknown message {}".format(message_id)) + return + message["content"] = content + log("http", "edited {} -> {}".format(message_id, json.dumps(content, ensure_ascii=False))) + # 契约 §3.8：data **就是**编辑后的完整消息 DTO，没有 message 包一层（WsEvent.kt:678）。 + self.store.broadcast( + {"type": "message_edited", "sessionId": session.id, "data": message}, + "message_edited", + ) + self._ok(message) + + def _delete_message(self, message_id: str) -> None: + with self.store.lock: + for session in self.store.sessions.values(): + for index, message in enumerate(session.messages): + if message["id"] == message_id: + del session.messages[index] + self.store.broadcast( + {"type": "message_deleted", "sessionId": session.id, "data": {"id": message_id}}, + "message_deleted", + ) + self._ok({"deleted": message_id}) + return + self._fail(404, "not_found", "unknown message {}".format(message_id)) + + def _rest_chat(self) -> None: + """`POST /api/v1/chat`（ChatApi.kt:21）——**额外**支持的写路径，形状照 ChatAckDto。""" + body, raw = self._parse_json_body() + if not isinstance(body, dict): + self._fail(400, "bad_request", "invalid json body") + return + session_id = body.get("sessionId") or MAIN_SESSION_ID + if session_id not in self.store.sessions: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + message = self.store.add_message( + session_id, + "user", + body.get("text", ""), + client_msg_id=body.get("clientMsgId"), + mode=body.get("mode"), + ) + self.store.broadcast( + {"type": "message_persisted", "sessionId": session_id, "data": {"message": message}}, + "message_persisted", + ) + if self.store.args.auto_reply: + send_reply_round(self.store, session_id, "收到：{}".format(body.get("text", "")[:20])) + self._ok( + { + "stream": True, + "queued": False, + "queueDepth": 0, + "messageId": message["id"], + "sessionId": session_id, + }, + status=202, + ) + + # —— /__control/** —— + + def _handle_control(self, method: str, path: str, query) -> None: + action = path[len("/__control/") :].strip("/") + + if action == "state" and method in ("GET", "POST"): + self._ok(self.store.state()) + return + + if action == "upstream" and method in ("GET", "POST"): + limit = self._int_param(query, "limit", 20) + with self.store.lock: + records = self.store.upstream_records[-limit:] + self._ok({"count": len(records), "records": records}) + return + + if action == "requests" and method in ("GET", "POST"): + limit = self._int_param(query, "limit", 20) + records = [] + if os.path.exists(self.store.http_log.path): + with open(self.store.http_log.path, "r", encoding="utf-8") as handle: + records = [json.loads(line) for line in handle if line.strip()] + self._ok({"count": len(records), "records": records[-limit:]}) + return + + if action == "reset" and method in ("GET", "POST"): + self.store.reset() + log("ctl ", "state reset to seed") + self._ok({"reset": True}) + return + + if action == "messages" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + count = self._int_param(query, "count", 1) + role = self._param(query, "role", "user") + prefix = self._param(query, "prefix", "seed") + template = self._param(query, "text", None) + platform = self._param(query, "platform", None) + if session_id not in self.store.sessions: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + first_id = last_id = None + for index in range(1, count + 1): + content = template.format(n=index) if template else "{} #{}".format(prefix, index) + message = self.store.add_message(session_id, role, content, platform=platform) + first_id = first_id or message["id"] + last_id = message["id"] + with self.store.lock: + session = self.store.sessions[session_id] + total = len(session.messages) + oldest = session.messages[0]["createdAt"] + newest = session.messages[-1]["createdAt"] + log("ctl ", "seeded {} msgs into {} (total={})".format(count, session_id, total)) + self._ok( + { + "session": session_id, + "added": count, + "total": total, + "firstId": first_id, + "lastId": last_id, + "oldestCreatedAt": oldest, + "newestCreatedAt": newest, + } + ) + return + + if action == "message" and method in ("GET", "POST"): + # 取某一条（1 基下标），用来在脚本里**精确**拿游标：例如「回放里最旧那条的前一条」。 + session_id = self._param(query, "session", MAIN_SESSION_ID) + index = self._int_param(query, "index", 1) + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + total = len(session.messages) + if index < 1 or index > total: + self._fail(404, "not_found", "index {} out of range (1..{})".format(index, total)) + return + message = dict(session.messages[index - 1]) + self._ok({"session": session_id, "index": index, "total": total, "message": message}) + return + + if action == "busy" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + mode = self._param(query, "mode", "thinking") + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + session.busy_mode = mode + sent = self.store.broadcast( + {"type": "session_busy", "sessionId": session_id, "data": {"sessionId": session_id, "mode": mode}}, + "control:busy", + ) + self._ok({"session": session_id, "mode": mode, "delivered": sent}) + return + + if action == "idle" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + mode = self._param(query, "mode", "idle") + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + session.busy_mode = None + sent = self.store.broadcast( + {"type": "session_idle", "sessionId": session_id, "data": {"sessionId": session_id, "mode": mode}}, + "control:idle", + ) + self._ok({"session": session_id, "mode": mode, "delivered": sent}) + return + + if action == "reply" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + text = self._param(query, "text", "这是一条来自假后端的回复。") + chunks = self._int_param(query, "chunks", 3) + mode = self._param(query, "frame", "default") + persist = self._param(query, "persist", "1") not in ("0", "false", "no") + if mode not in ("default", "chunk", "structured"): + self._fail(400, "bad_request", "frame must be default|chunk|structured") + return + if session_id not in self.store.sessions: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + result = send_reply_round(self.store, session_id, text, chunks=chunks, mode=mode, persist=persist) + result["delivered"] = len(self.store.clients) + self._ok(result) + return + + if action == "interrupted" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + mode = self._param(query, "mode", "idle") + dropped = self._int_param(query, "dropped", 0) + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + session.busy_mode = None + sent = self.store.broadcast( + { + "type": "session_interrupted", + "sessionId": session_id, + "data": {"sessionId": session_id, "mode": mode, "dropped": dropped}, + }, + "control:interrupted", + ) + self._ok({"session": session_id, "mode": mode, "dropped": dropped, "delivered": sent}) + return + + if action == "pending-full" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + mode = self._param(query, "mode", "thinking") + maximum = self._int_param(query, "max", 5) + client_msg_id = self._param(query, "clientMsgId", None) + message_id = self._param(query, "messageId", None) + data = {"sessionId": session_id, "mode": mode, "max": maximum} + # 两个可选键**整键缺席**，绝不填空串（WsEvent.kt:798）。 + if client_msg_id: + data["clientMsgId"] = client_msg_id + if message_id: + data["messageId"] = message_id + sent = self.store.broadcast( + {"type": "pending_full", "sessionId": session_id, "data": data}, + "control:pending_full", + ) + self._ok({"delivered": sent, "data": data}) + return + + if action == "error" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + code = self._param(query, "code", "chat_busy") + message = self._param(query, "message", "session queue is full") + client_msg_id = self._param(query, "clientMsgId", None) + message_id = self._param(query, "messageId", None) + data = {"code": code, "message": message, "sessionId": session_id} + if client_msg_id: + data["clientMsgId"] = client_msg_id + if message_id: + data["messageId"] = message_id + sent = self.store.broadcast({"type": "error", "data": data}, "control:error") + self._ok({"delivered": sent, "data": data}) + return + + if action == "frame" and method in ("GET", "POST"): + body, raw = self._parse_json_body() + payload = body + inline = self._param(query, "json", None) + if payload is None and inline: + try: + payload = json.loads(inline) + except json.JSONDecodeError as exc: + self._fail(400, "bad_request", "json param is not valid json: {}".format(exc)) + return + if payload is None and isinstance(raw, str): + self._fail(400, "bad_request", "body must be a json frame: {}".format(raw)) + return + if not isinstance(payload, dict) or not payload.get("type"): + self._fail(400, "bad_request", "a frame needs at least {\"type\": ...}") + return + sent = self.store.broadcast(payload, "control:frame") + self._ok({"delivered": sent, "frame": payload}) + return + + if action == "forget" and method in ("GET", "POST"): + session_id = self._param(query, "session", None) + if not session_id: + self._fail(400, "bad_request", "session is required") + return + with self.store.lock: + if session_id not in self.store.sessions: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + del self.store.sessions[session_id] + log("ctl ", "forgot session {} (its /messages now 404s)".format(session_id)) + self._ok({"forgot": session_id}) + return + + if action == "create" and method in ("GET", "POST"): + session_id = self._param(query, "session", None) + if not session_id: + self._fail(400, "bad_request", "session is required") + return + kind = self._param(query, "type", "sub") + title = self._param(query, "title", None) + pinned = self._param(query, "pinned", "0") not in ("0", "false", "no") + persona_id = self._param(query, "personaId", None) + with self.store.lock: + self.store.sessions[session_id] = Session( + session_id, + kind, + now_ms(), + title=title, + pinned=pinned, + persona_id=persona_id, + ) + self._ok({"session": session_id, "type": kind}) + return + + if action == "clear" and method in ("GET", "POST"): + session_id = self._param(query, "session", MAIN_SESSION_ID) + with self.store.lock: + session = self.store.sessions.get(session_id) + if session is None: + self._fail(404, "session_not_found", "unknown session {}".format(session_id)) + return + session.messages = [] + session.read_at = 0 + self._ok({"cleared": session_id}) + return + + if action == "ws-close" and method in ("GET", "POST"): + code = self._int_param(query, "code", 1001) + reason = self._param(query, "reason", "fake backend closed the socket") + with self.store.lock: + targets = list(self.store.clients) + for client in targets: + client.send_close(code, reason) + log("ctl ", "closed {} client sockets (code={})".format(len(targets), code)) + self._ok({"closed": len(targets), "code": code}) + return + + self._fail(404, "not_found", "no such control action: {}".format(action)) + + def _param(self, query, name, default): + values = query.get(name) + if not values: + return default + return values[0] + + def _int_param(self, query, name, default): + raw = self._param(query, name, None) + if raw in (None, ""): + return default + try: + return int(raw) + except ValueError: + return default + + +_BAD = object() # _parse_cursor 的失败哨兵 + + +class Server(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="fake_backend.py", + description="Cyrene 可编程假后端（开发用测试台，不是产品的一部分）", + ) + parser.add_argument("--host", default="127.0.0.1", help="监听地址（默认 127.0.0.1）") + parser.add_argument("--port", type=int, default=18080, help="监听端口（默认 18080；别用真后端的 10942）") + parser.add_argument( + "--history-replay", + type=int, + default=50, + dest="history_replay", + help="WS 握手 history 帧回放的条数（默认 50，与真后端一致）。比库里的条数小就是那个 after 洞", + ) + parser.add_argument( + "--stream-chunk", + action="store_true", + dest="stream_chunk", + help="回复走旧形状 stream_chunk + stream_end（用来验客户端那条回退分支）", + ) + parser.add_argument( + "--auto-reply", + action="store_true", + dest="auto_reply", + help="收到上行 chat 帧后自动回一轮（默认关：不自动回，保持确定性）", + ) + parser.add_argument( + "--interrupt-ack", + action="store_true", + dest="interrupt_ack", + help="收到上行 interrupt 帧后回一帧 session_interrupted{mode:idle,dropped:0}（默认关）", + ) + parser.add_argument( + "--out-dir", + default=os.path.join(HERE, "out"), + dest="out_dir", + help="记录目录（默认 <脚本目录>/out）：upstream.jsonl / downstream.jsonl / http.jsonl", + ) + parser.add_argument( + "--keep-records", + type=int, + default=1000, + dest="keep_records", + help="内存里保留的上行帧条数上限（默认 1000；文件里是全量）", + ) + return parser + + +def main(argv=None) -> int: + args = build_parser().parse_args(argv) + store = Store(args) + Handler.store = store + + server = Server((args.host, args.port), Handler) + host, port = server.server_address[0], server.server_address[1] + print("=" * 72) + print("Cyrene 假后端已启动 http://{}:{}".format(host, port)) + print(" WS ws://{}:{}/ws/chat?token=<任意>".format(host, port)) + print(" history 握手回放 {} 条（--history-replay）".format(args.history_replay)) + print(" 回复形状 {}".format("stream_chunk 旧形状" if args.stream_chunk else "message.* 结构化")) + print(" 自动回复 {} / interrupt 回执 {}".format(args.auto_reply, args.interrupt_ack)) + print(" 记录目录 {}".format(args.out_dir)) + print(" 控制台 GET /__control/state") + print("=" * 72) + try: + server.serve_forever() + except KeyboardInterrupt: + print("\nbye") + finally: + server.server_close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/devtools/fake-backend/ws_probe.py b/devtools/fake-backend/ws_probe.py new file mode 100644 index 0000000..8bf7d87 --- /dev/null +++ b/devtools/fake-backend/ws_probe.py @@ -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())