feat(ws): 队列满时定位并处置那条待发气泡(契约 v0.59 / 协同 G5)

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,单测里没有先例)。
This commit is contained in:
AskaEth committed 2026-10-04 19:41:40 +08:00
1 parent 8633ec9f06
commit 3b4a790e30
5 files changed
+736 -32

No files matched your search

@@ -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<InterruptedData>(data)
.let { SessionInterrupted(resolveSessionId(sessionId, it.sessionId), it.mode, it.dropped, turnId, seq) }
"pending_full" -> decode<PendingFullData>(data)
.let { PendingFull(resolveSessionId(sessionId, it.sessionId), it.mode, it.max, turnId, seq) }
"pending_full" -> decode<PendingFullData>(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<ChatDuplicateData>(data)
.let {
@@ -584,7 +626,18 @@ sealed class WsEvent {
it.sender, it.timestamp, it.messageId, turnId, seq, sessionId,
)
}
"error" -> Error(decode<ErrorData>(data).message, turnId, seq, sessionId)
// 定向错误帧(含队列满的伴生帧):编码是开放集合,原样带上,未知码不做语义处置。
"error" -> decode<ErrorData>(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,
)
@@ -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<PendingMessage>,
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 ->
@@ -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"))
}
}
@@ -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":"五"}}"""))
}
}
@@ -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<DraftDocument> = 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 })
}
}