From 3b4a790e3047ebc39dbec3a30da61b7348a093f5 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 19:41:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(ws):=20=E9=98=9F=E5=88=97=E6=BB=A1?= =?UTF-8?q?=E6=97=B6=E5=AE=9A=E4=BD=8D=E5=B9=B6=E5=A4=84=E7=BD=AE=E9=82=A3?= =?UTF-8?q?=E6=9D=A1=E5=BE=85=E5=8F=91=E6=B0=94=E6=B3=A1=EF=BC=88=E5=A5=91?= =?UTF-8?q?=E7=BA=A6=20v0.59=20/=20=E5=8D=8F=E5=90=8C=20G5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WS 上行没有 REST 那样的同步回执：队列满时服务端只广播一条 pending_full，而载荷里既没有 「被拒的是哪一条」也没有任何可对齐的键，于是那条乐观气泡永远挂「发送中…」，用户不知道它 其实根本没发出去。后端 v0.59（G5）已经把这两件事补齐：pending_full 载荷新增两个可选键 clientMsgId（被拒那条消息的幂等键，未携带则整键缺席）与 messageId（仅当该消息此前已落库 / 已认领时才回填，服务端绝不在此新铸 id），并新增一条只发发送者的定向帧 error{code:"chat_busy"}，与 REST 的 409 chat_busy 同码。本端把这两条接上。 data/ws/WsEvent - PendingFull 增 clientMsgId / messageId：缺席就是 null，绝不拍成空串（否则收编路径会去 对齐一条不存在的消息）。 - Error 帧增 code / clientMsgId / messageId，并从 data.sessionId 取会话归属（定向帧的帧壳 上没有 sessionId）。code 是开放集合：未知码原样保留，不抛、也不吞掉整帧。 未给定向 error 另立事件类——它与既有 error 是同一个 type，一个 type 无法用两个类表达； 两者语义的分叉放在下面的判定函数里，保持「一个帧类型一个事件类」的既有约定。 - PendingFullData / ErrorData 同步扩键，全部可选且默认 null。 ui/chat/ChatViewModel - 新增纯函数 queueFullOutcome（判定）与 ChatUiState.applyQueueFull / rollbackPending （状态迁移），两者都不碰协程，逻辑不埋在协程里、可直接单测。既有 rollback 改为委托 rollbackPending，行为逐字不变。 - 判定顺序即证据强度：clientMsgId 命中时，messageId 非空 => 收编（复用 alignDuplicate：那条 其实早就落库了，不是失败、不回滚、不把正文/图片/附件退回输入栏），缺席 => 回滚那一条并把 内容还给输入栏；clientMsgId 有值但不命中 => 放手（键只属于某一次具体发送，回滚任何一条都是 误伤）；clientMsgId 缺席（老服务端）=> 沿用 toolExecutionTargetIndex 的先例，只在待发恰好 一条时回滚它，两条及以上不猜、但也不装作没事，只如实提示一句。 - 广播 pending_full 与定向 error{chat_busy} 会同时到达发送者，两者共用 onQueueFull； 幂等靠状态本身——第一次已把气泡移出待发列表，第二次判定自然落到 Ignore，不会回滚两次、 也不会提示两遍。 - REST 409 chat_busy 的文案与 WS 侧合并成同一句（契约把两条入口定为同一个 code）， 并点明「这条没发出去、请手动重试」：全链路没有自动重试。 测试 +30（634 -> 664，0 失败 / 0 错误） - WsEventPendingFullTest：两个新键的有/无、老形状仍可解析、会话闸、无载荷与坏帧不抛。 - WsEventErrorFrameTest：chat_busy 定向帧全字段、可选键缺席为 null、未知 code 原样保留、 只带 code 的帧仍可解、只有 message 的老帧行为不变、坏帧不抛。 - QueueFullOutcomeTest：三种事实各自的判定与状态迁移（收编不动输入栏 / 回滚归还正文图片附件 / 不猜时一个气泡都不动），以及键对不上、空键、同一帧处理两次的幂等。 未验证：本端没有设备也没打过真后端，队列满的实机触发路径见交接报告；VM 里 「判定 -> 既有动作」的映射只由编译与纯函数单测覆盖（构造 ChatViewModel 需要 Android 环境的 AppContainer，单测里没有先例）。 --- .../kotlin/top/yeij/cyrene/data/ws/WsEvent.kt | 88 ++++++- .../top/yeij/cyrene/ui/chat/ChatViewModel.kt | 229 +++++++++++++++-- .../cyrene/data/ws/WsEventErrorFrameTest.kt | 109 ++++++++ .../cyrene/data/ws/WsEventPendingFullTest.kt | 109 ++++++++ .../cyrene/ui/chat/QueueFullOutcomeTest.kt | 233 ++++++++++++++++++ 5 files changed, 736 insertions(+), 32 deletions(-) create mode 100644 app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventErrorFrameTest.kt create mode 100644 app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventPendingFullTest.kt create mode 100644 app/src/test/kotlin/top/yeij/cyrene/ui/chat/QueueFullOutcomeTest.kt diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt index 9c36e07..e52746c 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt @@ -280,11 +280,26 @@ sealed class WsEvent { override val type: String get() = "session_interrupted" } - /** 单会话 pending 达上限（`max`，默认 5），拒绝入队。 */ + /** + * 单会话 pending 达上限（`max`，默认 5），拒绝入队。 + * + * 契约 v0.59（协同 `G5`）在载荷上增**两个可选键**，两个都可能**整键缺席**： + * - [clientMsgId] = **被拒那条消息**的客户端幂等键（那条消息没带键时缺席，**绝不伪造**）； + * - [messageId] = 该消息**此前已落库 / 已认领**时的 id（未被认领时缺席）。 + * + * 服务端**绝不在这里新铸 id**：队列满的判定发生在幂等键**认领之前**，代铸一个 id 会让客户端 + * 拿同一个 `clientMsgId` 重试时被判成「重复到达」而永远进不来。于是 [messageId] 非空恰好 + * 意味着「**这条其实早就进去了**」（它是重发撞上满队列）——消费方必须收编，**不是**回滚。 + * + * 该帧是**广播**（每条已认证连接都会收到，含 `viewer`），所以 [clientMsgId] 缺席时**无法 + * 判定被拒的是不是本端那条气泡**；判定规则见 `ChatViewModel.queueFullOutcome`。 + */ data class PendingFull( override val sessionId: String, val mode: String, val max: Int, + val clientMsgId: String? = null, + val messageId: String? = null, override val turnId: String? = null, override val seq: Long? = null, ) : WsEvent() { @@ -436,9 +451,27 @@ sealed class WsEvent { override val type: String get() = "proactive_message" } - /** 服务端错误（非法 JSON / 未知类型 / 语音·子代理·任务错误等）。 */ + /** + * 服务端错误帧。 + * + * 它是**定向**的（只回请求那条连接，不广播），因此**不进角色门**；帧壳通常**不带** + * `sessionId`（会话归属在 `data.sessionId` 里，见 [resolveSessionId]）、也不带 + * `turnId` / `seq`——它不属任何轮次。 + * + * [code] 是**机器可读**的业务码，取值是**开放集合**（契约 v0.59/G5 起至少含 `chat_busy`， + * 与 REST 的 `409 chat_busy` **同码**，客户端一个 code 覆盖两条入口；另有 `bad_request` 等）。 + * 未知码**原样保留**：不抛、不让整帧消失，只是客户端不做语义处置——[message] 照常透出。 + * + * [clientMsgId] / [messageId] 只在 `code == "chat_busy"`（队列满的定向伴生帧， + * `{type:"error", data:{code:"chat_busy", message, sessionId, clientMsgId?, messageId?}}`） + * 时有意义：[clientMsgId] = 被拒那条消息的幂等键，[messageId] 仅当它此前已落库 / 已认领时 + * 才出现（与 [PendingFull] 同一份键，语义也相同）。其余错误码下它们缺席。 + */ data class Error( val message: String, + val code: String? = null, + val clientMsgId: String? = null, + val messageId: String? = null, override val turnId: String? = null, override val seq: Long? = null, override val sessionId: String? = null, @@ -556,8 +589,17 @@ sealed class WsEvent { .let { SessionIdle(resolveSessionId(sessionId, it.sessionId), it.mode, turnId, seq) } "session_interrupted" -> decode(data) .let { SessionInterrupted(resolveSessionId(sessionId, it.sessionId), it.mode, it.dropped, turnId, seq) } - "pending_full" -> decode(data) - .let { PendingFull(resolveSessionId(sessionId, it.sessionId), it.mode, it.max, turnId, seq) } + "pending_full" -> decode(data).let { + PendingFull( + resolveSessionId(sessionId, it.sessionId), + it.mode, + it.max, + it.clientMsgId, + it.messageId, + turnId, + seq, + ) + } // 幂等回执（v0.56 §3.13）：帧壳带 sessionId，data 内也带一份（两者取有值者）。 "chat_duplicate" -> decode(data) .let { @@ -584,7 +626,18 @@ sealed class WsEvent { it.sender, it.timestamp, it.messageId, turnId, seq, sessionId, ) } - "error" -> Error(decode(data).message, turnId, seq, sessionId) + // 定向错误帧（含队列满的伴生帧）：编码是开放集合，原样带上，未知码不做语义处置。 + "error" -> decode(data).let { + Error( + it.message, + it.code, + it.clientMsgId, + it.messageId, + turnId, + seq, + sessionId ?: it.sessionId, + ) + } else -> null } @@ -669,8 +722,18 @@ private data class PendingData(val sessionId: String = "", val mode: String = "" @Serializable private data class InterruptedData(val sessionId: String = "", val mode: String = "", val dropped: Int = 0) +/** + * `pending_full` 载荷：`{sessionId, mode, max, clientMsgId?, messageId?}`。前三个键来自加入 + * `G5` 之前，逐字未变；后两个是 v0.59 的纯增量，服务端**整键缺席**而不是填空串。 + */ @Serializable -private data class PendingFullData(val sessionId: String = "", val mode: String = "", val max: Int = 0) +private data class PendingFullData( + val sessionId: String = "", + val mode: String = "", + val max: Int = 0, + val clientMsgId: String? = null, + val messageId: String? = null, +) /** `chat_duplicate` 载荷（v0.56 §3.13）：`{sessionId, clientMsgId, messageId}`。 */ @Serializable @@ -715,5 +778,16 @@ private data class ProactiveData( val messageId: String? = null, ) +/** + * `error` 帧载荷。`code` / `sessionId` / `clientMsgId` / `messageId` 都是 v0.59 起的可选键 + * （队列满的定向伴生帧带的正是这四个加 `message`），其余错误帧只带 `message`， + * 甚至**一个字段都不带**（如 `{code:"bad_request"}`）——都按「拿不到就不做事」解码。 + */ @Serializable -private data class ErrorData(val message: String = "") +private data class ErrorData( + val message: String = "", + val code: String? = null, + val sessionId: String? = null, + val clientMsgId: String? = null, + val messageId: String? = null, +) diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt index dda9537..c8a28fe 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt @@ -97,6 +97,142 @@ fun consumePendingFor( return pending.filterIndexed { i, _ -> i != index } } +/** + * 队列满被拒（**真没发出去**）的如实文案。 + * + * WS 的定向 `error{code:"chat_busy"}` 与 REST 的 `409 chat_busy` 共用这一句——契约把两者定为 + * **同一个 code**（客户端一个 code 覆盖两条入口），用户看到的也就该是同一件事、同一句话。 + * 「请稍后**手动**重试」是刻意的措辞：全链路**没有幂等之外的兜底、没有自动重试**， + * 要不要再发一次永远由用户决定。 + */ +const val QUEUE_FULL_MESSAGE = "她正忙且队列已满，这条没发出去，请稍后手动重试" + +/** + * 队列满但**定位不到是哪一条**时的如实文案（老服务端不带键、本会话又有多条待发）。 + * + * 它与 [QUEUE_FULL_MESSAGE] 分开，因为本端**什么都没做**：没有回滚任何气泡，也说不准是 + * 谁没发出去。宁可说得含糊（「可能」），也不能假装没事，更不能挑一条气泡乱回滚。 + */ +const val QUEUE_FULL_UNKNOWN_MESSAGE = "她忙不过来，这条可能没发出去" + +/** 队列满的机器可读码（契约 v0.59/G5，与 REST `409` 同码）。 */ +const val CHAT_BUSY_CODE = "chat_busy" + +/** + * 一帧「队列满」该对本地乐观气泡做什么（纯函数的结果，见 [queueFullOutcome]）。 + * + * 三档分别对应三种**事实**，绝不能互相顶替：[Adopt] 是「其实已经进去了」、[Rollback] 是 + * 「确实被拒」、[Ambiguous] 是「说不准，所以什么都不动、只如实说一句」。 + */ +sealed interface QueueFullOutcome { + /** + * 那条**此前已落库 / 已认领**（服务端给了 `messageId`）⇒ 收编：拿掉乐观气泡、对齐既有消息。 + * **不是失败**：重发撞上满队列，服务端只是拒绝第二次，第一次的内容早就在了。 + */ + data class Adopt(val localId: String, val messageId: String) : QueueFullOutcome + + /** 真的被拒 ⇒ 回滚这一条（正文 / 图片 / 附件一起还给输入栏）。 */ + data class Rollback(val localId: String, val reason: String) : QueueFullOutcome + + /** 定位不到是哪一条 ⇒ **不猜**，但也**不装作没事**，只如实提示。 */ + data class Ambiguous(val reason: String) : QueueFullOutcome + + /** 无事可做：本会话没有待发气泡，或这个键明确不属于本端发出的任何一条。 */ + data object Ignore : QueueFullOutcome +} + +/** + * 队列满（广播 `pending_full` 与定向 `error{code:"chat_busy"}` **共用**）的判定（纯函数）。 + * + * 判定顺序就是**证据强度**的顺序： + * + * 1. **`clientMsgId` 命中某条待发气泡**（键是本端自己铸的 `localId`，命中即同一条消息， + * 文本不同也不会认错）——此时 `messageId` 给不给决定了那条消息的**真相**： + * - 非空 ⇒ 已被认领 ⇒ [QueueFullOutcome.Adopt]； + * - 缺席 ⇒ 拒绝发生在认领之前 ⇒ [QueueFullOutcome.Rollback]。 + * 2. **`clientMsgId` 有值但不命中** ⇒ [QueueFullOutcome.Ignore]。带键的帧只属于**某一次具体的 + * 发送**，键对不上就说明它不是本端这条（别的端发的、或本端那条已经被收编过）；此时去回滚 + * 任何一条气泡都不是「对齐」，而是误伤。同一帧因广播 + 定向**到达两次**也走这一档： + * 第一次已经处置完了，第二次自然匹配不到。 + * 3. **`clientMsgId` 缺席**（老服务端，或那条消息本就没带键）⇒ **别猜**。沿用 + * `toolExecutionTargetIndex` 的先例，只在**无歧义**时才落位： + * - 待发列表空 ⇒ [QueueFullOutcome.Ignore]（没有气泡会挂住，没有要说的）； + * - 恰好一条 ⇒ 就是它（本会话只有这一条在飞，不可能是别的）； + * - 两条及以上 ⇒ [QueueFullOutcome.Ambiguous]：挑一条回滚等于替用户瞎猜， + * 而广播帧本身**不带**任何能指向本端某一条的信息。 + * + * 不修改入参，无副作用。`pending` 的顺序与长度都被如实使用（[Adopt] / [Rollback] 只针对**一条**）。 + */ +fun queueFullOutcome( + pending: List, + clientMsgId: String?, + messageId: String?, +): QueueFullOutcome { + // 空白等同缺席：服务端「整键缺席」，但真出现空串时按同一件事处理（与 consumePendingFor 一致）。 + val key = clientMsgId?.takeIf { it.isNotBlank() } + val known = messageId?.takeIf { it.isNotBlank() } + + if (key != null) { + val hit = pending.firstOrNull { it.localId == key } ?: return QueueFullOutcome.Ignore + return if (known != null) { + QueueFullOutcome.Adopt(hit.localId, known) + } else { + QueueFullOutcome.Rollback(hit.localId, QUEUE_FULL_MESSAGE) + } + } + + return when (pending.size) { + 0 -> QueueFullOutcome.Ignore + 1 -> if (known != null) { + QueueFullOutcome.Adopt(pending.single().localId, known) + } else { + QueueFullOutcome.Rollback(pending.single().localId, QUEUE_FULL_MESSAGE) + } + + else -> QueueFullOutcome.Ambiguous(QUEUE_FULL_UNKNOWN_MESSAGE) + } +} + +/** + * 把一次队列满的判定落到界面状态上（纯函数，VM 与测试共用）。 + * + * 三档的事实不同，落到状态上的动作也就必须不同： + * - [QueueFullOutcome.Adopt]：**只拿掉乐观气泡**——那条消息服务端已经有了，正文/图片/附件 + * 都不该退回输入栏（退回去等于诱导用户把一条**已经发出去**的消息再发一遍）； + * - [QueueFullOutcome.Rollback]：走 [rollbackPending]，正文 / 图片 / 附件一起还给输入栏； + * - [QueueFullOutcome.Ambiguous]：**一个气泡都不动**，只留一句提示（判据缺失时乱动就是误伤）； + * - [QueueFullOutcome.Ignore]：原样返回同一个实例（Compose 的 key 稳定性靠它）。 + * + * 补拉「本地还没见过的 messageId」是副作用（要走网络），留在调用方；这里只管状态。 + */ +fun ChatUiState.applyQueueFull(outcome: QueueFullOutcome): ChatUiState = when (outcome) { + is QueueFullOutcome.Adopt -> copy(pending = pending.filterNot { it.localId == outcome.localId }) + is QueueFullOutcome.Rollback -> rollbackPending(outcome.localId, outcome.reason) + is QueueFullOutcome.Ambiguous -> copy(error = outcome.reason) + QueueFullOutcome.Ignore -> this +} + +/** + * 回滚一条乐观气泡：把它从待发列表里拿掉，并把文本 / 图片 / 附件**还给输入栏**（纯函数）。 + * + * 挂掉的气泡本身就是权威副本，所以内容从它身上取： + * - 空文本（纯图片或纯附件）**不覆盖**用户当前正在输入的内容； + * - 图片与附件照原样回到待发位，不必让用户重挑（附件已经上传过的那几份在服务端是孤儿， + * 但本地引用仍然有效、可直接重发）； + * - 哪一条都没命中（气泡已被别的路径收走）时，输入栏与原因照旧，只是没有气泡可摘—— + * 同一件事被两条帧各报一次（如广播 `pending_full` 与定向 `error{chat_busy}`）时不会重复归还。 + */ +fun ChatUiState.rollbackPending(localId: String, reason: String): ChatUiState { + val failed = pending.firstOrNull { it.localId == localId } + return copy( + pending = pending.filterNot { it.localId == localId }, + draft = failed?.text?.ifBlank { draft } ?: draft, + draftImage = failed?.imageUri ?: draftImage, + draftDocuments = failed?.documents?.takeIf { it.isNotEmpty() } ?: draftDocuments, + error = reason, + ) +} + /** * 工具执行的细粒度阶段，与后端流程图 tool 节点 status **同一套枚举**（契约 v0.50 §3.1 / §3.5）。 * `null` = 还没收到任何 `tool_execution.*` 帧（只有调用与参数）。 @@ -269,6 +405,10 @@ data class ChatUiState( * * 本 VM 只收**属于 [sessionId] 的帧**（见 [onEvent] 的会话闸）。服务端广播不按会话订阅， * 且每个会话各有一个 VM（各自的 socket 都还活着），没有这道闸就会串台。 + * + * WS 上行**没有** REST 那样的同步回执：一条帧发出去之后，只有队列满（`pending_full` 广播 + + * 定向 `error{chat_busy}`）与幂等重复（`chat_duplicate`）这两类信号能说明「这条其实没进去 / + * 早就进去了」。两类的处置都在 [onQueueFull] / [alignDuplicate]，判据是纯函数，可单测。 */ class ChatViewModel( private val container: AppContainer, @@ -659,8 +799,9 @@ class ChatViewModel( is ChatAck.Duplicate -> alignDuplicate(localId, ack.messageId) // 409 队列满：这条**没有被受理、没有落库**（服务端在入队判定处直接拒绝）， - // 所以不能当成「已排队」——回滚气泡并把文本还回输入框，用的是既有忙碌文案的口径。 - ChatAck.QueueFull -> rollback(localId, "会话正忙，待处理消息已满，请稍后再发") + // 所以不能当成「已排队」——回滚气泡并把文本还回输入框。文案与 WS 路径的定向 + // `error{chat_busy}` 共用一句（契约把两者定为同一个 code；见 [QUEUE_FULL_MESSAGE]）。 + ChatAck.QueueFull -> rollback(localId, QUEUE_FULL_MESSAGE) is ChatAck.Unavailable -> rollback(localId, "服务端暂时接不住这一轮，请稍后重试") @@ -674,19 +815,32 @@ class ChatViewModel( } /** - * 幂等重复的对齐（REST `{duplicate:true}` 与 WS `chat_duplicate` 共用）： - * 去掉那条乐观气泡；本地若**没有**这条消息，再补拉最新一页。 + * 「这条早已受理过」的对齐：去掉那条乐观气泡；本地若**没有**这条消息，再补拉最新一页。 * - * 复用同一个 `clientMsgId` 重试时，正常路径本端早就收到过 `message_persisted`（第一次发的 - * 时候），但弱网 / 重连丢帧会留下「气泡没了、消息也没了」的空档——用 `messageId` 补一次 - * 增量拉取才是真的「对齐到服务端真相」，而不是让气泡凭空消失。 + * 两个来源共用：REST `{duplicate:true}` / WS `chat_duplicate`（幂等键重复到达），以及队列满 + * 帧里带了 `messageId` 的**收编**（[queueFullOutcome] 的 [QueueFullOutcome.Adopt]——那条其实 + * 早就落库了，只是重发撞上了满队列）。两种情况对客户端是同一件事：**服务端已有这条、 + * 本端不必再发、也不是失败**。 + * + * 第一段（摘气泡）与 [applyQueueFull] 的 Adopt 档等价，只是键来自帧而不是本地判定。 */ private fun alignDuplicate(clientMsgId: String, messageId: String?) { _state.update { state -> state.copy(pending = state.pending.filterNot { it.localId == clientMsgId }) } - val known = messageId != null && _state.value.messages.any { it.id == messageId } - if (messageId != null && !known) refreshLatest() + if (messageId != null) ensureMessageKnown(messageId) + } + + /** + * 本地不认识这条 `messageId` 时补拉最新一页（认得的就什么也不做，省一次往返）。 + * + * 为什么要补：复用同一个 `clientMsgId` 重试 / 收编时，正常路径本端早就收到过 + * `message_persisted`（第一次发的时候），但弱网 / 重连丢帧会留下「气泡没了、消息也没了」 + * 的空档——补一次增量拉取才是真的「对齐到服务端真相」，而不是让气泡凭空消失。 + */ + private fun ensureMessageKnown(messageId: String) { + if (_state.value.messages.any { it.id == messageId }) return + refreshLatest() } /** 静默补拉最新一页（成功才并进来；失败不改状态——这只是一次对齐尝试）。 */ @@ -720,22 +874,12 @@ class ChatViewModel( .fold(pending) { acc, message -> consumePendingFor(acc, message) } /** - * 回滚乐观气泡：把文本 / 图片 / 附件还给输入框并给出原因（挂掉的气泡本身就是权威副本）。 - * 空文本（纯图片或纯附件）不覆盖用户当前正在输入的内容；图片与附件照原样还给待发位， - * 不必让用户重挑（附件已经上传过的那几份在服务端是孤儿，但本地引用仍然有效、可直接重发）。 + * 回滚乐观气泡：把文本 / 图片 / 附件还给输入框并给出原因。 + * + * 迁移本身是纯函数 [rollbackPending]（可单测）；这里只把它接进状态流。 */ - private fun rollback(localId: String, reason: String) { - _state.update { state -> - val failed = state.pending.firstOrNull { it.localId == localId } - state.copy( - pending = state.pending.filterNot { it.localId == localId }, - draft = failed?.text?.ifBlank { state.draft } ?: state.draft, - draftImage = failed?.imageUri ?: state.draftImage, - draftDocuments = failed?.documents?.takeIf { it.isNotEmpty() } ?: state.draftDocuments, - error = reason, - ) - } - } + private fun rollback(localId: String, reason: String) = + _state.update { it.rollbackPending(localId, reason) } private fun onEvent(event: WsEvent) { // 广播是发给每条已认证连接的（服务端没有按会话订阅），所以每个 VM 都会收到**所有**会话的帧。 @@ -859,12 +1003,47 @@ class ChatViewModel( _state.update { it.copy(busyMode = null, queueDepth = 0) } } - is WsEvent.Error -> _state.update { it.copy(error = event.message) } + // 队列满（契约 v0.59/G5）：广播帧，**每条**连接都收得到，所以键对不上就必须放手。 + is WsEvent.PendingFull -> onQueueFull(event.sessionId, event.clientMsgId, event.messageId) + + is WsEvent.Error -> when { + // 队列满的定向伴生帧（只发发送者，与上面的广播帧**同一个 code、同一套处置**）。 + // 它不广播，所以走到这里的必然是本端那条连接；会话归属仍以帧里给的为准。 + event.code == CHAT_BUSY_CODE -> + onQueueFull(event.sessionId, event.clientMsgId, event.messageId) + + // 其余错误码原样透出，未知码也不吞：没有文案可说的（如 `bad_request` 只带 code） + // 就不摆一个空气泡，但事件本身已经完整解码、没有被丢弃。 + event.message.isNotBlank() -> _state.update { it.copy(error = event.message) } + + else -> Unit + } else -> Unit } } + /** + * 队列满的处置：广播 `pending_full` 与定向 `error{code:"chat_busy"}` **共用同一条路** + * （契约 v0.59/G5：「一个 code 覆盖两条入口」）。判定本身是纯函数 [queueFullOutcome]， + * 这里只做三件事：会话闸、把判定落到既有动作（[alignDuplicate] / [rollback] / 提示）、 + * 以及**保证重复到达不重复处置**。 + * + * 幂等靠的是**状态本身**：第一次处置要么把气泡移出待发列表（收编 / 回滚），要么什么都没做 + * （无歧义判定）。发送者本来就会同时收到广播帧与定向帧两条，第二次到达时待发列表已经变了， + * [queueFullOutcome] 自然落到 [QueueFullOutcome.Ignore]，不会回滚两次、也不会提示两遍。 + * + * @param frameSessionId 帧给的会话归属；空白 = 无法判定，按既有约定**放行** + */ + private fun onQueueFull(frameSessionId: String?, clientMsgId: String?, messageId: String?) { + if (!frameSessionId.isNullOrBlank() && frameSessionId != sessionId) return + val outcome = queueFullOutcome(_state.value.pending, clientMsgId, messageId) + _state.update { it.applyQueueFull(outcome) } + // 收编时本地可能还不认识那条消息（弱网 / 重连丢了 message_persisted）——补拉一页， + // 别只让气泡凭空消失。这是与幂等重复完全同一件事，所以共用 [ensureMessageKnown]。 + if (outcome is QueueFullOutcome.Adopt) ensureMessageKnown(outcome.messageId) + } + /** 新一轮开始时，若上一轮的流式状态已结束（active=false）就清掉，避免思维链/工具卡串台。 */ private fun freshTurnIfFinished() { _state.update { state -> diff --git a/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventErrorFrameTest.kt b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventErrorFrameTest.kt new file mode 100644 index 0000000..366d740 --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventErrorFrameTest.kt @@ -0,0 +1,109 @@ +package top.yeij.cyrene.data.ws + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * 定向 `error` 帧（契约 v0.59 §3.2，协同 `G5` 的伴生帧）。 + * + * 队列满时服务端**除广播 `pending_full` 外**，还向发送那条消息的连接单独回一条 + * `{type:"error", data:{code:"chat_busy", message, sessionId, clientMsgId?, messageId?}}` + * ——它是 WS 路径上 REST `409 chat_busy` 的等价物（**同码**，客户端一个 code 覆盖两条入口）。 + * + * 它**只发发送者、不广播**，因此帧壳上没有 `sessionId`（会话归属在 `data.sessionId` 里）， + * 也没有 `turnId` / `seq`。`code` 是**开放集合**：客户端不得对未知码抛异常、更不得因为 + * 认不出码就把整帧吞掉——帧里还有服务端给的文案，那是能直接给用户看的唯一东西。 + */ +class WsEventErrorFrameTest { + + /** 真实帧形状（服务端 `ws_route.go:notifyPendingFull`）。 */ + private val realBusyFrame = """ + {"type":"error","data":{"code":"chat_busy","message":"会话正忙，待处理消息已满，本条未入队","sessionId":"web","clientMsgId":"local-1","messageId":"m-42"}} + """.trimIndent() + + @Test + fun `the directed chat_busy frame decodes with every field`() { + val event = WsEvent.parse(realBusyFrame) as WsEvent.Error + + assertEquals("error", event.type) + assertEquals("chat_busy", event.code) + assertEquals("会话正忙，待处理消息已满，本条未入队", event.message) + // 会话归属**只在载荷里**：帧壳上没有 sessionId。 + assertEquals("web", event.sessionId) + assertEquals("local-1", event.clientMsgId) + assertEquals("m-42", event.messageId) + assertNull(event.turnId) + assertNull(event.seq) + } + + @Test + fun `the optional identity keys stay null when the server omits them`() { + val frame = """{"type":"error","data":{"code":"chat_busy","message":"满","sessionId":"web"}}""" + + val event = WsEvent.parse(frame) as WsEvent.Error + + assertNull("未携带就该是 null，不拍成空串", event.clientMsgId) + assertNull(event.messageId) + } + + @Test + fun `an unknown code is kept verbatim instead of throwing or swallowing the frame`() { + // 开放集合：服务端将来加码，老客户端**必须**还能把这句话显示出来。 + val event = WsEvent.parse( + """{"type":"error","data":{"code":"some_future_code","message":"以后才会有的话","sessionId":"web"}}""", + ) as WsEvent.Error + + assertEquals("some_future_code", event.code) + assertEquals("以后才会有的话", event.message) + assertEquals("web", event.sessionId) + } + + @Test + fun `a code-only error frame still decodes`() { + // 契约里真实存在这种形状：interrupt 的 scope 非法时只回 {code:"bad_request"}。 + val event = WsEvent.parse("""{"type":"error","data":{"code":"bad_request"}}""") as WsEvent.Error + + assertEquals("bad_request", event.code) + assertEquals("", event.message) + assertNull(event.sessionId) + } + + @Test + fun `a message-only error frame keeps working exactly as before`() { + // v0.59 之前的所有 error 帧都只有 message，这条通路一个字节都不该变。 + val event = WsEvent.parse("""{"type":"error","data":{"message":"voice unavailable"}}""") as WsEvent.Error + + assertEquals("voice unavailable", event.message) + assertNull(event.code) + assertNull(event.sessionId) + } + + @Test + fun `the shell sessionId wins over the payload one`() { + val event = WsEvent.parse( + """{"type":"error","sessionId":"sess-9c3","data":{"code":"chat_busy","message":"满","sessionId":"web"}}""", + ) as WsEvent.Error + + assertEquals("sess-9c3", event.sessionId) + } + + @Test + fun `a broken error payload is ignored and never throws`() { + assertNull(WsEvent.parse("""{"type":"error","data":"boom"}""")) + assertNull(WsEvent.parse("""{"type":"error","data":{"code":42}}""")) + } + + @Test + fun `an error carrying a sessionId is scoped to it`() { + val event = WsEvent.parse(realBusyFrame)!! + val sessionless = WsEvent.parse("""{"type":"error","data":{"message":"boom"}}""")!! + + assertTrue(event.isForSession("web")) + assertFalse(event.isForSession("web:other")) + // 帧里没给会话时是「无法判定」而不是「不属于本会话」，仍放行（旧服务端 / 无归属的帧）。 + assertTrue(sessionless.isForSession("web")) + } +} diff --git a/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventPendingFullTest.kt b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventPendingFullTest.kt new file mode 100644 index 0000000..2a0e05f --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventPendingFullTest.kt @@ -0,0 +1,109 @@ +package top.yeij.cyrene.data.ws + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * 下行帧 `pending_full`（契约 v0.59 / 协同 `G5`）。 + * + * 载荷在既有的 `{sessionId, mode, max}` 上**纯增量**加了两个可选键：`clientMsgId`（被拒那条 + * 消息的幂等键）与 `messageId`（仅当它此前已落库 / 已认领）。判定这条气泡该收编还是该回滚 + * 全靠这两个键，所以「有没有键」必须能被分辨——服务端**整键缺席**，客户端也就必须区分 + * 「键不在」与「键是空串」，而不是把两者拍成同一个值。 + */ +class WsEventPendingFullTest { + + /** 真实帧形状（服务端 `session_reject.go:pendingFullPayload`）。 */ + private val realFrame = """ + {"type":"pending_full","sessionId":"web","data":{"sessionId":"web","mode":"thinking","max":5,"clientMsgId":"local-1","messageId":"m-42"}} + """.trimIndent() + + /** 队列满但那条消息**还没被认领**：只有幂等键，没有 messageId。 */ + private val unclaimedFrame = """ + {"type":"pending_full","data":{"sessionId":"web","mode":"executing","max":5,"clientMsgId":"local-2"}} + """.trimIndent() + + /** 老形状（v0.59 之前）：两个新键一个都不在。 */ + private val legacyFrame = """ + {"type":"pending_full","sessionId":"web","data":{"sessionId":"web","mode":"thinking","max":5}} + """.trimIndent() + + @Test + fun `the v059 frame carries both identity keys`() { + val event = WsEvent.parse(realFrame) as WsEvent.PendingFull + + assertEquals("pending_full", event.type) + assertEquals("web", event.sessionId) + assertEquals("thinking", event.mode) + assertEquals(5, event.max) + assertEquals("local-1", event.clientMsgId) + assertEquals("m-42", event.messageId) + // 状态机信号不认识轮次：帧壳没有这两个字段。 + assertNull(event.turnId) + assertNull(event.seq) + } + + @Test + fun `a missing messageId stays null and is never fabricated`() { + val event = WsEvent.parse(unclaimedFrame) as WsEvent.PendingFull + + assertEquals("local-2", event.clientMsgId) + // 缺席就是 null（不是空串）：「服务端没给 id」与「服务端给了个空 id」是两件事， + // 后者若是被拍成空串，收编路径就会去对齐一条不存在的消息。 + assertNull(event.messageId) + } + + @Test + fun `the legacy shape still parses with both keys absent`() { + // 老服务端（或本端自己发的帧根本没带幂等键）→ 两个键都缺席。 + // 判定这一档的规则见 queueFullOutcome：只在无歧义时才落位。 + val event = WsEvent.parse(legacyFrame) as WsEvent.PendingFull + + assertEquals("web", event.sessionId) + assertEquals(5, event.max) + assertNull(event.clientMsgId) + assertNull(event.messageId) + } + + @Test + fun `the shell sessionId wins and the payload sessionId fills in`() { + val fromShell = WsEvent.parse( + """{"type":"pending_full","sessionId":"web:abc","data":{"mode":"thinking","max":5}}""", + ) as WsEvent.PendingFull + val fromPayload = WsEvent.parse( + """{"type":"pending_full","data":{"sessionId":"web:xyz","mode":"thinking","max":5}}""", + ) as WsEvent.PendingFull + + assertEquals("web:abc", fromShell.sessionId) + assertEquals("web:xyz", fromPayload.sessionId) + } + + @Test + fun `pending_full is scoped to its own session`() { + val event = WsEvent.parse(realFrame)!! + + assertTrue(event.isForSession("web")) + // 广播是发给每条连接的、不按会话订阅：别的会话的队列满绝不能去动本会话的气泡。 + assertFalse(event.isForSession("web:other")) + } + + @Test + fun `a pending_full without any payload still parses instead of being dropped`() { + val event = WsEvent.parse("""{"type":"pending_full","sessionId":"web"}""") as WsEvent.PendingFull + + assertEquals("web", event.sessionId) + assertEquals("", event.mode) + assertEquals(0, event.max) + assertNull(event.clientMsgId) + } + + @Test + fun `a structurally broken pending_full is ignored and never throws`() { + // `data` 不是对象 ⇒ 解码失败 ⇒ 整帧按坏帧安全忽略（与未知 type 同一条出口）。 + assertNull(WsEvent.parse("""{"type":"pending_full","sessionId":"web","data":"oops"}""")) + assertNull(WsEvent.parse("""{"type":"pending_full","sessionId":"web","data":{"max":"五"}}""")) + } +} diff --git a/app/src/test/kotlin/top/yeij/cyrene/ui/chat/QueueFullOutcomeTest.kt b/app/src/test/kotlin/top/yeij/cyrene/ui/chat/QueueFullOutcomeTest.kt new file mode 100644 index 0000000..93242bd --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/ui/chat/QueueFullOutcomeTest.kt @@ -0,0 +1,233 @@ +package top.yeij.cyrene.ui.chat + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue +import org.junit.Test +import top.yeij.cyrene.domain.chat.DraftDocument + +/** + * 队列满的判定与状态迁移（契约 v0.59 / 协同 `G5`）。 + * + * 这条链路此前是「那条气泡永远挂发送中…」：WS 上行没有同步回执，队列满时服务端只广播一条 + * `pending_full`（载荷连是哪条都没说），客户端既无法回滚也无法收编。[queueFullOutcome] 是判定 + * 本体（纯函数），[ChatUiState.applyQueueFull] 把它落到状态上——两者都不碰协程，因此可以在这里 + * 逐条钉死「什么时候动哪条气泡、什么时候一个都不许动」。 + * + * 三个档位对应三种**不同的事实**，测试也按事实分组：收编（其实已经进去了）、回滚（确实被拒）、 + * 不猜（说不准，只能如实说一句）。 + */ +class QueueFullOutcomeTest { + + private fun pending( + localId: String, + text: String = "你好", + imageUri: String? = null, + documents: List = emptyList(), + ) = PendingMessage(localId = localId, text = text, imageUri = imageUri, documents = documents) + + private val draft = DraftDocument(uri = "content://doc", displayName = "笔记.md") + + // —— 收编：messageId 非空 = 那条其实早已落库 / 已认领 —— + + @Test + fun `a non-empty messageId adopts the bubble instead of rolling it back`() { + val pending = listOf(pending("local-a"), pending("local-b")) + + val outcome = queueFullOutcome(pending, clientMsgId = "local-b", messageId = "m-42") + + // 重发撞上满队列：服务端只是拒绝了**第二次**，第一次的内容早就在了 ⇒ 收编，不是失败。 + assertEquals(QueueFullOutcome.Adopt("local-b", "m-42"), outcome) + } + + @Test + fun `adopting leaves the draft and the error untouched`() { + val state = ChatUiState( + pending = listOf(pending("local-a", text = "这条", imageUri = "content://img", documents = listOf(draft))), + draft = "用户正在输入的下一条", + draftImage = null, + draftDocuments = emptyList(), + error = null, + ) + + val after = state.applyQueueFull( + queueFullOutcome(state.pending, clientMsgId = "local-a", messageId = "m-1"), + ) + + // 关键：收编**不能**把正文 / 图片 / 附件退回输入栏——那等于诱导用户把一条**已经发出去** + // 的消息再发一遍（而且会覆盖他正在输入的内容）。 + assertTrue(after.pending.isEmpty()) + assertEquals("用户正在输入的下一条", after.draft) + assertNull(after.draftImage) + assertTrue(after.draftDocuments.isEmpty()) + assertNull(after.error) + } + + // —— 回滚：messageId 缺席 = 拒绝发生在幂等键认领之前 —— + + @Test + fun `a missing messageId rolls the identified bubble back`() { + val pending = listOf(pending("local-a"), pending("local-b")) + + val outcome = queueFullOutcome(pending, clientMsgId = "local-a", messageId = null) + + assertEquals(QueueFullOutcome.Rollback("local-a", QUEUE_FULL_MESSAGE), outcome) + } + + @Test + fun `rolling back returns text image and documents to the composer`() { + val state = ChatUiState( + pending = listOf(pending("local-a", text = "被拒的正文", imageUri = "content://img", documents = listOf(draft))), + draft = "", + draftImage = null, + draftDocuments = emptyList(), + ) + + val after = state.applyQueueFull( + queueFullOutcome(state.pending, clientMsgId = "local-a", messageId = null), + ) + + assertTrue(after.pending.isEmpty()) + assertEquals("被拒的正文", after.draft) + assertEquals("content://img", after.draftImage) + assertEquals(listOf(draft), after.draftDocuments) + assertEquals(QUEUE_FULL_MESSAGE, after.error) + } + + @Test + fun `rolling back a pure-image bubble does not clobber what the user is typing`() { + val state = ChatUiState( + pending = listOf(pending("local-a", text = "", imageUri = "content://img")), + draft = "正在输入", + ) + + val after = state.applyQueueFull( + queueFullOutcome(state.pending, clientMsgId = "local-a", messageId = null), + ) + + assertEquals("正在输入", after.draft) + assertEquals("content://img", after.draftImage) + assertEquals(QUEUE_FULL_MESSAGE, after.error) + } + + @Test + fun `rolling back a bubble that is already gone changes nothing but the reason`() { + // 同一件事会被两条帧各报一次（广播 pending_full + 定向 error{chat_busy}）， + // 第二次到达时气泡早被摘走了：这时绝不能把输入栏再覆盖一遍。 + val state = ChatUiState(pending = emptyList(), draft = "用户后来打的字", error = null) + + val after = state.applyQueueFull( + queueFullOutcome(state.pending, clientMsgId = "local-a", messageId = null), + ) + + // 待发列表本来就是空的 ⇒ 判定落到「无歧义才落位」那一档：没有待发消息 ⇒ 静默。 + assertSame(state, after) + } + + // —— 老路径：clientMsgId 缺席，只在无歧义时才落位 —— + + @Test + fun `an absent clientMsgId rolls back the only pending bubble`() { + val pending = listOf(pending("local-a")) + + val outcome = queueFullOutcome(pending, clientMsgId = null, messageId = null) + + // 本会话只有这一条在飞，不可能是别的 ⇒ 无歧义。 + assertEquals(QueueFullOutcome.Rollback("local-a", QUEUE_FULL_MESSAGE), outcome) + } + + @Test + fun `an absent clientMsgId with several pending bubbles guesses nothing`() { + val pending = listOf(pending("local-a"), pending("local-b")) + + val outcome = queueFullOutcome(pending, clientMsgId = null, messageId = null) + + // 挑一条回滚等于替用户瞎猜（被拒的可能是别的端发的那条）⇒ 不猜，但也不装作没事。 + assertEquals(QueueFullOutcome.Ambiguous(QUEUE_FULL_UNKNOWN_MESSAGE), outcome) + } + + @Test + fun `the ambiguous verdict moves no bubble at all`() { + val state = ChatUiState( + pending = listOf(pending("local-a", text = "甲"), pending("local-b", text = "乙")), + draft = "不动的草稿", + error = null, + ) + + val after = state.applyQueueFull(queueFullOutcome(state.pending, null, null)) + + assertEquals(listOf("local-a", "local-b"), after.pending.map { it.localId }) + assertEquals("不动的草稿", after.draft) + assertEquals(QUEUE_FULL_UNKNOWN_MESSAGE, after.error) + } + + @Test + fun `an absent clientMsgId with nothing pending says nothing`() { + // 没有气泡会挂住，也就没有要说的：不能凭空弹一句提示。 + val state = ChatUiState() + + assertSame(state, state.applyQueueFull(queueFullOutcome(emptyList(), null, null))) + assertEquals(QueueFullOutcome.Ignore, queueFullOutcome(emptyList(), null, null)) + } + + // —— 键对不上就放手 —— + + @Test + fun `an unknown clientMsgId never falls back to text matching`() { + val pending = listOf(pending("local-a", text = "同文本")) + + val outcome = queueFullOutcome(pending, clientMsgId = "local-other", messageId = null) + + // 带键的帧只属于**某一次具体的发送**（别的端发的、或本端那条已被收编）。 + // 回滚任何一条都不是「对齐」，而是误伤。 + assertEquals(QueueFullOutcome.Ignore, outcome) + } + + @Test + fun `handling the same frame twice does not roll back twice`() { + val state = ChatUiState(pending = listOf(pending("local-a", text = "只发一次")), draft = "", error = null) + + val first = state.applyQueueFull(queueFullOutcome(state.pending, "local-a", null)) + // 广播帧与定向帧都会到发送者手上；第二次判定必须落到 Ignore（气泡已经不在待发列表里）。 + val outcome2 = queueFullOutcome(first.pending, "local-a", null) + val second = first.applyQueueFull(outcome2) + + assertEquals(QueueFullOutcome.Ignore, outcome2) + assertEquals("只发一次", second.draft) + assertEquals(QUEUE_FULL_MESSAGE, second.error) + } + + @Test + fun `a blank key counts as absent on both sides`() { + val pending = listOf(pending("local-a")) + + // 服务端是「整键缺席」，但真给空串时按同一件事处理（与 consumePendingFor 一致）。 + assertEquals( + QueueFullOutcome.Rollback("local-a", QUEUE_FULL_MESSAGE), + queueFullOutcome(pending, clientMsgId = " ", messageId = ""), + ) + } + + @Test + fun `a blank messageId is not an adoption`() { + val pending = listOf(pending("local-a")) + + // 空串不是 id：必须按「没有被认领」处理，否则会去对齐一条不存在的消息。 + assertEquals( + QueueFullOutcome.Rollback("local-a", QUEUE_FULL_MESSAGE), + queueFullOutcome(pending, clientMsgId = "local-a", messageId = " "), + ) + } + + @Test + fun `no verdict mutates the pending input`() { + val pending = listOf(pending("local-a"), pending("local-b")) + + queueFullOutcome(pending, clientMsgId = "local-a", messageId = "m-1") + queueFullOutcome(pending, clientMsgId = null, messageId = null) + queueFullOutcome(pending, clientMsgId = "local-z", messageId = null) + + assertEquals(listOf("local-a", "local-b"), pending.map { it.localId }) + } +}