diff --git a/devtools/fake-backend/README.md b/devtools/fake-backend/README.md index 9892a12..73e4b6c 100644 --- a/devtools/fake-backend/README.md +++ b/devtools/fake-backend/README.md @@ -38,6 +38,7 @@ python fake_backend.py --port 18080 | `--stream-chunk` | 关 | 回复走旧形状 `stream_chunk` + `stream_end`（验客户端的回退分支） | | `--auto-reply` | 关 | 收到上行 `chat` 帧后自动回一轮（默认关，保持确定性） | | `--interrupt-ack` | 关 | 收到上行 `interrupt` 帧后回 `session_interrupted{mode:"idle",dropped:0}` | +| `--hold-replies` | 关 | 上行消息**只记录、不落库**：收到 `chat` / `POST /api/v1/chat` 后不广播 `message_persisted`，由 `POST /__control/persist` 显式补落库（见第 6 节） | | `--out-dir DIR` | `<脚本目录>/out` | 记录目录 | ## 2. 客户端怎么指过来 @@ -133,7 +134,88 @@ python ws_probe.py --port 18080 --listen 4 想临时切回结构化形状，不改启动参数也行：`…/__control/reply?session=web&text=hi&frame=structured`。 -## 6. 其它可编排场景（都试过） +## 6. 场景四：忙时被拒 —— 让「待发气泡还挂着」可复现（`--hold-replies`） + +**要验什么**：客户端手里**还挂着**一条没发出去的气泡时，收到「队列满 / 忙时被拒」会怎样。 +判据是客户端的纯函数 `queueFullOutcome`（`ChatViewModel.kt:240`），三档： + +- 本会话待发 **0** 条 ⇒ 刻意**什么都不说**（没有气泡会挂住，不能凭空弹提示）； +- 恰好 **1** 条 ⇒ **回滚**：气泡摘掉、正文 / 图片 / 附件退回输入栏，另加一条横幅； +- **≥2** 条 ⇒ **不猜**：一个气泡都不动，横幅只说「可能没发出去」（`QUEUE_FULL_UNKNOWN_MESSAGE`）。 + +**为什么需要这个开关**：默认行为下，客户端一发上行 `chat`（或 `POST /api/v1/chat`），假后端**立刻** +把那条用户消息落库并广播 `message_persisted`；客户端拿 `clientMsgId` 一比就把乐观气泡收走 +（`consumePendingFor`，`ChatViewModel.kt:116`）。**一个往返之后待发列表就空了** —— 于是「服务端把这条 +拒掉」这类场景永远凑不出来：手里根本没有气泡可回滚，判定只会落到「0 条 ⇒ 静默」那一档。 +真后端忙时不会立刻落库（先排队或先拒绝），所以这是假后端的**建模缺陷**，这个开关就是修它。 + +**它做什么**：`--hold-replies` 打开后，收到上行消息**只记录、不落库、不广播** `message_persisted`： + +- 上行帧照常进 `out/upstream.jsonl`，`/__control/state` 的 `wsUpstreamByType` 照常计数（判据来源不变）； +- **回复路径不受影响**：`--stream-chunk` / `--auto-reply` / `/__control/reply` 自己的 `persist=` + 该怎么走还怎么走 —— 那是另一个开关，关的是「上行那条的落库」这一件事； +- 那条消息挂进 `state.heldMessages`（带 `holdId` / 先占的 `messageId` / `clientMsgId` / `heldAt`）， + 由 `POST /__control/persist` **显式**补上落库那一步。 + +于是场景可以分两段做：「先把气泡挂住」→「再落库，或者干脆发一帧拒绝」。运行期也能开关： +`POST /__control/hold-replies?on=1|0`（等价于启动参数，改完立刻生效）。 + +```bash +B=http://127.0.0.1:18099 + +# 1) 起服务（换个端口，方便和默认那一档并排） +python fake_backend.py --port 18099 --hold-replies + +# 2) 发一条 chat：它会被挂住 —— 「挂住」的判据是 2 秒里一条回帧都没有 +python ws_probe.py --port 18099 --listen 2 \ + --send '{"type":"chat","data":{"sessionId":"web","clientMsgId":"local-1","text":"这条先挂着"}}' +curl -s "$B/__control/state" # heldMessages 里有它；wsUpstreamByType.chat=1 +grep -c message_persisted out/downstream.jsonl # 0：一条都没广播 + +# 3) 显式补落库 —— 这时才广播 message_persisted +# 先在一个终端起探针（它监听 5 秒）： +python ws_probe.py --port 18099 --ping --listen 5 +# 再在另一个终端补落库（回执里的 delivered 应当 >= 1）： +curl -s -X POST "$B/__control/persist?id=local-1" + +# 4) 忙时被拒：再发一条（local-2）挂着，然后注入 pending_full —— 这是「1 条待发」那一档 +# 还是分两个终端：探针先发 chat 再监听 6 秒，把那段时间留给另一个终端注入 +python ws_probe.py --port 18099 --listen 6 \ + --send '{"type":"chat","data":{"sessionId":"web","clientMsgId":"local-2","text":"这条会被拒"}}' +# 另一个终端注入（探针还在监听时执行，回执里的 delivered 应当是 1）： +curl -s -X POST "$B/__control/pending-full?session=web&mode=thinking&max=5&clientMsgId=local-2" +# 探针那边就能看到那帧原样到达： +# <<< frame 1 [after-send] {"type":"pending_full", … + +# 客户端期望：气泡回滚、正文回到输入栏、横幅是 QUEUE_FULL_MESSAGE +# 「她正忙且队列已满，这条没发出去，请稍后手动重试」（ChatViewModel.kt:138） +``` + +**在 App 上跑同一条路**：带 `--hold-replies` 起服务、客户端指过来，在聊天页发一条，然后 +`curl -s -X POST "$B/__control/pending-full?session=web&mode=thinking&max=5"`（**不带** `clientMsgId`， +走的就是「1 条待发」那一档：本会话只有这一条在飞，判定无歧义）。想让帧**精确**指向那条气泡，就把键 +填成客户端铸的 `localId` —— 它就在 `out/upstream.jsonl` 那条 chat 帧上（`data.clientMsgId`）。 + +帧上带不带那两个可选键，客户端的处置完全不同（与 `queueFullOutcome` 的判定顺序一一对应）： + +| 帧上的键 | 客户端看到的效果 | +|---|---| +| `&clientMsgId=<那条 localId>`，**不带** `messageId` | 命中那条气泡 ⇒ **回滚 + 横幅**（本场景要验的就是它） | +| `&clientMsgId=<那条 localId>&messageId=m-1` | **收编**：只摘掉气泡，正文**不**退回输入栏（「其实早就进去了」） | +| 不带 `clientMsgId`，本会话 1 条待发 | 就是它 ⇒ 回滚 + 横幅 | +| 不带 `clientMsgId`，本会话 ≥2 条待发 | **不猜**：一个气泡都不动，横幅只说「可能没发出去」 | +| 不带 `clientMsgId`，本会话 0 条待发 | 什么都不说（刻意静默） | + +定向那一帧同理，与广播帧是**同一个 code、同一句文案**： + +```bash +curl -s -X POST "$B/__control/error?session=web&code=chat_busy&clientMsgId=local-2" +``` + +同一件事被两条帧各报一次（广播 `pending_full` + 定向 `error{chat_busy}`）时，第二次到达气泡早已摘走， +客户端不会重复归还（`QueueFullOutcomeTest` 钉住了这条）。 + +## 7. 其它可编排场景（都试过） ```bash B=http://127.0.0.1:18080 @@ -143,11 +225,17 @@ curl -s -X POST "$B/__control/forget?session=web:5eed000000000000000000000000000 curl -s "$B/api/v1/sessions/web:5eed0000000000000000000000000001/messages?limit=1" # 队列满：广播 pending_full（可选带 clientMsgId / messageId，两个键都能缺席） +# 忙时被拒的完整打法（先让气泡挂着再拒）见第 6 节 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" +# 上行落库挂起 / 恢复，以及显式补落库（第 6 节那两个端点） +curl -s -X POST "$B/__control/hold-replies?on=1" +curl -s -X POST "$B/__control/hold-replies?on=0" +curl -s -X POST "$B/__control/persist?session=web&id=local-zz" + # 忙碌 / 空闲 / 已打断 curl -s -X POST "$B/__control/busy?session=web&mode=executing" curl -s -X POST "$B/__control/idle?session=web&mode=idle" @@ -183,7 +271,7 @@ curl -s "$B/__control/requests?limit=20" > Git Bash 的坑：命令行里的中文会被转成本地代码页（GBK）再交给子进程，服务端按 UTF-8 解 ⇒ 400 `invalid json body`。 > 所以**带中文的请求体写进文件再用 `--data-binary @file`**，控制端点的参数用 URL 编码（`%E4%BD%A0%E5%A5%BD`）。 -## 7. 端点速查 +## 8. 端点速查 `/api/v1/**`（一律包信封：成功 `{data}` / 失败 `{error:{code,message}}`，`ApiEnvelope.kt:8`）： @@ -211,11 +299,13 @@ curl -s "$B/__control/requests?limit=20" | `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/hold-replies?on=` | 运行期开关：上行消息要不要挂起（等价于 `--hold-replies`） | +| `POST /__control/persist?session=&id=&index=` | 把挂起的上行消息**显式**落库并广播 `message_persisted`；不给选择器就按 FIFO 落最旧那条 | | `POST /__control/ws-close?code=&reason=` | 踢掉所有客户端 WS（用来触发客户端的重连 / 补洞路径） | | `POST /__control/forget` / `create` / `clear` / `reset` | 会话级造场景 | | `GET /__control/state` / `upstream` / `requests` | 当前状态、上行帧、HTTP 请求流水 | -## 8. 记录文件（`out/`，每次启动追加，删掉即清空） +## 9. 记录文件（`out/`，每次启动追加，删掉即清空） - `out/upstream.jsonl` —— **每条上行帧一行**：`{ts, epochMs, client, type, sessionId, raw, frame}`。 `raw` 是客户端发来的**原样文本**。「客户端到底发没发那一帧」以这里为准。 @@ -223,7 +313,10 @@ curl -s "$B/__control/requests?limit=20" - `out/http.jsonl` —— 每条 HTTP 请求：`{method, path, query, headers}`。 `Authorization` 只留 sha256 指纹、WS 的 `?token=` 同样只留指纹 —— 就算它是假 token，也没理由把凭据原文写进可能被 commit 的文件。 -## 9. 形状出处（它照谁回话） +挂起的上行消息（`--hold-replies`）**不进文件** —— 它只活在内存里，看 `GET /__control/state` 的 +`heldMessages`。`out/*.jsonl` 仍照常记「上行帧收到了什么 / 下发帧广播了什么」。 + +## 10. 形状出处（它照谁回话） | 实现 | 依据 | |---|---| @@ -246,7 +339,7 @@ curl -s "$B/__control/requests?limit=20" | `interrupt{scope}` 取值与「只取消在跑回合」 | `WsOutbound.kt:59-78` + `InterruptScope.kt:27` | | `stream_chunk` + `stream_end` 回退形状 | `WsEvent.kt:205-228` + `ChatViewModel.kt:1451-1472` | -## 10. 已知不确定 / 没做 +## 11. 已知不确定 / 没做 - **`history` 帧壳**：我发了 `{type, sessionId:"web", seq, data:{messages}}`。契约说 v0.36 起每个结构化流帧的帧壳 都带 `sessionId`（`WsEvent.kt:21-27`），但客户端**对 `history` 不读帧壳**（`ChatViewModel.kt:1410-1412` @@ -258,5 +351,16 @@ curl -s "$B/__control/requests?limit=20" `observability_*` 推送、`subagent_ask` / `task_output_query`。上行收到这些帧**只记录、不处置**。因此 App 里 管理员页面、附件上传这类路径在这个假后端上跑不通。 - **`message_persisted` 与流帧的先后顺序**是我按「正文流完再落库」安排的，没有对着真后端核过。 +- **`--hold-replies` 的三个代价**（都刻意保留，因为改掉就不再是「只关那一件事」了）： + 1. `POST /api/v1/chat` 的回执形状**逐字不变**（仍 `{stream:true,queued:false,queueDepth:0,messageId,sessionId}`）， + 而那条消息此刻**并没有进库** —— 这个回执是「假后端在说已受理」的近似；想如实区分，看 + `/__control/state` 的 `heldMessages`。 + 2. 挂起期间**不认领**幂等键（与「队列满的判定发生在认领之前」一致）⇒ 同一 `clientMsgId` 重发会得到 + **第二条**挂起记录，而不是 `chat_duplicate`。 + 3. `createdAt` 在**补落库那一刻**才盖 ⇒ 若期间库里已经有更晚的消息，这条会以「晚到」的方式插进去， + 而 WS `history` 用的是**列表序**（不做排序）⇒ 这时的 `history` 顺序会与 REST 按 `createdAt` 的 + 排序不同（默认模式下两者本来就一致，所以只有挂起场景看得到这个差）。 +- 挂起的消息**只活在内存里**（进程一停就没了）；`/__control/reset` 会清空挂起列表，但**不动**模式开关 + （`holdReplies` 是模式，不是库里的状态）。 - **不做 TLS**（只有 `http://`），也不支持 WebSocket 扩展 / permessage-deflate。 - 控制端点的响应形状（`{"data":{...}}` 里的键）**是自造的**，不要照它写客户端代码。 diff --git a/devtools/fake-backend/fake_backend.py b/devtools/fake-backend/fake_backend.py index 226f96c..a70416e 100644 --- a/devtools/fake-backend/fake_backend.py +++ b/devtools/fake-backend/fake_backend.py @@ -216,6 +216,10 @@ class Store: self.clients = [] self.upstream_records = [] self.seen_client_msg_ids = {} + # 上行落库模式（--hold-replies / POST /__control/hold-replies）：挂起的上行消息一条不落库。 + self.hold_replies = args.hold_replies + self.held_messages = [] + self.hold_seq = 0 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")) @@ -234,6 +238,9 @@ class Store: self.state_seq = 0 self.seen_client_msg_ids = {} self.upstream_records = [] + # 挂起的上行消息属于「库里的状态」，一起复位；hold_replies 是**模式**，复位不动它。 + self.held_messages = [] + self.hold_seq = 0 main = Session(MAIN_SESSION_ID, "main", SEED_MAIN_CREATED_AT) sub = Session( "web:5eed0000000000000000000000000001", @@ -268,17 +275,23 @@ class Store: mood=None, documents=None, created_at=None, + message_id=None, ) -> dict: - """HistoryMessageDto（MessageDto.kt:13）。可选键为空时**整键省略**，与后端 omitempty 一致。""" + """HistoryMessageDto（MessageDto.kt:13）。可选键为空时**整键省略**，与后端 omitempty 一致。 + + [message_id] 非空 = 用挂起时刻先占好的号（见 [hold_message]），不再动 `message_seq`。 + """ with self.lock: session = self.sessions.get(session_id) if session is None: raise KeyError(session_id) - self.message_seq += 1 + if message_id is None: + self.message_seq += 1 + message_id = "msg-{:06d}".format(self.message_seq) 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), + "id": message_id, "sessionId": session_id, "role": role, "content": content, @@ -297,6 +310,57 @@ class Store: session.messages.append(message) return message + def hold_message(self, session_id: str, role: str, content: str, client_msg_id=None, mode=None) -> dict: + """上行消息**挂起**：不落库、不广播 `message_persisted`（--hold-replies 打开时走这条）。 + + 为什么要这个模式：默认行为下客户端一发 `chat`，一个往返就收到那条用户消息的 + `message_persisted`，乐观气泡立刻被收走 ⇒ 「服务端把这条**拒掉**」这类场景根本凑不出来 + （手里没有还挂着的待发气泡）。真后端忙时也是先排队 / 先拒绝，不会立刻落库。 + + id 在这里**先占号**（与真正落库共用同一个 `message_seq`，不会撞号），`createdAt` 留到 + 真正落库那一刻再盖（[persist_held_message]）。幂等键**不在**这一刻认领 —— 与契约一致： + 队列满的判定发生在认领之前（WsEvent.kt:398），所以挂起期间重发同一帧会得到第二条挂起 + 记录，而不是 `chat_duplicate`。 + """ + with self.lock: + self.message_seq += 1 + self.hold_seq += 1 + record = { + "holdId": "hold-{:06d}".format(self.hold_seq), + "sessionId": session_id, + "messageId": "msg-{:06d}".format(self.message_seq), + "role": role, + "clientMsgId": client_msg_id, + "mode": mode, + "content": content, + "heldAt": iso_now(), + } + self.held_messages.append(record) + return record + + def persist_held_message(self, hold: dict) -> dict: + """把一条挂起的上行消息**显式**落库（控制端点 `POST /__control/persist`）。 + + 落库动作与默认路径逐字一致：进同一个会话、认领幂等键、由调用方广播 `message_persisted`。 + """ + with self.lock: + message = self.add_message( + hold["sessionId"], + hold["role"], + hold["content"], + client_msg_id=hold.get("clientMsgId"), + mode=hold.get("mode"), + message_id=hold["messageId"], + ) + if hold.get("clientMsgId"): + self.seen_client_msg_ids[hold["clientMsgId"]] = message["id"] + self.held_messages.remove(hold) + return message + + def held_for(self, session_id=None) -> list: + with self.lock: + return [h for h in self.held_messages if session_id is None or h["sessionId"] == session_id] + def history_replay(self, session_id: str, limit: int) -> list: """握手 history 帧的回放：**最近 N 条、升序**。N 与库里的条数之差就是那个洞。""" with self.lock: @@ -429,6 +493,8 @@ class Store: "streamChunk": self.args.stream_chunk, "autoReply": self.args.auto_reply, "interruptAck": self.args.interrupt_ack, + "holdReplies": self.hold_replies, + "heldMessages": [dict(h) for h in self.held_messages], "outDir": self.out_dir, "sessions": sessions, "wsClients": len(self.clients), @@ -643,6 +709,18 @@ def handle_upstream(store: Store, client, frame) -> None: ) return try: + if store.hold_replies: + # 挂起：**不**落库、**不**广播 message_persisted（这一处是默认路径里的第一个广播点）。 + # 上行帧已经被 record_upstream 记过（out/upstream.jsonl + wsUpstreamByType）， + # 回复路径照旧不受影响。 + hold = store.hold_message(session_id, "user", data.get("text", ""), client_msg_id=client_msg_id) + log( + "hold", + "chat 挂起 {} clientMsgId={}（等 /__control/persist）".format(hold["holdId"], client_msg_id), + ) + if store.args.auto_reply: + send_reply_round(store, session_id, "收到：{}".format(data.get("text", "")[:20])) + return message = store.add_message( session_id, "user", @@ -1224,6 +1302,34 @@ class Handler(BaseHTTPRequestHandler): if session_id not in self.store.sessions: self._fail(404, "session_not_found", "unknown session {}".format(session_id)) return + if self.store.hold_replies: + # 挂起：回执形状逐字不变（`ChatAckDto`），但那条消息**没有**落库、也**没有**广播 + # message_persisted（这是默认路径里的第二个广播点）。`messageId` 是挂起时先占的号， + # 它会在 /__control/persist 那一刻真的进库。 + hold = self.store.hold_message( + session_id, + "user", + body.get("text", ""), + client_msg_id=body.get("clientMsgId"), + mode=body.get("mode"), + ) + log( + "hold", + "rest chat 挂起 {} messageId={}（等 /__control/persist）".format(hold["holdId"], hold["messageId"]), + ) + 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": hold["messageId"], + "sessionId": session_id, + }, + status=202, + ) + return message = self.store.add_message( session_id, "user", @@ -1401,6 +1507,65 @@ class Handler(BaseHTTPRequestHandler): self._ok({"session": session_id, "mode": mode, "dropped": dropped, "delivered": sent}) return + if action == "hold-replies" and method in ("GET", "POST"): + # 运行期开关（等价于启动参数 --hold-replies），用来在同一个进程里做「先挂着 → 再落库」两段。 + on = self._param(query, "on", "1") not in ("0", "false", "no") + self.store.hold_replies = on + with self.store.lock: + held = len(self.store.held_messages) + log("ctl ", "hold-replies {} (held={})".format("on" if on else "off", held)) + self._ok({"holdReplies": on, "held": held}) + return + + if action == "persist" and method in ("GET", "POST"): + # 显式补上挂起那一刻没做的那次落库：进库 + 广播 message_persisted（与默认路径同一形状）。 + session_id = self._param(query, "session", None) + ident = self._param(query, "id", None) + index = self._int_param(query, "index", None) + candidates = self.store.held_for(session_id) + if ident: + candidates = [ + hold + for hold in candidates + if ident in (hold["holdId"], hold["messageId"], hold.get("clientMsgId")) + ] + elif index is not None: + if index < 1 or index > len(candidates): + self._fail(404, "not_found", "index {} out of range (1..{})".format(index, len(candidates))) + return + candidates = [candidates[index - 1]] + else: + candidates = candidates[:1] # 不给选择器就按 FIFO 落最旧的那条 + if not candidates: + self._fail( + 404, + "no_held_message", + "no held message matches (hold-replies={}, held={}); see GET /__control/state".format( + self.store.hold_replies, len(self.store.held_messages) + ), + ) + return + hold = candidates[0] + message = self.store.persist_held_message(hold) + sent = self.store.broadcast( + {"type": "message_persisted", "sessionId": message["sessionId"], "data": {"message": message}}, + "message_persisted", + ) + log("ctl ", "persisted {} as {} (delivered={})".format(hold["holdId"], message["id"], sent)) + self._ok( + { + "persisted": True, + "session": message["sessionId"], + "holdId": hold["holdId"], + "messageId": message["id"], + "clientMsgId": hold.get("clientMsgId"), + "delivered": sent, + "remaining": len(self.store.held_messages), + "message": message, + } + ) + 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") @@ -1571,6 +1736,13 @@ def build_parser() -> argparse.ArgumentParser: dest="interrupt_ack", help="收到上行 interrupt 帧后回一帧 session_interrupted{mode:idle,dropped:0}（默认关）", ) + parser.add_argument( + "--hold-replies", + action="store_true", + dest="hold_replies", + help="上行消息**只记录、不落库**：收到 chat / POST /api/v1/chat 后不广播 message_persisted，" + "由 POST /__control/persist 显式补落库（默认关，默认行为逐字不变）。用来复现「忙时被拒 / 排队 / 队列满」", + ) parser.add_argument( "--out-dir", default=os.path.join(HERE, "out"), @@ -1600,6 +1772,13 @@ def main(argv=None) -> int: 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( + "挂起（--hold-replies：等 POST /__control/persist 才落库）" + if args.hold_replies + else "立即（收到即落库 + message_persisted）" + ) + ) print(" 记录目录 {}".format(args.out_dir)) print(" 控制台 GET /__control/state") print("=" * 72)