feat(chat): 忙时并行回合的客户端半边 —— 六态 + 灰态 + (sessionId, turnId) 两级分桶 + 新帧消费(全部可降级)

依据:Cyrene_v2/DevDocs/_specs/忙时并行-规范草案.md(§1 行为 / §2 接口 / §4 主人裁决)。
后端 v0.65 尚未落地这些帧,所以本次的新键与新帧全部按可降级实现:帧不来、字段缺席时,
App 在今天的真后端上行为与改前逐字相同(下面的「降级保证」逐条给了证据)。

一、改了哪些文件

新增(纯逻辑,可单测)
- app/src/main/kotlin/top/yeij/cyrene/domain/chat/ParallelTurns.kt
  两级分桶键 TurnBucket(sessionId, turnId) 与 turnBucketOf / closingBucketOf;
  六态枚举 TurnPhase 与 turnPhaseOf;灰态收口判据 closesAnswering;
  消息↔回合绑定 TurnBinding 及其纯函数(withTurnBegin / withSteeredMessage / closingTurn /
  allClosed / bindingOfMessage);路小标文案 legBadgeOf。
- app/src/main/kotlin/top/yeij/cyrene/ui/chat/TurnStreams.kt
  StreamingLeg(bucket, StreamingState) 与 withLeg / beginLeg / finishingLeg / withoutLeg;
  六态在状态上的推导 ChatUiState.phaseOf(消息 / 乐观气泡);
  路小标与「跳到提问」的查表 legBadgeForMessage / legBadgeForTurn / jumpIndexForTurn。

修改
- data/api/dto/MessageDto.kt:HistoryMessageDto 增可选 turnId(历史行 / 落库行都用它)。
- data/api/dto/ChatDto.kt:ChatRequest 增可选 parallel(REST 与 WS chat 同形)。
- data/ws/WsOutbound.kt:chat 增可选 parallel、interrupt 增可选 turnId,省略即整键不进线上。
- data/ws/WsEvent.kt:新增 turn_begin / turn.end / steer_ack 三个事件与其载荷;
  session_busy / session_idle / pending_drain 增可选 activeTurns,session_interrupted 增
  activeTurns + cancelledTurns,pending_ack 增可选 reason。
- ui/chat/ChatViewModel.kt:流式解析器与「本回合见过结构化正文帧」由单值升为按桶的 map/set,
  新增「本会话最近见过的 turnId」锚点;消费 turn_begin / steer_ack / turn.end;
  助手落库行按其 turnId 收口灰态并只摘掉它自己那一路的流式气泡;排队回执记进 queuedKeys。
- ui/chat/MessageBubble.kt:MessageBubble / PendingBubble 增 phase 与 legBadge,
  正文气泡增灰态配色,新增并行路小标(可点即跳);角标文案抽成纯函数 bubbleFooter。
- ui/chat/ChatScreen.kt:按桶取每一路的流式状态并渲染;传入六态与小标;忙碌行加「并行 N 路」。
- 新增 4 个测试文件(共 82 条):ParallelTurnsTest、TurnStreamsTest、
  WsEventTurnFramesTest、WsOutboundParallelTest。

二、为什么这么做

- 界面必须按 (sessionId, turnId) 两级分桶(规范 §1.7 第 1 件硬要求):两路并行时两个不同
  turnId 的 message.delta / thinking.delta / tool_call 会交错到达,照老代码「一律累加到当前那
  一个流式气泡」会把两条回复糊成一条(规范 §1.4 点名的那件事)。现在每一路各有一份解析器、
  一份思维链、一份工具卡表、一个流式气泡。
- 灰态(④ 已读、未答)的唯一权威收口事件是 turn.end(规范 §1.2)。本端另外在 stream_end
  (既有的「整轮结束」帧,v0.36 起带 turnId)与「同 turnId 的助手落库行」上也收口:
  前者在 turn.end 到位之前是唯一能收口的帧,后者是规范给的兜底。三者顺序不定、重复到达都无害
  (纯函数幂等)。
- 六态判定与收口判据的不变量都抽成纯函数,可直接单测;判定顺序即证据强度:
  被拒 > 已绑定未收口 > 已绑定已收口 > 本地乐观气泡 > 排队回执。

三、字段缺失时的降级保证(本次最要紧的一条)

- 全链路没有一处「turnId 为空就不渲染」。turnBucketOf 在帧壳没给 turnId 时先回落本会话最近
  见过的 turnId、再回落匿名桶(turnId 为空串)⇒ 所有无 turnId 的帧落在同一个桶里,
  那就是老代码那唯一一个流式气泡。
- turn_begin / steer_ack.turnId / turn.end 不来 ⇒ 一条绑定都建不起来 ⇒ 六态里只可能出现
  ①②③⑤,灰态根本不会进入;界面与改前逐字相同。
- activeTurns 缺席时不动这个计数(null 是「不知道」,0 是「确实空闲」,两件事不许互相顶替);
  pending_ack.reason 缺席就是 null,绝不伪造。
- 上行三个新字段(chat.parallel、interrupt.turnId、ChatRequest.parallel)本端今天一个字节都
  不发:真并行要用户显式声明,而声明入口不在本次范围内。服务端照旧走排队 + 回合内 steer。
- 已被单测钉住的证据:
  ParallelTurnsTest「所有无 turnId 的帧落进同一个桶」「两个空白 turnId 不算匹配,绝不收口」;
  TurnStreamsTest「单路时长度恒为 1」「没有任何 turn 帧时已落库=②、本地气泡=①」;
  WsEventTurnFramesTest「老形状的帧解出来新键一律为 null(不是 0、不是空串)」;
  WsOutboundParallelTest「省略时那条帧逐字节相同,且不带该键」。

四、等 v0.65 的部分

- turn_begin / steer_ack.turnId / turn.end 的真实到达,历史行的可选 turnId,activeTurns /
  pending_ack.reason / cancelledTurns 的真实取值。
- 真并行的触发入口(谁在什么时候把 chat.parallel 置 true)——本次只备好线上形状,不接 UI。
- 仍未决且影响客户端的两项:graph_update 两路互相顶掉(规范未决 10,等 design/19 裁决);
  按 turnId 单独核算用量(规范未决 14,可选)。

五、规范里与本端既有定案冲突的一处(如实登记)

规范 §1.1 ⑥「被拒」要求气泡带红色「未送达 · 会话正忙」并可重试;本端被拒(pending_full /
定向 chat_busy)的既有定案是摘掉乐观气泡 + 横幅如实告知 + 正文还给输入栏(可再按一次发送,
幂等键保证不会落成第二条消息)。既有行为由现有测试钉住,本次不改;六态判定里保留了 Rejected
档并由单测直接覆盖,只是本端的现实路径走的是回滚那一支。

验证:assembleDebug 绿 + testDebugUnitTest 绿(0 失败 0 错误 0 跳过)。
This commit is contained in:
AskaEth committed 2026-10-05 09:10:45 +08:00
1 parent 5e6dc76ae8
commit 29299b190e
13 files changed
+2124 -116

No files matched your search

@@ -46,6 +46,15 @@ data class ChatRequest(
* WS `chat` 帧的 4096 字节读上限仍在,且上行帧里没有承载附件的余地。
*/
val documents: List<DocumentRefDto> = emptyList(),
/**
* **请求把本条作为独立一路处理**(忙时并行规范 §2.2,REST 与 WS `chat` **同形**)。
*
* `null` = **整键省略**(`encodeDefaults = false`),与加入该字段之前**逐字节相同**。
* 本端今天**不发它**:真并行要用户显式声明,而声明入口不在本次范围内(客户端只备好形状)。
* 服务端照准与否都不报错——额度满时落回排队并回 `202 {queued:true}`(伴一条
* `pending_ack{reason:"parallel_limit"}`),已被在飞路 steer 吃进时回 `200 {stream:true}`。
*/
val parallel: Boolean? = null,
)
/**
@@ -43,6 +43,22 @@ data class HistoryMessageDto(
*/
val documents: List<DocumentRefDto> = emptyList(),
val createdAt: Long,
/**
* 产出 / 吃进这条消息的那一路的回合标识(忙时并行规范 §2.7,**纯增量**)。
*
* 语义按角色分(规范 §2.7 明写):
* - **user 行** = 吃进它 / 起它的那一路的 `turnId`(被 steer 吃进的行 = 吃进它的那一轮);
* - **assistant 行** = 产出它的那一路的 `turnId`。
*
* 于是重连后能把「回复 ↔ 提问」精确配对,两条并行回复也天然分属两个 `turnId`(分桶有据),
* **不必靠时间戳猜**(历史顺序的真源是 `(created_at, rowid)` = **完成先后**,
* 不承诺与提问相邻)。
*
* **旧行整键省略**(不是空串,规范 §2.7 末条)⇒ [null] 时客户端回落既有的「按顺序渲染」,
* 与加入这一列之前逐字相同。四处同源:`message_persisted` / `GET /messages` / 握手 `history`
* 回放 / `channel_persisted`(本端不消费最后一条)。
*/
val turnId: String? = null,
)
/**
@@ -227,12 +227,89 @@ sealed class WsEvent {
override val type: String get() = "stream_end"
}
/**
* **新帧 `turn_begin`**(忙时并行规范 §2.4):编排器**每次启动一路**时发一次,
* 把「服务端刚起的这一路 ↔ 它归属的用户消息 id」钉死。
*
* 客户端靠它(而不是靠猜)做三件事:① 按 `messageIds` 把那些气泡置为**灰态**(④);
* ② 建立「消息 ↔ 回合」绑定,好在收口时用该 `turnId` 的 `turn.end` 配对;③ 按 `turnId` 分桶。
*
* [kind] ∈ `"turn"` \| `"parallel"`:服务端**实际**如何起这条路(新回合 / 续跑 = `"turn"`,
* 声明并行且照准 = `"parallel"`)。**仅供展示**,[messageIds] 才是判据——规范 §2.4 明写
* 「客户端的状态机不必依赖它」,本端因此**不**用它做任何分支。
*
* [leg] 是该路的会话内序号(1 基,`1` = 唯一 / 第一路),界面上的「并行路 N」标就取它。
* [activeTurns] 是**本次启动后**该会话的在途路数。
*
* 帧壳的 `sessionId` 是权威归属(`data` 内不重复,规范 §2.4);本端仍按既有口径
* 「`data` 有值就优先,否则用帧壳」解析,两边都不给时按「无法判定」放行(见 [isForSession])。
*
* **帧壳 `turnId` 是这一路的唯一权威标识**(规范附录明写:它必须与该路全部流帧同值)。
* 它缺席(`null`)时本端**不做任何绑定**——没有 key 的绑定永远收不了口,那不是绑定,
* 是一个永远灰着的气泡;此时整帧只消费 [activeTurns],其余部分与今天一致。
*/
data class TurnBegin(
override val sessionId: String,
val kind: String = "",
val messageIds: List<String> = emptyList(),
val leg: Int = 0,
val activeTurns: Int = 0,
override val turnId: String? = null,
override val seq: Long? = null,
) : WsEvent() {
override val type: String get() = "turn_begin"
}
/**
* **`turn.end`**:一路的**整轮**结束(覆盖最终答案 / 工具循环终止 / 预算耗尽 / 出错四种出口)。
*
* 它是灰态(④)收口的**唯一权威事件**(规范 §1.2):拿帧壳的 [turnId] 与消息绑定的 `turnId`
* 比对,相等即收口。规范特意说明为什么不能用正文到齐来收口——一轮里有多个正文段(`message.end`
* 只结束**一段**),只有 `turn.end` 结束**整轮**。
*
* 它由 loop 发出、经网关的流式广播路径下发(`broadcastStream` 会给它打上 `sessionId` + `turnId`),
* 因此**一直存在于线上**,只是本端此前把它当未知帧丢掉了(`type` 不在 `when` 里 ⇒ `parse` 回
* `null`)。补上它的解析不改变任何既有行为(此前没有任何消费者)。
*/
data class TurnEnd(
override val sessionId: String? = null,
override val turnId: String? = null,
override val seq: Long? = null,
) : WsEvent() {
override val type: String get() = "turn.end"
}
/**
* `steer_ack`:这条消息**已被本回合吃进**(已从 pending 移除、尚未被正面回答)⇒ 灰态(④)。
*
* 契约 v0.57 起 [messageId] / [atStep] 就在,**一字未变**;忙时并行规范 §2.5 **只新增帧壳
* `turnId`**——并行下它是**必需**字段:没有它就不知道这条消息被**哪一路**吃进,灰态无法与
* 正确的 `turn.end` 配对(这正是「两路并行」相对「单路」多出来的唯一一件事)。
*
* 帧壳 `turnId` 缺席时本端**不消费这一帧**(老服务端就是这样)——那样绑出来的绑定永远收不了口,
* 宁可保持今天的样子(什么都不发生),也不留下一个永远灰着的气泡。
*/
data class SteerAck(
override val sessionId: String,
val messageId: String,
val atStep: Int,
override val turnId: String? = null,
override val seq: Long? = null,
) : WsEvent() {
override val type: String get() = "steer_ack"
}
/** 会话进入忙碌态。`mode` ∈ `idle` / `thinking` / `executing` / `subagent`。 */
data class SessionBusy(
override val sessionId: String,
val mode: String,
override val turnId: String? = null,
override val seq: Long? = null,
/**
* 该帧发出时的**在途路数**(忙时并行规范 §1.5 / §2.6,**纯增量**)。
* `null` = 服务端没给这个键(v0.65 之前)⇒ 客户端**不动**任何计数,行为与今天逐字相同。
*/
val activeTurns: Int? = null,
) : WsEvent() {
override val type: String get() = "session_busy"
}
@@ -244,6 +321,12 @@ sealed class WsEvent {
val queueDepth: Int,
override val turnId: String? = null,
override val seq: Long? = null,
/**
* 入队的原因码(忙时并行规范 §2.6,**纯增量**)。已知取值 `"parallel_limit"`
* (声明了并行但额度已满、已落回排队)。其它情况**整键缺席**(`null`)——
* 服务端明写「绝不伪造」,所以这里也**绝不**给它一个默认值。
*/
val reason: String? = null,
) : WsEvent() {
override val type: String get() = "pending_ack"
}
@@ -255,16 +338,26 @@ sealed class WsEvent {
val queueDepth: Int,
override val turnId: String? = null,
override val seq: Long? = null,
/** 见 [SessionBusy.activeTurns](忙时并行规范 §2.6)。 */
val activeTurns: Int? = null,
) : WsEvent() {
override val type: String get() = "pending_drain"
}
/** 会话回到空闲、队列清空。 */
/**
* 会话回到空闲、队列清空。
*
* ⚠️ **时点在并行时收紧**(规范 §1.5):它现在只在**所有在途路都收尾**且 pending 空时才发;
* 单路时与今天逐字相同。所以「`session_idle` 到了 = 这个会话真的闲了」这条既有推论仍然成立,
* 客户端**不必**改任何既有逻辑。
*/
data class SessionIdle(
override val sessionId: String,
val mode: String,
override val turnId: String? = null,
override val seq: Long? = null,
/** 见 [SessionBusy.activeTurns](忙时并行规范 §2.6)。 */
val activeTurns: Int? = null,
) : WsEvent() {
override val type: String get() = "session_idle"
}
@@ -284,6 +377,13 @@ sealed class WsEvent {
val dropped: Int,
override val turnId: String? = null,
override val seq: Long? = null,
/** 见 [SessionBusy.activeTurns](忙时并行规范 §2.6)。 */
val activeTurns: Int? = null,
/**
* 本次**真的取消了几路**(忙时并行规范 §2.6,**纯增量**)。
* 单路时它只会是 `0` 或 `1`;`null` = 服务端没给这个键。
*/
val cancelledTurns: Int? = null,
) : WsEvent() {
override val type: String get() = "session_interrupted"
}
@@ -623,6 +723,10 @@ sealed class WsEvent {
"stream_chunk" -> StreamChunk(decode<StreamChunkData>(data).delta, turnId, seq, sessionId)
"stream_end" -> decode<StreamEndData>(data)
.let { StreamEnd(it.stopReason, it.steps, it.error, turnId, seq, sessionId) }
// `turn.end` 的载荷是**空的**(后端 `streamEventData` 里 `StreamKindTurnEnd` 直接返回
// `nil`),全部信息都在帧壳上(`sessionId` + `turnId`)——所以这里不 decode payload,
// 也**不**因为 `data` 缺席而把它当坏帧(那正是「灰态收不了口」的成因)。
"turn.end" -> TurnEnd(sessionId, turnId, seq)
else -> null
}
@@ -634,15 +738,63 @@ sealed class WsEvent {
data: JsonElement?,
): WsEvent? = when (type) {
"session_busy" -> decode<SessionModeData>(data)
.let { SessionBusy(resolveSessionId(sessionId, it.sessionId), it.mode, turnId, seq) }
.let {
SessionBusy(
resolveSessionId(sessionId, it.sessionId), it.mode, turnId, seq, it.activeTurns,
)
}
"pending_ack" -> decode<PendingData>(data)
.let { PendingAck(resolveSessionId(sessionId, it.sessionId), it.mode, it.queueDepth, turnId, seq) }
.let {
PendingAck(
resolveSessionId(sessionId, it.sessionId), it.mode, it.queueDepth, turnId, seq, it.reason,
)
}
"pending_drain" -> decode<PendingData>(data)
.let { PendingDrain(resolveSessionId(sessionId, it.sessionId), it.mode, it.queueDepth, turnId, seq) }
.let {
PendingDrain(
resolveSessionId(sessionId, it.sessionId), it.mode, it.queueDepth, turnId, seq, it.activeTurns,
)
}
"session_idle" -> decode<SessionModeData>(data)
.let { SessionIdle(resolveSessionId(sessionId, it.sessionId), it.mode, turnId, seq) }
.let {
SessionIdle(
resolveSessionId(sessionId, it.sessionId), it.mode, turnId, seq, it.activeTurns,
)
}
"session_interrupted" -> decode<InterruptedData>(data)
.let { SessionInterrupted(resolveSessionId(sessionId, it.sessionId), it.mode, it.dropped, turnId, seq) }
.let {
SessionInterrupted(
resolveSessionId(sessionId, it.sessionId),
it.mode,
it.dropped,
turnId,
seq,
it.activeTurns,
it.cancelledTurns,
)
}
// 新帧 `turn_begin`(忙时并行规范 §2.4):帧壳带 sessionId + turnId,载荷给
// kind / messageIds / leg / activeTurns(**不含** sessionId)。
"turn_begin" -> decode<TurnBeginData>(data)
.let {
TurnBegin(
resolveSessionId(sessionId, it.sessionId),
it.kind,
it.messageIds.filter(String::isNotBlank),
it.leg,
it.activeTurns,
turnId,
seq,
)
}
// `steer_ack`(v0.57 形状 + 忙时并行规范 §2.5 的帧壳 turnId):载荷只有
// messageId / atStep;会话归属两边都可能有(取有值者)。
"steer_ack" -> decode<SteerAckData>(data)
.let {
SteerAck(
resolveSessionId(sessionId, it.sessionId), it.messageId, it.atStep, turnId, seq,
)
}
"pending_full" -> decode<PendingFullData>(data).let {
PendingFull(
resolveSessionId(sessionId, it.sessionId),
@@ -787,13 +939,55 @@ private data class StreamEndData(
)
@Serializable
private data class SessionModeData(val sessionId: String = "", val mode: String = "")
private data class SessionModeData(
val sessionId: String = "",
val mode: String = "",
/**
* `session_busy` / `session_idle` 的**可选**在途路数(忙时并行规范 §2.6)。
* `null` = 服务端没给这个键(v0.65 之前)——**不是 0**:0 是「确实没有在途路」,
* 缺席是「不知道」,两者绝不能互相顶替(拿缺席当 0 会把并行两路误报成空闲)。
*/
val activeTurns: Int? = null,
)
@Serializable
private data class PendingData(val sessionId: String = "", val mode: String = "", val queueDepth: Int = 0)
private data class PendingData(
val sessionId: String = "",
val mode: String = "",
val queueDepth: Int = 0,
/** 见 [WsEvent.SessionBusy.activeTurns](`pending_ack` / `pending_drain` 各带一份)。 */
val activeTurns: Int? = null,
/** 入队原因码(规范 §2.6,目前只有 `"parallel_limit"`);缺席 = `null`,绝不伪造。 */
val reason: String? = null,
)
@Serializable
private data class InterruptedData(val sessionId: String = "", val mode: String = "", val dropped: Int = 0)
private data class InterruptedData(
val sessionId: String = "",
val mode: String = "",
val dropped: Int = 0,
val activeTurns: Int? = null,
val cancelledTurns: Int? = null,
)
/**
* `turn_begin` 载荷(忙时并行规范 §2.4):`{kind, messageIds, leg, activeTurns}`。
*
* `sessionId` 在规范里**不出现在载荷中**(只在帧壳);这里仍留一个可选键并让它优先——与
* [WsEvent.resolveSessionId] 对其它帧的既有口径一致(帧壳缺席时服务端若给了,用它比丢掉好)。
*/
@Serializable
private data class TurnBeginData(
val sessionId: String = "",
val kind: String = "",
val messageIds: List<String> = emptyList(),
val leg: Int = 0,
val activeTurns: Int = 0,
)
/** `steer_ack` 载荷(v0.57 一字未变):`{messageId, atStep}`;`sessionId` 由帧壳给。 */
@Serializable
private data class SteerAckData(val sessionId: String = "", val messageId: String = "", val atStep: Int = 0)
/**
* `pending_full` 载荷:`{sessionId, mode, max, clientMsgId?, messageId?}`。前三个键来自加入
@@ -1,6 +1,7 @@
package top.yeij.cyrene.data.ws
import kotlinx.serialization.json.JsonElement
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
@@ -32,7 +33,7 @@ sealed interface WsOutbound {
data()?.let { put("data", it) }
}.toString()
/** `chat`:`{sessionId?, text, image?, deviceId?, clientMsgId?}`;text+image 至少一项。 */
/** `chat`:`{sessionId?, text, image?, deviceId?, clientMsgId?, parallel?}`;text+image 至少一项。 */
data class Chat(
val text: String,
val sessionId: String? = null,
@@ -43,6 +44,17 @@ sealed interface WsOutbound {
* 重复到达服务端回下行帧 `chat_duplicate`、**不落库、不起回合**。省略 = 老行为逐字不变。
*/
val clientMsgId: String? = null,
/**
* **请求把本条作为独立一路处理**(忙时并行规范 §2.2,**纯增量**)。
*
* `null` = **整键省略**:行为与加入该字段之前**逐字相同**(忙时落回排队 + 回合内 steer)。
* 本端**今天一个字节都不发它**——真并行需要用户显式声明,而声明入口(UI)不在本次范围内;
* 这里只把线上形状备好,好让「接上它」是一次纯增量而不是一次协议改动。
*
* 服务端**照准也不报错**(额度满 ⇒ 落回排队并回 `pending_ack{reason:"parallel_limit"}`;
* 已被在飞路 steer 吃进 ⇒ 回 `steer_ack`),所以这个字段永远不该引来一个错误帧。
*/
val parallel: Boolean? = null,
override val reqId: String? = null,
) : WsOutbound {
override val type: String get() = "chat"
@@ -53,10 +65,13 @@ sealed interface WsOutbound {
image?.let { put("image", it) }
deviceId?.let { put("deviceId", it) }
clientMsgId?.let { put("clientMsgId", it) }
// 布尔字面量走 JsonPrimitive:`put(key, Boolean)` 的重载在不同 kotlinx 版本里形态不一,
// 显式构造不会有歧义,也保证 `null` 时**整键省略**(与后端 `omitempty` 同义)。
parallel?.let { put("parallel", JsonPrimitive(it)) }
}
}
/** `interrupt`:`{sessionId, scope?}`。命中空闲会话为 no-op。 */
/** `interrupt`:`{sessionId, scope?, turnId?}`。命中空闲会话为 no-op。 */
data class Interrupt(
val sessionId: String,
/**
@@ -67,6 +82,18 @@ sealed interface WsOutbound {
* 不做取值校验、不静默回退成默认(服务端的开放取值域由服务端裁决)。
*/
val scope: String? = null,
/**
* 只取消**这一路**(忙时并行规范 §2.3,**纯增量**)。
*
* 含义按 [scope] 分(规范的表):`"current"` + 给了 `turnId` ⇒ **只停该路**,其余路与 pending
* 不受影响;`"current"` + 省略 ⇒ 取消**全部在途路**、保留 pending(现状语义的自然推广);
* 省略 / `""` / `"all"` ⇒ 忽略本字段(取消全部 + 清 pending)。
*
* `null` = **整键省略** ⇒ 与加入该字段之前**逐字节相同**。本端今天同样一个字节都不发它:
* 唯一的调用点(`ChatViewModel` 的「她正忙时按发送」偏好)走的正是「取消全部在途路」那一档,
* 不需要按路精确停(规范未决问题 12 的建议默认:默认目标 = 全部在途路)。
*/
val turnId: String? = null,
override val reqId: String? = null,
) : WsOutbound {
override val type: String get() = "interrupt"
@@ -74,6 +101,7 @@ sealed interface WsOutbound {
override fun data(): JsonElement = buildJsonObject {
put("sessionId", sessionId)
scope?.let { put("scope", it) }
turnId?.let { put("turnId", it) }
}
}
@@ -0,0 +1,268 @@
package top.yeij.cyrene.domain.chat
/**
* 忙时并行回合(多路)的客户端模型:**分桶、六态、灰态收口**三件纯逻辑。
*
* 规格来源:`Cyrene_v2/DevDocs/_specs/忙时并行-规范草案.md`(§1 行为规范 / §2 接口规范 / §4 主人裁决)。
* 本文件**只有纯数据与纯函数**:不碰协程、不碰 Android、不碰网络,所以三层逻辑都能直接单测。
* 消费方是 `ChatViewModel`(接线)与 `ChatScreen` / `MessageBubble`(渲染)。
*
* ## 为什么要分桶
* 两路并行时,同一 `sessionId` 下**两个不同 `turnId`** 的 `message.delta` / `thinking.delta` /
* `tool_call.*` 会**交错到达**。若照老代码那样「一律累加到当前那一个流式气泡」,两条回复会被糊成
* 一条(规范 §1.4 明写这一点)。所以落桶键是 `(sessionId, turnId)` 两级(规范 §1.7 第 1 条)。
*
* ## 字段缺失必须能降级(本文件的头号纪律)
* `turnId` 在真后端 v0.36 起就打在**每个结构化流帧**的帧壳上,但本文件**不假设**它一定在:
* 帧壳没给(`null` / 空白)时,帧落回**本会话最近见过的那条路**([turnBucketOf] 的 `sessionTurnId`),
* 一条都还没见过时落进**匿名桶**(`turnId == ""`)。于是「所有帧都不带 `turnId`」时全部帧落进
* **同一个桶**——也就是**今天那一个流式气泡**,行为逐字相同。
* 同理 `turn_begin` / `steer_ack` 不来时,六态里只会出现 ①②③⑤,灰态(④)根本不会进入。
*/
/** 「路序号未知」:`steer_ack` 只给 `turnId`、不给 `leg`(规范 §2.5 的载荷只有 messageId / atStep)。 */
const val UNKNOWN_LEG: Int = 0
/**
* 流式分桶键:`sessionId` → `turnId` 两级(规范 §1.4 / §1.7 第 1 条)。
*
* [turnId] 为空串 = **匿名桶**:帧壳没给 `turnId`、且本会话还没见过任何 `turnId` 时的落点。
* 它是「老的单路行为」在分桶模型里的等价物——所有匿名帧落在这同一个桶里,与今天逐字相同。
*/
data class TurnBucket(val sessionId: String, val turnId: String) {
/** 是否匿名桶(帧壳没给 `turnId`)。 */
val isAnonymous: Boolean get() = turnId.isEmpty()
override fun toString(): String = if (isAnonymous) "$sessionId/*" else "$sessionId/$turnId"
}
/**
* 帧该落哪个桶(纯函数)。
*
* 优先级:**帧壳给的 `turnId`** > **本会话最近见过的 `turnId`** > **匿名桶**。
*
* 第二条是「字段缺失 ⇒ 行为与今天逐字相同」那条硬要求的落地处:帧壳没给 `turnId` 时,帧跟着
* 本会话当前那一路走——老代码里它本来就会落进那**唯一**一个流式气泡。因此
* `turnBucketOf("web", null, null)` 恒等于匿名桶,全部帧都落在一处,与分桶之前完全一致。
*
* @param frameTurnId 帧壳上的 `turnId`(可能 `null` / 空白)。
* @param sessionTurnId 本会话最近一次见到的**非空** `turnId`(由调用方记住并喂进来)。
*/
fun turnBucketOf(sessionId: String, frameTurnId: String?, sessionTurnId: String?): TurnBucket =
TurnBucket(sessionId, frameTurnId.orNullIfBlank() ?: sessionTurnId.orNullIfBlank() ?: "")
/**
* **收口帧**(`stream_end` / `turn.end` / 落库的助手行)该落哪个桶(纯函数)。
*
* 与 [turnBucketOf] 的区别只有一条、但很要紧:**收口帧不许新开桶**。
*
* 理由是实证的:老代码里 `stream_end` 无论带着哪个 `turnId`,都会把**唯一**那个流式气泡收口
* (`active = false`)。若照 [turnBucketOf] 给一个认不出来的 `turnId` 开新桶,那个新桶收口、
* **真正在跑的那个桶永远收不了口**——光标一直转、`stream_end` 这一帧等于白给。
* 所以这里的顺序是:
*
* 1. 帧壳的 `turnId` 认得出来(在 [existing] 里)⇒ 就是它;
* 2. 否则本会话最近见过的 `turnId` 认得出来 ⇒ 用它;
* 3. 否则 [existing] 里还有桶 ⇒ 用**最近更新的那一个**(单路场景下它就是正确的那一个);
* 4. 一个桶都没有 ⇒ 匿名桶(也不新开,调用方据此可以对「没有在跑的路」整帧跳过)。
*
* @param existing 当前**已经存在**的桶的 `turnId` 列表,**顺序即更新时间**(最后的最近)。
*/
fun closingBucketOf(
sessionId: String,
frameTurnId: String?,
sessionTurnId: String?,
existing: List<String>,
): TurnBucket {
val frame = frameTurnId.orNullIfBlank()
val session = sessionTurnId.orNullIfBlank()
val target = when {
frame != null && existing.contains(frame) -> frame
session != null && existing.contains(session) -> session
existing.isEmpty() -> ""
else -> existing.last()
}
return TurnBucket(sessionId, target)
}
/**
* 一条消息在「忙时并行」语义下可能处的**六态**(规范 §1.1)。
*
* 名字逐字对齐规范表格里的态名,避免两套叫法:
* - [Sending] ① 发送中(本地乐观气泡);
* - [Sent] ② 已送达(正常气泡,还没进任何一路);
* - [Queued] ③ 排队中(气泡角标「排队中」**浅灰**——规范 §1.3 明写:**`queued` 不是灰态**,
* `pending_ack` 不许渲染成灰态,`queued` 只靠 `turn_begin` **直接换态**);
* - [Answering] ④ 已受理·回答中(**灰态「已读、未答」**);
* - [Answered] ⑤ 已答(终态);
* - [Rejected] ⑥ 被拒(终态,用户可手动重发)。
*/
enum class TurnPhase { Sending, Sent, Queued, Answering, Answered, Rejected }
/**
* 六态判定(纯函数,规范 §1.1 的表格逐行对照)。
*
* 判据只有四个,且**顺序即证据强度**(与 `queueFullOutcome` 同一条纪律):
* 1. [rejected] = 收到过明确拒绝(定向 `error{chat_busy}` / 广播 `pending_full`)⇒ ⑥。
* 它压过一切:这条消息**根本没进去**,不许再拿「有没有被接手」去描述它;
* 2. [binding] 非空且**未收口** ⇒ ④ 灰态(`turn_begin` 的 `messageIds` 含它,或 `steer_ack` 绑上它);
* 3. [binding] 非空且**已收口** ⇒ ⑤;
* 4. [binding] 为空且是本地乐观气泡 ⇒ [queued] ? ③ : ①;
* 5. [binding] 为空且已落库 ⇒ [queued] ? ③ : ②。
*
* 第 4/5 条里 [queued] 与非 [queued] 分开,是因为服务端的 `pending_ack` **不带消息身份**
* (规范 §1.1 ③ 的进入条件明写「现状帧,形状不变」)——客户端只能按「这条我发出去之后收到了入队
* 回执」来认,认不出来时为 [Sent] / [Sending](今天的样子),**绝不瞎猜成排队**。
*
* @param local 这条是不是**本地乐观气泡**(还没落库)。
* @param binding 这条消息被哪一路接手(`null` = 没被任何一路接手)。
* @param queued 收到过「它进队列」的回执(`pending_ack` / REST `202 queued`)。
* @param rejected 收到过明确的拒绝。
*/
fun turnPhaseOf(
local: Boolean,
binding: TurnBinding?,
queued: Boolean,
rejected: Boolean,
): TurnPhase = when {
rejected -> TurnPhase.Rejected
binding != null && !binding.closed -> TurnPhase.Answering
binding != null -> TurnPhase.Answered
local -> if (queued) TurnPhase.Queued else TurnPhase.Sending
queued -> TurnPhase.Queued
else -> TurnPhase.Sent
}
/**
* 灰态(④)**收口判据**(纯函数,规范 §1.2)。
*
* 规范给的是**唯一口径**:收口事件 = 该消息所属回合的 `turn.end`,用 `turn.end` **帧壳的 `turnId`**
* 与这条消息绑定的 `turnId` 比对,**相等即收口**;此外该 `turnId` 的助手 `message_persisted`
* 同样收口(两者顺序不定,任一到即收口)。
*
* 所以这里就一件事:两个 id **都非空且相等**。任一为空 ⇒ `false`(**不许**用「都为空所以相等」
* 收口——那会让「没有 turnId 的老客户端 / 老服务端」上的消息被随便一帧误收口)。
*
* @param boundTurnId 这条消息绑定的 `turnId`。
* @param closingTurnId 收口帧带来的 `turnId`(`turn.end` 帧壳 / 助手落库行的 `turnId`)。
*/
fun closesAnswering(boundTurnId: String?, closingTurnId: String?): Boolean {
val bound = boundTurnId.orNullIfBlank() ?: return false
val closing = closingTurnId.orNullIfBlank() ?: return false
return bound == closing
}
/**
* 消息 ↔ 回合的绑定:**一路**(规范 §2.4 的 `turn_begin` 说的是「编排器每次启动一路发一次」)。
*
* 它由两件事建立:`turn_begin`(一次给全 `messageIds` / `leg` / `activeTurns`)与 `steer_ack`
* (只补「某条消息被哪一路吃进」,没有 `leg`)。
* 只在 `turnId` **非空**时才会被建立——`turnId` 是这张表的键,也是收口的唯一凭据(见
* [closesAnswering]);没有它建出来的绑定永远收不了口,那不是绑定,是一个永远灰着的气泡。
*/
data class TurnBinding(
/** 这一路的权威回合标识(与该路全部流帧的 `turnId` 同值)。 */
val turnId: String,
/** 该路的会话内序号([UNKNOWN_LEG] = `steer_ack` 建的表,服务端没告诉序号)。 */
val leg: Int = UNKNOWN_LEG,
/** 归属这一路的**用户消息 id**(`turn_begin.messageIds` / `steer_ack.messageId`)。 */
val messageIds: List<String> = emptyList(),
/** 服务端最近一次报的该会话在途路数(`turn_begin.activeTurns`;0 = 没报过)。 */
val activeTurns: Int = 0,
/** 该路是否已收口(`turn.end` / 同 `turnId` 的助手落库行)。 */
val closed: Boolean = false,
)
/** 这条消息被哪一路接手;`null` = 没被任何一路接手(也是「按顺序渲染」那档的老行为)。 */
fun List<TurnBinding>.bindingOfMessage(messageId: String): TurnBinding? =
firstOrNull { binding -> messageId.isNotBlank() && binding.messageIds.contains(messageId) }
/**
* 收下一条 `turn_begin`(纯函数,返回新表)。
*
* 同 [TurnBinding.turnId] 已存在时**合并**而不是新增:`messageIds` 取并集(保持原顺序在前),
* `leg` / `activeTurns` 用新报的值,**`closed` 保持原值**——一次重复到达的 `turn_begin` 不该把一条
* 已经收口的路「重新打开」(那会让气泡回到灰态并且再也等不到第二次收口)。
* 表里没有它时按**追加**处理:两路的顺序就是它们被启动的顺序。
*/
fun List<TurnBinding>.withTurnBegin(
turnId: String,
leg: Int,
messageIds: List<String>,
activeTurns: Int,
): List<TurnBinding> {
val key = turnId.orNullIfBlank() ?: return this
val index = indexOfFirst { it.turnId == key }
if (index < 0) {
return this + TurnBinding(key, leg, messageIds.filter(String::isNotBlank), activeTurns)
}
val old = this[index]
val merged = old.copy(
leg = leg.takeIf { it > UNKNOWN_LEG } ?: old.leg,
messageIds = (old.messageIds + messageIds.filter(String::isNotBlank)).distinct(),
activeTurns = activeTurns.coerceAtLeast(0),
)
return toMutableList().also { it[index] = merged }
}
/**
* 收下一条 `steer_ack`(纯函数,返回新表):把 [messageId] 绑到 [turnId] 这一路上——这就是
* 「这条消息被在飞的那一轮吃进去了,它进灰态」的凭据(规范 §2.5)。
*
* 该 `turnId` 还没有表项时**建一个只有这一条消息的表**(`steer_ack` 不给 `leg` / `activeTurns`,
* 所以那两项留 [UNKNOWN_LEG] / 0)。`turnId` 或 `messageId` 空白 ⇒ **原样返回**:
* 绑不出东西来(见 [TurnBinding] 的 KDoc)。
*/
fun List<TurnBinding>.withSteeredMessage(turnId: String, messageId: String): List<TurnBinding> {
val key = turnId.orNullIfBlank() ?: return this
val message = messageId.orNullIfBlank() ?: return this
val index = indexOfFirst { it.turnId == key }
if (index < 0) return this + TurnBinding(key, UNKNOWN_LEG, listOf(message))
val old = this[index]
if (old.messageIds.contains(message)) return this
return toMutableList().also { it[index] = old.copy(messageIds = old.messageIds + message) }
}
/**
* 收口(纯函数,返回新表):把 [closingTurnId] 那一路上所有消息的绑定标记为 `closed`。
*
* 判据就是 [closesAnswering](两个 id 都非空且相等),因此一个空白 / 认不出来的 `closingTurnId`
* **一条都不收口**——这正是老服务端(没有 `turnId`)上的行为:什么都不发生。
* 没有命中任何一条时**原样返回同一个实例**(Compose 的稳定性与测试里的 `assertSame` 靠它)。
*/
fun List<TurnBinding>.closingTurn(closingTurnId: String?): List<TurnBinding> {
var changed = false
val next = map { binding ->
if (!binding.closed && closesAnswering(binding.turnId, closingTurnId)) {
changed = true
binding.copy(closed = true)
} else {
binding
}
}
return if (changed) next else this
}
/**
* 把**全部**在途路收口(纯函数):`session_idle` 的规范语义是「所有在途路都已收尾且 pending 空」
* (规范 §1.5),所以那一帧到达时**不可能**还有没收口的路——本函数就是这条语义的兜底落地。
* 它是兜底而不是主路径:主路径仍是各自 `turnId` 的 `turn.end`(规范 §1.2)。
*/
fun List<TurnBinding>.allClosed(): List<TurnBinding> {
if (none { !it.closed }) return this
return map { if (it.closed) it else it.copy(closed = true) }
}
/**
* 路标文案(规范 §1.4 A 的「小标」):`leg >= 2` 才成句,其余返回 `null`。
*
* **为什么 1 路不挂标**:单路是今天的常态(`parallel:true` 是客户端显式声明才有的能力,
* 规范 §0 的默认行为一条),给唯一一路也挂个「并行路 1」等于给每条回复都加一个新的视觉噪音,
* 却什么都说明不了。所以「有没有小标」本身就精确表达「这里是不是真的有多路」。
* `leg` 未知([UNKNOWN_LEG])同理不挂——不能把「不知道第几路」说成「第 0 路」。
*/
fun legBadgeOf(leg: Int): String? = if (leg >= 2) "并行路 $leg" else null
/** 空白串等同于缺席(服务端是「整键缺席」,但真给空串时按同一件事处理,与既有各纯函数一致)。 */
private fun String?.orNullIfBlank(): String? = this?.takeIf { it.isNotBlank() }
@@ -448,6 +448,10 @@ fun ChatScreen(
// 当前 baseUrl(判据与取舍全在 ImageHostRebase 里,这里只喂地址)。
// 只在组合期读一次:换服务器必然换会话/重进页面,不需要额外订阅。
imageBaseUrl = container.authManager.currentBaseUrl,
// 六态(规范 §1.1)在**气泡上**的落地:③ 排队中 / ④ 灰态各有
// 自己的外观,①②⑤ 与今天逐字相同。判定是纯函数,接线层不自己猜。
phase = state.phaseOf(item.message),
legBadge = state.legBadgeForMessage(item.message),
onDocumentClick = { ref ->
previewNotice = null
previewTarget = DocumentTarget.of(ref)
@@ -466,6 +470,7 @@ fun ChatScreen(
is ChatDisplayItem.Pending -> PendingBubble(
item = item.item,
phase = state.phaseOf(item.item),
onDocumentClick = { draft ->
previewNotice = null
previewTarget = DocumentTarget.of(draft)
@@ -475,7 +480,23 @@ fun ChatScreen(
},
)
ChatDisplayItem.Streaming -> state.streaming?.let { StreamingBubble(it) }
// 每一路一条流式气泡(按 `(sessionId, turnId)` 两级分桶,规范 §1.4)。
// 拿不到这一路的状态说明它刚刚被收口掉了(同一帧里 `streams` 变了),
// 此时按「什么都不画」处理——不是渲染一个空气泡。
is ChatDisplayItem.Streaming -> state.streamOf(item.bucket)?.let { streaming ->
val jump = state.jumpIndexForTurn(item.bucket.turnId)
StreamingBubble(
streaming = streaming,
legBadge = state.legBadgeForTurn(item.bucket.turnId),
// 跳不到就不给入口(见 StreamingBubble 的 KDoc)。
onJumpToQuestion = jump?.let { index ->
{
searchNotice = "已跳到这一路的提问"
pendingScroll = index
}
},
)
}
}
}
if (state.hasMore) {
@@ -813,12 +834,18 @@ private fun connectionLabel(state: ChatUiState): String = when (state.connection
* `pending_ack`/`pending_drain` 与 REST `202 {queued:true,queueDepth}`([ChatUiState.queueDepth])。
* REST 的 202 只给得到条数、给不到忙碌通道,所以队列深度**单独也能成句**——否则 202 之后到
* `session_busy` 之前会有一小段什么都看不见的空档。
*
* 第四个来源是**在途路数**([ChatUiState.activeTurns],忙时并行规范 §1.5):`> 1` 时说一句
* 「并行 N 路」。**只有真的多路才说**——单路时 `activeTurns` 是 1(或服务端根本没给、状态里是 0),
* 这句话就不出现,于是 v0.65 之前这一行逐字不变(规范 §1.5 明写「不消费则行为不变」)。
*/
private fun busyLabel(state: ChatUiState): String = when {
state.busyMode != null && state.queueDepth > 0 -> "正在思考… 已排队 ${state.queueDepth} 条"
state.busyMode != null -> "正在思考…"
state.queueDepth > 0 -> "已排队 ${state.queueDepth} 条"
else -> ""
private fun busyLabel(state: ChatUiState): String {
val parallel = if (state.activeTurns > 1) "并行 ${state.activeTurns} 路" else null
return listOfNotNull(
state.busyMode?.let { "正在思考…" },
parallel,
state.queueDepth.takeIf { it > 0 }?.let { "已排队 $it 条" },
).joinToString(" ").ifBlank { "" }
}
/**
@@ -52,10 +52,18 @@ import top.yeij.cyrene.domain.chat.MessageFingerprint
import top.yeij.cyrene.domain.chat.RetryKeyMemory
import top.yeij.cyrene.domain.chat.SendChannel
import top.yeij.cyrene.domain.chat.SharedTextPlan
import top.yeij.cyrene.domain.chat.TurnBinding
import top.yeij.cyrene.domain.chat.TurnBucket
import top.yeij.cyrene.domain.chat.allClosed
import top.yeij.cyrene.domain.chat.closingBucketOf
import top.yeij.cyrene.domain.chat.closingTurn
import top.yeij.cyrene.domain.chat.editableMessageOf
import top.yeij.cyrene.domain.chat.interruptScopeForSend
import top.yeij.cyrene.domain.chat.readBounded
import top.yeij.cyrene.domain.chat.toDocumentFailure
import top.yeij.cyrene.domain.chat.turnBucketOf
import top.yeij.cyrene.domain.chat.withSteeredMessage
import top.yeij.cyrene.domain.chat.withTurnBegin
import top.yeij.cyrene.domain.parse.ParsedBubble
import top.yeij.cyrene.domain.parse.StreamingContentParser
import top.yeij.cyrene.domain.share.SharedContent
@@ -429,8 +437,15 @@ sealed interface ChatDisplayItem {
override val key: String get() = "p:${item.localId}"
}
data object Streaming : ChatDisplayItem {
override val key: String get() = "s"
/**
* 一路正在生长的助手回复。
*
* 它带**整把桶钥匙**(`sessionId` + `turnId`)而不是一个序号:key 必须在跨会话时也唯一
* (这个 VM 只服务一个会话,但 key 是列表项的稳定身份,不该依赖「当前只会有一个会话」这种
* 隐含前提),也必须在两路并行时区分得开(规范 §1.4:按 `turnId` 分桶是硬要求)。
*/
data class Streaming(val bucket: TurnBucket) : ChatDisplayItem {
override val key: String get() = "s:${bucket.sessionId}:${bucket.turnId}"
}
}
@@ -438,7 +453,17 @@ data class ChatUiState(
/** 已落库消息,升序(旧 → 新)。 */
val messages: List<HistoryMessageDto> = emptyList(),
val pending: List<PendingMessage> = emptyList(),
val streaming: StreamingState? = null,
/**
* 流式中的助手回复,**按 `(sessionId, turnId)` 两级分桶**(忙时并行规范 §1.4 / §1.7 第 1 条)。
*
* 从「一个气泡」变成「一串路」的原因只有一个:两路并行时两个不同 `turnId` 的
* `message.delta` / `thinking.delta` / `tool_call.*` 会**交错到达**,按「当前气泡」累加就会把
* 两条回复糊成一条。桶的纯粹逻辑在 `TurnStreams.kt`,这里只存结果。
*
* **单路时它恒为「长度 ≤ 1」**(每回合一个全新随机 `turnId`;帧壳不带 `turnId` 时全部帧落
* 匿名桶),因此对今天的行为没有可见差异。
*/
val streams: List<StreamingLeg> = emptyList(),
val draft: String = "",
/** 已选好、还没发出去的图片(本地 URI 字符串);null = 没有待发图片。 */
val draftImage: String? = null,
@@ -469,6 +494,30 @@ data class ChatUiState(
val connection: WsState = WsState.Idle,
/** 编排忙碌态:`idle`/`thinking`/`executing`/`subagent`;null = 空闲。 */
val busyMode: String? = null,
/**
* 该会话的**在途路数**(忙时并行规范 §1.5 / §2.6:`session_busy` / `session_idle` /
* `pending_drain` / `session_interrupted` 的可选 `activeTurns`,以及 `turn_begin.activeTurns`)。
*
* `0` = 不知道或确实没有在途路。**服务端不给这个键时不改它**(`null` 与 `0` 是两件事:
* 前者是「不知道」,后者是「确实空闲」),因此 v0.65 之前它恒为 `0`,界面一个像素都不变。
*/
val activeTurns: Int = 0,
/**
* 消息 ↔ 回合的绑定表(`turn_begin` / `steer_ack` 建立,`turn.end` / 助手落库行收口)。
*
* 这是灰态(④)的全部依据:**进灰态**靠它,**收口**也靠它(规范 §1.7 第 2 条)。
* 空表 = 一次 `turn_begin` / `steer_ack` 都没收到 = 今天的样子(没有灰态)。
*/
val turns: List<TurnBinding> = emptyList(),
/**
* 收到过「这条进队列了」回执的键:本地乐观气泡的 `localId`,以及落库后的服务端 `messageId`。
*
* 两种写法都要认,因为 `pending_ack`(广播、**不带消息身份**)与 `message_persisted`
* (带 `clientMsgId` = 本端 `localId`)的先后顺序不确定。它是六态里 ③ 排队中的唯一判据
* (规范 §1.1 ③ 的进入条件是「现状帧,形状不变」——服务端没说被拒的是哪一条,所以本端只按
* 「我发出去之后收到了入队回执」认,**不猜**)。
*/
val queuedKeys: Set<String> = emptySet(),
/**
* 待处理队列深度。两个来源:WS 的 `pending_ack` / `pending_drain`,
* 以及 REST `202 {queued:true,queueDepth:n}` 的受理回执(同一个队列,同一份语义)。
@@ -496,8 +545,11 @@ data class ChatUiState(
get() = buildList {
messages.forEach { add(ChatDisplayItem.Committed(it)) }
pending.forEach { add(ChatDisplayItem.Pending(it)) }
if (streaming != null) add(ChatDisplayItem.Streaming)
streams.forEach { add(ChatDisplayItem.Streaming(it.bucket)) }
}
/** 取某一路的流式状态(渲染层按 [ChatDisplayItem.Streaming] 的桶回查)。 */
fun streamOf(bucket: TurnBucket): StreamingState? = streams.leg(bucket)?.state
}
/**
@@ -554,10 +606,18 @@ class ChatViewModel(
private var socket: ChatSocket? = null
private var repository: MessageRepository? = null
private var oldestLoadedAt: Long? = null
private var streamParser: StreamingContentParser? = null
/**
* 本回合是否已经收到过**结构化正文帧**(`message.start` / `message.delta`)。
* **每一路**一个流式正文解析器(键与 `ChatUiState.streams` 同源)。
*
* 老代码只有一个 `streamParser`:两路并行的 delta 交替进来时,它会用第二路的正文**续写**第一路
* 的缓冲——两条回复糊成一条(规范 §1.4)。按桶分开之后,每一路各自累计自己的那段正文。
* 单路时这个 map 里恒只有一项,行为与原来逐字相同。
*/
private val legParsers = mutableMapOf<TurnBucket, StreamingContentParser>()
/**
* **每一路**「这一回合是否已经收到过结构化正文帧」(`message.start` / `message.delta`)。
*
* 只服务 [WsEvent.StreamChunk] 那一条兜底分支:契约 §3.1 的回退路径里 `delta` 是**整段** `res.Text`
* 而不是增量(`02` 回执十三 §1 后端确认),所以它**不能**与同回合的结构化帧叠加——两者都到,
@@ -571,7 +631,17 @@ class ChatViewModel(
* 归零晚一步的代价是**下一轮**的兜底帧被判成「与结构化帧并存」而整帧丢弃——那正是这条分支
* 存在的意义(未知帧不丢内容)的反面。
*/
private var structuredTextFramesThisTurn = false
private val legSawStructuredText = mutableSetOf<TurnBucket>()
/**
* 本会话**最近一次见到的非空 `turnId`**(帧壳给的)。
*
* 它是「字段缺失 ⇒ 行为与今天逐字相同」那道闸的锚点:帧壳没给 `turnId` 时,这一步的帧落回
* 本会话当前那一条路([turnBucketOf]),也就是老代码里它本来会落进的**唯一**那个气泡。
* 只在**开启一路**的帧上更新(`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*`),
* 收口帧不更新——收口帧若带着一个认不出来的 id,把锚点改成它只会让后续的无 id 帧落错地方。
*/
private var sessionTurnId: String? = null
/**
* 掉线那一刻的 `after` 游标(契约 §1 行 6):为 null = 当时没有需要补的洞。
@@ -1289,9 +1359,11 @@ class ChatViewModel(
// 202 {queued:true}:已**落库并续接排队**。轻量态交给既有忙碌标签(busyLabel 认
// queueDepth);队列清空时 WS 的 `pending_drain` / `session_idle` 会把它归零。
// 顺带按 `localId` 记下「这条在排队」——六态 ③ 的判据之一(与 WS 的 `pending_ack` 同源,
// 两者是同一个队列的两种到达方式)。REST 这条比广播帧精确:它**知道**是哪一条。
is ChatAck.Queued -> {
retryKey = retryKey.forget(localId)
_state.update { it.copy(queueDepth = ack.queueDepth) }
_state.update { it.copy(queueDepth = ack.queueDepth, queuedKeys = it.queuedKeys + localId) }
}
// 幂等重复:服务端没有重复落库、也没有起第二个回合——对齐既有消息,**不是错误**。
@@ -1484,22 +1556,18 @@ class ChatViewModel(
// —— 流式正文 ——
is WsEvent.MessageStart -> {
if (event.index == 0) {
streamParser = StreamingContentParser()
structuredTextFramesThisTurn = false
freshTurnIfFinished()
}
structuredTextFramesThisTurn = true
val parser = streamParser ?: StreamingContentParser().also { streamParser = it }
val bubbles = parser.onDelta("")
updateStreaming { it.copy(bubbles = bubbles, active = true) }
val bucket = openBucket(event.turnId)
if (event.index == 0) beginTurnBody(bucket)
legSawStructuredText += bucket
val bubbles = parserFor(bucket).onDelta("")
updateStreaming(bucket) { it.copy(bubbles = bubbles, active = true) }
}
is WsEvent.MessageDelta -> {
structuredTextFramesThisTurn = true
val parser = streamParser ?: StreamingContentParser().also { streamParser = it }
val bubbles = parser.onDelta(event.delta)
updateStreaming { it.copy(bubbles = bubbles, active = true) }
val bucket = openBucket(event.turnId)
legSawStructuredText += bucket
val bubbles = parserFor(bucket).onDelta(event.delta)
updateStreaming(bucket) { it.copy(bubbles = bubbles, active = true) }
}
/**
@@ -1515,7 +1583,7 @@ class ChatViewModel(
* 结构化帧并存。所以 W26 那版「照 [WsEvent.MessageDelta] 追加进同一个气泡」是**错的**:
* 它真到达时会把同一段内容**追加成重影**。现在按同一份回执给的口径分成两支:
*
* - 同一回合内**已经收到过结构化帧**([structuredTextFramesThisTurn])⇒ **整帧忽略**:
* - 同一回合内**已经收到过结构化帧**([legSawStructuredText])⇒ **整帧忽略**:
* 它是兜底,不能与结构化帧叠加;
* - 否则 ⇒ [StreamingContentParser.reset] 之后**整段喂进去**,即用这一整段**替换**当前
* 流式气泡的正文(不是追加)。多条兜底帧先后到达时后到的那段覆盖前一段——「整段」
@@ -1526,76 +1594,120 @@ class ChatViewModel(
* 那一轮的正文只有这一个出口)。删掉它,静默丢正文的风险面只会更大。
*
* ## 顺手补上的一处边界:回退路径没有 `message.start`
* [WsEvent.MessageStart] 的 `index == 0` 是既有代码里唯一会 [freshTurnIfFinished] 的地方,
* 而回退路径没有这一帧 ⇒ 上一轮的思维链 / 工具卡会串到这一轮来。后端确认回退路径**无
* 触发源**(所以这不是一个能复现的 bug),这里只是把这块补成与 `message.start` 逐字相同的
* 收口。它跑不到第二次:第一帧之后 `active` 已经是 true,「已结束才清」的判据自然不再成立。
* [WsEvent.MessageStart] 的 `index == 0` 是既有代码里唯一会收口「上一轮已结束的流式状态」
* 的地方(老代码的 `freshTurnIfFinished`),而回退路径没有这一帧 ⇒ 上一轮的思维链 / 工具卡
* 会串到这一轮来。后端确认回退路径**无触发源**(所以这不是一个能复现的 bug),这里只是把
* 这块补成与 `message.start` 逐字相同的收口。它跑不到第二次:第一帧之后 `active` 已经是
* true,「已结束才清」的判据自然不再成立。
*
* 会话归属照旧由 `onEvent` 开头那道闸收敛(本帧带 `sessionId`,见 [isForSession])。
*/
is WsEvent.StreamChunk -> {
if (!structuredTextFramesThisTurn) {
freshTurnIfFinished()
val parser = streamParser ?: StreamingContentParser().also { streamParser = it }
val bucket = openBucket(event.turnId)
if (!legSawStructuredText.contains(bucket)) {
beginTurnBody(bucket)
// reset 而不是追加:`delta` 是整段,见上面「它的语义是整段」一段。
val parser = parserFor(bucket)
parser.reset()
val bubbles = parser.onDelta(event.delta)
updateStreaming { it.copy(bubbles = bubbles, active = true) }
updateStreaming(bucket) { it.copy(bubbles = bubbles, active = true) }
}
}
// —— 思维链 ——
is WsEvent.ThinkingStart -> {
if (event.index == 0) freshTurnIfFinished()
updateStreaming { it.copy(thinkingActive = true) }
val bucket = openBucket(event.turnId)
if (event.index == 0) beginTurnBody(bucket)
updateStreaming(bucket) { it.copy(thinkingActive = true) }
}
is WsEvent.ThinkingDelta -> updateStreaming {
it.copy(thinking = it.thinking + event.delta, thinkingActive = true)
is WsEvent.ThinkingDelta -> openBucket(event.turnId).let { bucket ->
updateStreaming(bucket) { it.copy(thinking = it.thinking + event.delta, thinkingActive = true) }
}
is WsEvent.ThinkingEnd -> updateStreaming { it.copy(thinkingActive = false) }
is WsEvent.ThinkingEnd -> openBucket(event.turnId).let { bucket ->
updateStreaming(bucket) { it.copy(thinkingActive = false) }
}
// —— 工具调用 ——
is WsEvent.ToolCallStart -> updateStreaming { streaming ->
if (streaming.tools.any { it.id == event.id }) {
streaming
} else {
streaming.copy(tools = streaming.tools + ToolCallState(event.id, event.index))
// 这几帧只改**自己那一路**的工具卡表:两路并行时各自的工具批次互不可见(不变量 I3′)。
is WsEvent.ToolCallStart -> openBucket(event.turnId).let { bucket ->
updateStreaming(bucket) { streaming ->
if (streaming.tools.any { it.id == event.id }) {
streaming
} else {
streaming.copy(tools = streaming.tools + ToolCallState(event.id, event.index))
}
}
}
is WsEvent.ToolCallDelta -> updateStreaming { streaming ->
streaming.copy(
tools = streaming.tools.map {
if (it.id == event.id) it.copy(arguments = it.arguments + event.delta) else it
},
)
is WsEvent.ToolCallDelta -> openBucket(event.turnId).let { bucket ->
updateStreaming(bucket) { streaming ->
streaming.copy(
tools = streaming.tools.map {
if (it.id == event.id) it.copy(arguments = it.arguments + event.delta) else it
},
)
}
}
is WsEvent.ToolCallEnd -> updateStreaming { streaming ->
streaming.copy(
tools = streaming.tools.map { if (it.id == event.id) it.copy(done = true) else it },
)
is WsEvent.ToolCallEnd -> openBucket(event.turnId).let { bucket ->
updateStreaming(bucket) { streaming ->
streaming.copy(
tools = streaming.tools.map { if (it.id == event.id) it.copy(done = true) else it },
)
}
}
is WsEvent.ToolExecutionStart -> onToolExecution(event.id, event.tool, event.phase)
is WsEvent.ToolExecutionStart -> onToolExecution(event.turnId, event.id, event.tool, event.phase)
is WsEvent.ToolExecutionUpdate ->
onToolExecution(event.id, event.tool, event.phase, event.delta)
onToolExecution(event.turnId, event.id, event.tool, event.phase, event.delta)
is WsEvent.ToolExecutionEnd -> onToolExecution(event.id, event.tool, event.phase)
is WsEvent.ToolExecutionEnd -> onToolExecution(event.turnId, event.id, event.tool, event.phase)
is WsEvent.StreamEnd -> {
val bubbles = streamParser?.finish().orEmpty()
// 这一轮到此收口:下一轮的兜底帧要按「没见过结构化帧」重新判(见字段 KDoc)。
structuredTextFramesThisTurn = false
updateStreaming {
it.copy(bubbles = bubbles, thinkingActive = false, active = false)
val bucket = bucketToClose(event.turnId)
val opened = legParsers[bucket]
// 这一路到此收口:下一轮的兜底帧要按「没见过结构化帧」重新判(见字段 KDoc)。
legSawStructuredText -= bucket
legParsers -= bucket
_state.update { state ->
state.copy(
// 解析器不在手上(这一轮只到了工具帧 / 一帧正文都没有)时传 `null`:
// 只收口标志位,保留已经显示出来的正文段。
streams = state.streams.finishingLeg(bucket, opened?.finish()),
// 灰态收口(规范 §1.2):`turn.end` 是唯一权威口径,而 `stream_end` 是本端
// 既有的「整轮结束」帧、又带 `turnId`(v0.36 起),在 `turn.end` 到位之前
// 它是**唯一**能收口的事件。两者顺序不定、重复到达也无害(纯函数幂等)。
turns = state.turns.closingTurn(event.turnId),
)
}
}
is WsEvent.MessagePersisted -> onPersisted(event.message)
// `turn.end`:灰态收口的**权威**事件(规范 §1.2)——一轮里可能有多个正文段
// (`message.end` 只结束**一段**),只有它结束**整轮**(覆盖最终答案 / 工具循环终止 /
// 预算耗尽 / 出错四种出口)。本端只消费它的帧壳 `turnId`,不碰流式气泡:
// 气泡的收口归 `stream_end` 管(两者各司其职)。
is WsEvent.TurnEnd -> _state.update {
it.copy(turns = it.turns.closingTurn(event.turnId))
}
is WsEvent.MessagePersisted -> onPersisted(event.message, event.turnId)
// 灰态的**另一个入口**:这条消息被在飞的那一轮吃进去了(规范 §2.5)。
// 帧壳 `turnId` 缺席时**不消费这一帧**——那样绑出来的绑定永远收不了口,
// 宁可保持今天的样子(什么都不发生),也不留下一个永远灰着的气泡。
is WsEvent.SteerAck -> if (event.sessionId == sessionId) {
_state.update { state ->
state.copy(
turns = state.turns.withSteeredMessage(event.turnId.orEmpty(), event.messageId),
)
}
}
// 新帧 `turn_begin`:把「刚起的这一路 ↔ 它归属的用户消息」钉死(规范 §2.4)。
is WsEvent.TurnBegin -> if (event.sessionId == sessionId) onTurnBegin(event)
// D3:编辑 / 删除的实时增量,与本端自己发起时的 REST 回执**走同一条落地路径**
// ([applyMessageEdited] / [applyMessageDeleted]),所以「响应 + 广播」双到达
@@ -1606,15 +1718,28 @@ class ChatViewModel(
is WsEvent.MessageDeleted -> applyMessageDeleted(event.id)
is WsEvent.SessionBusy -> if (event.sessionId == sessionId) {
_state.update { it.copy(busyMode = event.mode) }
_state.update { it.copy(busyMode = event.mode).withActiveTurns(event.activeTurns) }
}
is WsEvent.PendingAck -> if (event.sessionId == sessionId) {
_state.update { it.copy(busyMode = event.mode, queueDepth = event.queueDepth) }
_state.update { state ->
// 「忙时这条入队了」——`pending_ack` 是**广播**且**不带消息身份**(规范 §1.1 ③
// 的进入条件明写「现状帧,形状不变」),所以只能把它认到**本会话当前待发的那几条**
// 乐观气泡上。一条待发气泡都没有时什么都不标记,绝不瞎猜。
//
// 这一帧**不带来** `activeTurns`(规范 §2.6 只把该计数挂在
// session_busy / session_idle / pending_drain / session_interrupted 上),
// 所以这里也不动它:少一个来源不等于「在途路数变成 0」。
state.copy(
busyMode = event.mode,
queueDepth = event.queueDepth,
queuedKeys = state.queuedKeys + state.pending.map { it.localId },
)
}
}
is WsEvent.PendingDrain -> if (event.sessionId == sessionId) {
_state.update { it.copy(queueDepth = event.queueDepth) }
_state.update { it.copy(queueDepth = event.queueDepth).withActiveTurns(event.activeTurns) }
}
// 幂等重复(契约 v0.56 §3.13):这条消息**早已受理过**,本端重发(弱网 / 切后台)撞上了
@@ -1624,11 +1749,23 @@ class ChatViewModel(
}
is WsEvent.SessionIdle -> if (event.sessionId == sessionId) {
_state.update { it.copy(busyMode = null, queueDepth = 0) }
// `session_idle` 的规范语义是「**所有**在途路都已收尾且 pending 空」(规范 §1.5)
// ⇒ 到这里不可能还有没收口的路。灰态**主路径**仍是各路自己的 `turn.end`,
// 这里只是兜底,且幂等(已收口的路不会被再改一次)。
_state.update {
it.copy(busyMode = null, queueDepth = 0, turns = it.turns.allClosed())
.withActiveTurns(event.activeTurns)
}
}
is WsEvent.SessionInterrupted -> if (event.sessionId == sessionId) {
_state.update { it.copy(busyMode = null, queueDepth = 0) }
// 被打断 ⇒ 在途路全部收尾。灰态**主路径**仍是各路自己的 `turn.end`(规范 §2.5.8:
// 被取消路若吃过 steer 消息,那些消息留在历史里、其 `turn.end` 仍会发出);
// 这里只是兜底,同样是幂等的。
_state.update {
it.copy(busyMode = null, queueDepth = 0, turns = it.turns.allClosed())
.withActiveTurns(event.activeTurns)
}
}
// 队列满(契约 v0.59/G5):广播帧,**每条**连接都收得到,所以键对不上就必须放手。
@@ -1682,26 +1819,109 @@ class ChatViewModel(
if (outcome is QueueFullOutcome.Adopt) ensureMessageKnown(outcome.messageId)
}
/** 新一轮开始时,若上一轮的流式状态已结束(active=false)就清掉,避免思维链/工具卡串台。 */
private fun freshTurnIfFinished() {
_state.update { state ->
val streaming = state.streaming
if (streaming != null && !streaming.active) state.copy(streaming = StreamingState()) else state
}
/**
* 算这一帧该落哪个桶,并顺势更新「本会话最近见过的 `turnId`」。
*
* 只给**开启 / 进行**这一类的帧用(`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*` /
* `stream_chunk`):它们的 `turnId` 就是这一路的身份。收口帧走 [bucketToClose]。
*/
private fun openBucket(frameTurnId: String?): TurnBucket {
frameTurnId?.takeIf { it.isNotBlank() }?.let { sessionTurnId = it }
return turnBucketOf(sessionId, frameTurnId, sessionTurnId)
}
private fun updateStreaming(block: (StreamingState) -> StreamingState) {
/**
* 收口帧该落哪个桶:**认不出来就落在最近更新的那个桶上,绝不新开一个**。
*
* 为什么这条不能省:老代码里 `stream_end` 无论带着什么 `turnId` 都会把**唯一**那个流式气泡收口。
* 若给它新开一个桶,那个新桶收口、真正在跑的那个桶永远收不了口——光标一直转、这一帧等于白给。
* 判据本身是纯函数([closingBucketOf]),这里只喂「当前已存在的桶」。
*/
private fun bucketToClose(frameTurnId: String?): TurnBucket =
closingBucketOf(sessionId, frameTurnId, sessionTurnId, streamsOf().map { it.bucket.turnId })
/** 当前已存在的桶(解析器表与界面上的路表取并集:两者正常同步,但收口帧不该被这个前提连累)。 */
private fun streamsOf(): List<StreamingLeg> {
val fromState = _state.value.streams
val missing = legParsers.keys.filter { key -> fromState.none { it.bucket == key } }
return fromState + missing.map { StreamingLeg(it, StreamingState()) }
}
/** 这一路的解析器(没有就建一个,与老代码 `?: StreamingContentParser().also { … }` 同义)。 */
private fun parserFor(bucket: TurnBucket): StreamingContentParser =
legParsers.getOrPut(bucket) { StreamingContentParser() }
/**
* 新一轮开始(`message.start` 的 `index == 0`,以及回退路径没有 `message.start` 时):
*
* 1. 这一路换一个**新解析器**(老代码在同一个位置 `streamParser = StreamingContentParser()`);
* 2. 清掉这一路的「见过结构化帧」判据;
* 3. 状态上丢掉**其它已经结束**的路(老代码的 `freshTurnIfFinished`)——还在跑的另一路不动。
*/
private fun beginTurnBody(bucket: TurnBucket) {
legParsers[bucket] = StreamingContentParser()
legSawStructuredText -= bucket
_state.update { it.copy(streams = it.streams.beginLeg(bucket)) }
}
private fun updateStreaming(bucket: TurnBucket, block: (StreamingState) -> StreamingState) {
_state.update { it.copy(streams = it.streams.withLeg(bucket, block)) }
}
/**
* 新帧 `turn_begin` 的处置(忙时并行规范 §2.4):把这一路与它归属的用户消息钉死。
*
* 两件事,各自独立降级:
* 1. **`activeTurns`** 只要给了就消费(忙徽标上的「并行 N 路」靠它);不给就不动。
* 2. **消息 ↔ 回合绑定**只在帧壳 `turnId` 非空时建立——`turnId` 是收口的唯一凭据,
* 没有它建出来的绑定**永远收不了口**(灰着不动的气泡比不灰更糟)。帧壳缺席时本端
* 只当这一帧没来过(= 今天的样子)。
*
* 顺带把被这一刻绑上的消息从 [ChatUiState.queuedKeys] 里摘掉:它们已经从「排队中」进了
* 「回答中」,留着只会让 ③ 的判据在未来某次重放里被重新读到(六态判定里绑定优先,所以这不是
* 正确性问题,纯粹是不留无用的状态)。
*/
private fun onTurnBegin(event: WsEvent.TurnBegin) {
_state.update { state ->
state.copy(streaming = block(state.streaming ?: StreamingState()))
val turnId = event.turnId?.takeIf { it.isNotBlank() }
val turns = if (turnId == null) {
state.turns
} else {
state.turns.withTurnBegin(turnId, event.leg, event.messageIds, event.activeTurns)
}
state.copy(
turns = turns,
queuedKeys = state.queuedKeys - event.messageIds.toSet(),
).withActiveTurns(event.activeTurns)
}
}
/**
* 把服务端报的 `activeTurns` 落到状态上(`null` = **没给这个键** ⇒ 一个字节都不动)。
*
* 「缺席」与「0」是两件事(见 [ChatUiState.activeTurns]):把缺席当 0 会在并行两路时把
* 忙徽标擦成空闲;把它当「不知道」才是如实的。
*/
private fun ChatUiState.withActiveTurns(reported: Int?): ChatUiState =
if (reported == null) this else copy(activeTurns = reported.coerceAtLeast(0))
/**
* 收口三帧 `tool_execution.*`:按 `id` 关联到对应的工具卡(关联规则见
* [toolExecutionTargetIndex]),落 phase / 工具名 / 局部输出。落不到就整帧忽略。
*
* 落哪张卡由**两件事**决定:先按 `turnId` 选**这一路**(两路并行时各自的工具批次互不可见,
* 不变量 I3′),再在那一路上按 `id` 关联。帧壳没给 `turnId` 时落回本会话当前那一路
* ([openBucket])——单路时这就是老代码里那唯一一份工具卡表。
*/
private fun onToolExecution(id: String?, tool: String, phase: String, delta: String? = null) {
updateStreaming { streaming ->
private fun onToolExecution(
frameTurnId: String?,
id: String?,
tool: String,
phase: String,
delta: String? = null,
) {
val bucket = openBucket(frameTurnId)
updateStreaming(bucket) { streaming ->
streaming.copy(tools = applyToolExecution(streaming.tools, id, tool, phase, delta))
}
}
@@ -1710,31 +1930,38 @@ class ChatViewModel(
* 落库消息到达:
* - user:把对应的乐观气泡去掉——**首选 `clientMsgId`**(v0.56 起服务端回填),
* 没有键时才回落到「同文本」匹配(见 [consumePendingFor])
* - assistant:用落库内容取代流式临时气泡
* - assistant:用落库内容取代流式临时气泡,并按它的 `turnId` **收口灰态**(规范 §1.2 的兜底)
*
* @param frameTurnId 帧壳上的 `turnId`。规范说助手落库行自己会带 `turnId`(§2.7),但那个键
* 要等 v0.65 才有 ⇒ 这里按「行上的 `turnId` 优先、帧壳兜底」取值,**两者都没有时按本会话
* 当前那一路**([bucketToClose]),于是老服务端上「落库取代流式气泡」这条既有行为逐字不变。
*/
private fun onPersisted(message: HistoryMessageDto?) {
private fun onPersisted(message: HistoryMessageDto?, frameTurnId: String?) {
val msg = message ?: return
if (msg.sessionId != sessionId) return
val clearStreaming = msg.role == "assistant"
// 流式结束:把这一轮的思维链挂到落库消息上,保留 10 分钟后再自动消失
val streamedThinking = if (clearStreaming) {
_state.value.streaming?.thinking?.takeIf { it.isNotBlank() }
} else {
null
}
// 流式结束:把这一轮的思维链挂到落库消息上,保留 10 分钟后再自动消失。
// 按**这一路**取(两路并行时第二路的思维链不该被第一路的落库行吃掉)。
val bucket = if (clearStreaming) bucketToClose(msg.turnId ?: frameTurnId) else null
val streamedThinking = bucket
?.let { _state.value.streamOf(it)?.thinking }
?.takeIf { it.isNotBlank() }
val persisted = if (streamedThinking != null) msg.copy(thinking = streamedThinking) else msg
if (clearStreaming) {
streamParser = null
// 落库正文已经取代了流式临时气泡,这一轮的兜底判据跟着一起收口(见字段 KDoc)。
structuredTextFramesThisTurn = false
if (clearStreaming && bucket != null) {
legParsers -= bucket
// 落库正文已经取代了流式临时气泡,这一路的兜底判据跟着一起收口(见字段 KDoc)。
legSawStructuredText -= bucket
}
_state.update { state ->
state.copy(
messages = mergeMessages(state.messages, listOf(persisted)),
pending = if (msg.role == "user") consumePendingFor(state.pending, msg) else state.pending,
streaming = if (clearStreaming) null else state.streaming,
streams = if (bucket == null) state.streams else state.streams.withoutLeg(bucket),
// 灰态收口(规范 §1.2 的兜底事件):助手回复落库 ⇒ 这条消息所属的那一路结束。
// 与 `turn.end` 顺序不定、重复到达也无害(纯函数幂等)。
turns = if (clearStreaming) state.turns.closingTurn(msg.turnId ?: frameTurnId) else state.turns,
)
}
@@ -39,6 +39,7 @@ import top.yeij.cyrene.domain.chat.DocumentChipModel
import top.yeij.cyrene.domain.chat.DocumentChips
import top.yeij.cyrene.domain.chat.DraftDocument
import top.yeij.cyrene.domain.chat.ImageHostRebase
import top.yeij.cyrene.domain.chat.TurnPhase
import top.yeij.cyrene.domain.enums.LocalEnumCatalog
import top.yeij.cyrene.domain.parse.ContentPart
import top.yeij.cyrene.domain.parse.ParsedBubble
@@ -127,6 +128,11 @@ fun isContentlessUserMessage(message: HistoryMessageDto): Boolean =
* @param imageBaseUrl 当前的服务器地址(`AuthManager.currentBaseUrl`)。只用来把历史图片里**绑死在
* 上传那一刻 host 上的同源绝对地址**换到现在的 host(`02/F6①`:`POST /upload` 的 `url` 按请求时的
* `r.Host` 拼、且原样落库 ⇒ 换地址后旧图取不回)。空串 = 不替换([ImageHostRebase] 会原样返回)。
* @param phase 这条消息在**太忙时并行六态**里的哪一态(见 [ChatUiState.phaseOf] / 规范 §1.1)。
* 只有 ③ 排队中与 ④ 回答中改变外观,①②⑤ 一律按今天的样子画(所以老服务端上逐字不变)。
* @param legBadge 「并行 N 路」小标(`leg >= 2` 才有,见 [legBadgeOf])。挂在来源角标那一行里:
* 规范 §1.4 A 要的就是「同一路的两条消息在界面上看得出来是一路」——提问侧挂同一个标,
* 回复侧在 [StreamingBubble] 上挂一次,用户就能把它们配起来。
* @param onEdit 用户选中「编辑」:交给上层打开编辑弹层(本组件不自己发请求)。
* @param onDelete 用户选中「删除」:交给上层打开删除确认(不可恢复,必须二次确认)。
*/
@@ -134,6 +140,8 @@ fun isContentlessUserMessage(message: HistoryMessageDto): Boolean =
fun MessageBubble(
message: HistoryMessageDto,
imageBaseUrl: String = "",
phase: TurnPhase = TurnPhase.Sent,
legBadge: String? = null,
onDocumentClick: (DocumentRefDto) -> Unit = {},
onImageClick: (String) -> Unit = {},
onEdit: (HistoryMessageDto) -> Unit = {},
@@ -155,10 +163,13 @@ fun MessageBubble(
// 每个气泡各自记住菜单开着没有:重建 / 滚动复用不会把上一个气泡的菜单带过来。
var menuOpen by remember(message.id) { mutableStateOf(false) }
val openMenu = if (canAct) { { menuOpen = true } } else null
// 灰态(规范 §1.1 ④「已读、未答」):正文区灰 + 光标指示。判据只认 `Answering` 一档,
// 所以 ③ 排队中**不会**被渲染成灰态——规范 §1.3 特意点名不要合并这两态。
val gray = phase == TurnPhase.Answering
MessageGroup(
isUser = isUser,
footer = messageSourceLabel(message, LocalEnumCatalog.current),
footer = bubbleFooter(messageSourceLabel(message, LocalEnumCatalog.current), phase, legBadge),
onOpenMenu = openMenu,
) {
message.thinking?.takeIf { it.isNotBlank() }?.let { ThinkingCard(it) }
@@ -174,7 +185,14 @@ fun MessageBubble(
ContentlessNote(onLongPress = openMenu)
} else {
bubbles.forEachIndexed { index, bubble ->
Bubble(bubble, isUser, isLast = index == bubbles.lastIndex, onLongPress = openMenu)
Bubble(
bubble = bubble,
isUser = isUser,
isLast = index == bubbles.lastIndex,
showCursor = gray,
onLongPress = openMenu,
gray = gray,
)
}
}
}
@@ -196,6 +214,33 @@ fun MessageBubble(
}
}
/**
* 气泡下方那一行角标的文案(纯函数,可单测)。
*
* 三段内容各自独立,用「·」连起来:**并行路小标**(多路时才在)→ **六态说明**(只有 ③④ 成句)
* → **来源角标**(既有规则,见 [messageSourceLabel])。
*
* 为什么 ③④ 要在这里说话:它们**都不是错误、也都不是「已送达」**,而气泡本身在两种态下都长得像
* 一条正常消息。不说一句,用户就分不清「她还没开始理我」与「已经在回答我了」——
* 规范 §1.1 特意把这两态分开,界面上就必须能分开。
*
* [sourceLabel] 由调用方从组合里取(`messageSourceLabel(message, LocalEnumCatalog.current)`):
* 那个 composition local **只能在可组合函数里读**,所以这里收进来当参数,好让本函数保持纯粹、
* 能被单测直接钉死。
*/
fun bubbleFooter(sourceLabel: String?, phase: TurnPhase, legBadge: String?): String? {
val parts = buildList {
legBadge?.let { add(it) }
when (phase) {
TurnPhase.Queued -> add("排队中")
TurnPhase.Answering -> add("已读,正在回答…")
else -> Unit
}
sourceLabel?.let { add(it) }
}
return parts.takeIf { it.isNotEmpty() }?.joinToString(" · ")
}
/**
* 乐观发送中的用户消息(还没落库)。
*
@@ -203,14 +248,22 @@ fun MessageBubble(
* 也不必把几 MB 的字符串塞进 UI 状态。落库后由 `message.image` 接管。
* 附件同理显示**本地显示名**(那时还没有服务端存储名,也还没有任何服务端结论)。
* 这张本地图同样可点进全屏查看([onImageClick]):读的是本地那一份字节。
*
* [phase] 只会是 ① 发送中或 ③ 排队中(乐观气泡还没有服务端 id,绑不到任何一路——
* 见 [ChatUiState.phaseOf])。③ 的角标文案按规范 §1.1 用「排队中」,语气与「发送中…」并列而不
* 与「已读」混同:**排队 ≠ 被受理**(规范 §1.3)。
*/
@Composable
fun PendingBubble(
item: PendingMessage,
phase: TurnPhase = TurnPhase.Sending,
onDocumentClick: (DraftDocument) -> Unit = {},
onImageClick: (String) -> Unit = {},
) {
MessageGroup(isUser = true, footer = "发送中…") {
MessageGroup(
isUser = true,
footer = if (phase == TurnPhase.Queued) "排队中" else "发送中…",
) {
DocumentChipRow(
chips = item.documents.map(DocumentChips::of),
alignEnd = true,
@@ -332,10 +385,28 @@ private fun DocumentChip(model: DocumentChipModel, onRemove: (() -> Unit)?, onCl
}
}
/** 正在流式的助手回复:思维链 → 工具卡 → 气泡(末尾带光标)。 */
/**
* 正在流式的助手回复:思维链 → 工具卡 → 气泡(末尾带光标)。
*
* ## 并行路小标(规范 §1.4 A / 主人裁决 ①)
* [legBadge] 非空时在最上方挂一枚小标(「并行路 2」),点它跳到**这一路归属的那条提问**
* ([onJumpToQuestion])。这就是规范给的单列形态:回复仍按历史顺序先后成条,靠小标而不是靠分栏
* 表达「这是另起的一路」。单路(`leg < 2`)时 [legBadge] 为 `null`,**一个像素都不多画**——
* 挂标这件事本身就是「这里有真正的多路」的信号([legBadgeOf] 的 KDoc)。
*
* 跳不到(那条提问不在已加载的窗口里、或这一路没绑消息)时**不显示「跳到提问」**:一个按下去
* 什么都不发生的入口比没有入口更糟。[onJumpToQuestion] 传 `null` 即这一档。
*/
@Composable
fun StreamingBubble(streaming: StreamingState) {
fun StreamingBubble(
streaming: StreamingState,
legBadge: String? = null,
onJumpToQuestion: (() -> Unit)? = null,
) {
MessageGroup(isUser = false) {
if (legBadge != null) {
LegBadge(text = legBadge, onClick = onJumpToQuestion)
}
if (streaming.thinking.isNotBlank() || streaming.thinkingActive) {
ThinkingCard(text = streaming.thinking, active = streaming.thinkingActive)
}
@@ -412,13 +483,40 @@ private fun Bubble(
isLast: Boolean,
showCursor: Boolean = false,
onLongPress: (() -> Unit)? = null,
gray: Boolean = false,
) {
AnnotationLine(bubble.head)
val body = if (showCursor && isLast && bubble.body.isNotEmpty()) bubble.body + CURSOR else bubble.body
if (body.isNotEmpty()) BubbleBody(body, isUser, onLongPress)
if (body.isNotEmpty()) BubbleBody(body, isUser, onLongPress, gray)
AnnotationLine(bubble.tail)
}
/**
* 并行路小标(规范 §1.4 A 的「小标」):一枚圆角小药丸,可点即跳。
*
* 配色刻意用 `primary` 文字 + `surfaceContainerHigh` 底:它既不是正文(不是内容),也不像来源角标
* 那样只是脚注——它是**一个入口**。规范把「点击可跳到对应提问」列为推荐形态,所以能跳时它必须
* 看得出可点([onClick] 非空才带涟漪与手势)。
*/
@Composable
private fun LegBadge(text: String, onClick: (() -> Unit)?) {
val shape = RoundedCornerShape(10.dp)
Row(
modifier = Modifier
.clip(shape)
.background(MiuixTheme.colorScheme.surfaceContainerHigh)
.then(if (onClick == null) Modifier else Modifier.clickable(onClick = onClick))
.padding(horizontal = 8.dp, vertical = 4.dp),
verticalAlignment = Alignment.CenterVertically,
) {
Text(
text = if (onClick == null) text else "$text · 跳到提问",
style = MiuixTheme.textStyles.footnote2,
color = MiuixTheme.colorScheme.primary,
)
}
}
/** 旁白/动作:灰斜体小字,依附在气泡上/下。 */
@Composable
private fun AnnotationLine(parts: List<ContentPart>) {
@@ -477,8 +575,22 @@ private fun ContentlessNote(onLongPress: (() -> Unit)? = null) {
}
}
/**
* 气泡本体。
*
* [gray] = **灰态**(规范 §1.1 ④ 的「已读、未答」):底色一律退回中性的
* `surfaceContainer`、文字退回 `onSurfaceVariantSummary`,**两种角色都画成同一种灰**——
* 这一态的语义是「这条已经在被处理,但还没有答案」,它既不是用户说的话(不该是主人的青色),
* 也不是她的回答(不该是助手的正常底色)。正文一个字都不改,只是**褪色**,让人一眼看出
* 「没丢,只是在等」。
*/
@Composable
private fun BubbleBody(text: String, isUser: Boolean, onLongPress: (() -> Unit)? = null) {
private fun BubbleBody(
text: String,
isUser: Boolean,
onLongPress: (() -> Unit)? = null,
gray: Boolean = false,
) {
Box(
modifier = Modifier
.widthIn(max = 300.dp)
@@ -486,10 +598,10 @@ private fun BubbleBody(text: String, isUser: Boolean, onLongPress: (() -> Unit)?
// 多出来的这一层是为了让长按的涟漪被圆角裁住(否则它会溢出到方角外)。
.clip(BubbleShape)
.background(
color = if (isUser) {
MiuixTheme.colorScheme.primary
} else {
MiuixTheme.colorScheme.surfaceContainer
color = when {
gray -> MiuixTheme.colorScheme.surfaceContainer
isUser -> MiuixTheme.colorScheme.primary
else -> MiuixTheme.colorScheme.surfaceContainer
},
)
.then(
@@ -511,7 +623,11 @@ private fun BubbleBody(text: String, isUser: Boolean, onLongPress: (() -> Unit)?
Text(
text = text,
style = MiuixTheme.textStyles.main,
color = if (isUser) MiuixTheme.colorScheme.onPrimary else MiuixTheme.colorScheme.onSurfaceContainer,
color = when {
gray -> MiuixTheme.colorScheme.onSurfaceVariantSummary
isUser -> MiuixTheme.colorScheme.onPrimary
else -> MiuixTheme.colorScheme.onSurfaceContainer
},
)
}
}
@@ -0,0 +1,180 @@
package top.yeij.cyrene.ui.chat
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
import top.yeij.cyrene.domain.chat.TurnBucket
import top.yeij.cyrene.domain.chat.TurnPhase
import top.yeij.cyrene.domain.chat.bindingOfMessage
import top.yeij.cyrene.domain.chat.legBadgeOf
import top.yeij.cyrene.domain.chat.turnPhaseOf
import top.yeij.cyrene.domain.parse.ParsedBubble
/**
* 流式气泡的**两级分桶**(规范 §1.4 / §1.7 第 1 条)在整个 UI 层的那一半。
*
* 这一层只回答两个问题,且都做成纯函数(`ChatViewModel` 与测试共用,渲染层也直接读它):
* 1. **一帧该改哪一路的流式气泡**([withLeg] / [beginLeg] / [finishingLeg] / [withoutLeg]);
* 2. **一条消息现在处在六态里的哪一态**([ChatUiState.phaseOf])。
*
* 老代码里只有一个 `streaming: StreamingState?`,所有 `message.delta` 都往那一个气泡上堆——
* 两路并行时两条回复会被糊成一条。这里的每条 [StreamingLeg] 各带自己的 [TurnBucket],
* 于是「同一会话、不同 turnId」的增量天然分开。
*
* ## 单路时为什么逐字不变
* 桶按 `turnId` 分,而真后端每个回合的 `turnId` 都是**新建的随机值**(网关 `newTurnID()`)——
* 单路时任何时刻**只有一个**活着的桶,等价于「只有那一个气泡」;帧壳没给 `turnId` 时全部帧落进
* 匿名桶([top.yeij.cyrene.domain.chat.turnBucketOf]),同样是**只有一个**桶。所以本文件对
* 「服务端 v0.65 之前」的每一个现实路径都只是把原来那一个气泡换了个容器。
*/
/**
* 一路正在生长的助手回复。
*
* [bucket] 是它的身份(`sessionId` + `turnId`),[state] 就是老代码里那个 `StreamingState`——
* 思维链 / 工具卡 / 正文段一个字都没改,只是现在**一路一份**。
*/
data class StreamingLeg(val bucket: TurnBucket, val state: StreamingState)
/** 取某一路(没有则 `null`)。 */
fun List<StreamingLeg>.leg(bucket: TurnBucket): StreamingLeg? = firstOrNull { it.bucket == bucket }
/**
* 改某一路:**有则改、无则新建并追加到末尾**(顺序即两路的启动顺序,界面上先起的在上)。
*
* 新建时以 [StreamingState] 的默认值为起点,所以「一路的第一帧」与老代码里
* `state.streaming ?: StreamingState()` 是同一条线。
*/
fun List<StreamingLeg>.withLeg(
bucket: TurnBucket,
block: (StreamingState) -> StreamingState,
): List<StreamingLeg> {
val index = indexOfFirst { it.bucket == bucket }
if (index < 0) return this + StreamingLeg(bucket, block(StreamingState()))
return toMutableList().also { it[index] = it[index].copy(state = block(it[index].state)) }
}
/**
* 新一轮开始(`message.start` 且 `index == 0`)该对桶表做什么——**等价于老代码的
* `freshTurnIfFinished()` + 新解析器**那两步里的状态那一半:
*
* 1. **丢掉其它已经结束的路**:老代码在「上一轮已 `active = false`」时会把它整个清成新的
* `StreamingState()`(内容已被 `message_persisted` 取代,留着就是重影)。按路分桶之后,
* 「已经结束」要按**路**判:[StreamingState.active] 为 `false` 的路丢掉,
* **还在跑的另一路一个字节都不动**——那正是并行两路各自独立生长所要求的。
* 2. **这一路如果是「已结束」状态被重新用到**(同一 `turnId` 又开了新的正文段)⇒ 重置成空态,
* 与老代码同一条线;仍然 `active` 的路保持思维链 / 工具卡,只换正文(老代码也是如此)。
*
* 这一路还不在表里时**原样返回**:新的桶由紧接着的 [withLeg] 建(不在这里凭空塞一个空态进去,
* 那会让「收到过收口帧但没收到过任何正文帧」的坏时序多出一个看不见的空气泡)。
*/
fun List<StreamingLeg>.beginLeg(bucket: TurnBucket): List<StreamingLeg> {
val kept = filter { it.bucket == bucket || it.state.active }
if (kept.none { it.bucket == bucket }) return kept
return kept.map {
if (it.bucket == bucket && !it.state.active) it.copy(state = StreamingState()) else it
}
}
/**
* 收口一路(`stream_end` 之类):把这一路标成「不再生长」。
*
* [bubbles] 是收口那一刻的正文段(解析器 `finish()` 的结果);**`null` = 这一路手上没有解析器**
* (例如这一轮一帧正文都没到、或只到了工具帧),此时保留它已有的正文段,只收口那几个标志位——
* 「没有新正文」不该被写成「正文是空的」(那会把已经显示出来的字擦掉)。
*
* **这一路不在表里时什么都不做**(`this` 原样返回)——收口帧不许**新开**一路(见
* [top.yeij.cyrene.domain.chat.closingBucketOf])。老代码在这里会顺手创建一个空的
* `StreamingState(active = false)`,那在界面上是一个不渲染任何东西、却会占一个列表项的幽灵;
* 不创建它,用户看得到的每一个像素都与今天相同。
*/
fun List<StreamingLeg>.finishingLeg(
bucket: TurnBucket,
bubbles: List<ParsedBubble>?,
): List<StreamingLeg> {
val index = indexOfFirst { it.bucket == bucket }
if (index < 0) return this
return toMutableList().also {
val state = it[index].state
it[index] = it[index].copy(
state = state.copy(
bubbles = bubbles ?: state.bubbles,
thinkingActive = false,
active = false,
),
)
}
}
/** 丢掉一路(它的助手回复已经落库,流式临时气泡该让位了)。 */
fun List<StreamingLeg>.withoutLeg(bucket: TurnBucket): List<StreamingLeg> =
if (none { it.bucket == bucket }) this else filterNot { it.bucket == bucket }
/**
* 这条**已落库**消息处在哪一态(六态判定,规范 §1.1)。
*
* 三个输入都取自状态本身:绑定的路([ChatUiState.turns])、有没有排队的回执
* ([ChatUiState.queuedKeys],按服务端 `messageId` 或本地幂等键 `clientMsgId` 两种写法各查一次——
* 队列回执是**先**到还是 `message_persisted` **先**到并不确定)。
*
* `rejected` 恒 `false` 是刻意的、也是如实的:本端被拒(`pending_full` / 定向 `chat_busy`)的处置是
* **摘掉乐观气泡 + 横幅如实告知 + 正文还给输入栏**(既有定案,见 `rollbackPending`),
* 那条消息根本不在消息列表里,**没有一个气泡可以挂红标**。
* 六态判定保留 `Rejected` 档(规范 §1.1 ⑥)并由单测直接覆盖,只是本端的现实路径走的是回滚那一支。
*/
fun ChatUiState.phaseOf(message: HistoryMessageDto): TurnPhase = turnPhaseOf(
local = false,
binding = turns.bindingOfMessage(message.id),
queued = isQueuedMarked(message.id) || isQueuedMarked(message.clientMsgId),
rejected = false,
)
/**
* 这条**乐观气泡**处在哪一态(六态判定)。
*
* 乐观气泡还没有服务端 id,所以绑定的路**永远查不到**(`turn_begin` 的 `messageIds` 是服务端 id)
* ——它要么是 ① 发送中,要么(收到过入队回执)是 ③ 排队中。这正是规范 §1.1 里 ① / ③ 的
* 那两种视觉:半透明「发送中…」与浅灰「排队中」。
*/
fun ChatUiState.phaseOf(item: PendingMessage): TurnPhase = turnPhaseOf(
local = true,
binding = null,
queued = isQueuedMarked(item.localId),
rejected = false,
)
/**
* 这条消息在界面上该挂什么小标(规范 §1.4 A:`leg >= 2` 才挂「并行 N 路」)。
*
* 两个方向都要认,因为「同一路的两条消息」有两头:
* - **提问侧**(user 行):靠 `turn_begin.messageIds` / `steer_ack` 建立的**绑定**;
* - **回复侧**(assistant 行):靠历史行上那个**可选** `turnId`(规范 §2.7,v0.65 起才有)。
*
* 老服务端上两条都查不到(没有 `turn_begin`,历史行的 `turnId` 整键省略)⇒ 恒 `null` ⇒
* 一个标都不挂,与今天逐字相同。
*/
fun ChatUiState.legBadgeForMessage(message: HistoryMessageDto): String? {
turns.bindingOfMessage(message.id)?.let { return legBadgeOf(it.leg) }
val producedBy = message.turnId?.takeIf { it.isNotBlank() } ?: return null
return turns.firstOrNull { it.turnId == producedBy }?.let { legBadgeOf(it.leg) }
}
/** 这一路的小标(按 `turnId` 查;`null` = 一路 / 序号未知,两种都不挂标)。 */
fun ChatUiState.legBadgeForTurn(turnId: String): String? =
turns.firstOrNull { it.turnId == turnId }?.let { legBadgeOf(it.leg) }
/**
* 从某一路跳到**它归属的那条提问**:返回聊天列表(`reverseLayout`)里的项下标;
* `null` = 定位不了(这条提问不在已加载的窗口里、或这一路没有绑到任何消息)。
*
* 下标算法与服务端搜索的定位**同一套**([planSearchJump]):`items` 的顺序是
* 「已落库消息 → 待发气泡 → 流式路」,所以「懒列表下标 = items.size - 1 - 升序下标」对已落库
* 消息恒成立。规范 §1.4 A 把「点小标跳到对应提问」列为推荐形态——它就是这条。
*/
fun ChatUiState.jumpIndexForTurn(turnId: String): Int? {
val binding = turns.firstOrNull { it.turnId == turnId } ?: return null
val index = messages.indexOfFirst { binding.messageIds.contains(it.id) }
if (index < 0) return null
return (items.size - 1 - index).coerceAtLeast(0)
}
private fun ChatUiState.isQueuedMarked(key: String?): Boolean =
!key.isNullOrBlank() && queuedKeys.contains(key)
@@ -0,0 +1,224 @@
package top.yeij.cyrene.data.ws
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Test
/**
* 忙时并行的**下行帧**:新帧 `turn_begin`、`turn.end`、`steer_ack` 的帧壳 `turnId`,
* 以及忙态帧的可选计数与 `pending_ack.reason`(规范 §2.4 ~ §2.7)。
*
* 两组断言各占一半:
* - **新键到位**(并行真的开起来时客户端有东西可用);
* - **老帧逐字不变**(v0.64 形状的帧解出来,新键一律是 `null`——`null` 是「没给」,
* 不是 0、不是空串;消费方据此一个字节都不动)。
*/
class WsEventTurnFramesTest {
// —— 新帧 turn_begin(规范 §2.4)——
@Test
fun `turn begin is parsed with every field of the spec`() {
val frame = """
{"type":"turn_begin","sessionId":"web","turnId":"turn-9f3c1a2b3c4d5e6f",
"data":{"kind":"parallel","messageIds":["msg-7f2a19c4e5"],"leg":2,"activeTurns":2}}
""".trimIndent()
val event = WsEvent.parse(frame)
assertTrue("turn_begin 必须被认出来(它是灰态的唯一入口)", event is WsEvent.TurnBegin)
event as WsEvent.TurnBegin
assertEquals("web", event.sessionId)
assertEquals("turn-9f3c1a2b3c4d5e6f", event.turnId)
assertEquals("parallel", event.kind)
assertEquals(listOf("msg-7f2a19c4e5"), event.messageIds)
assertEquals(2, event.leg)
assertEquals(2, event.activeTurns)
}
@Test
fun `a turn begin without a shell turnId keeps its message ids but no turn id`() {
// 帧壳 turnId 是这一路的唯一权威标识:没有它就不能建立绑定(绑了也收不了口)。
// 解析层如实给出 null,由消费方决定「只消费 activeTurns」。
val frame = """{"type":"turn_begin","sessionId":"web","data":{"kind":"turn","messageIds":["m-1"],"leg":1}}"""
val event = WsEvent.parse(frame) as WsEvent.TurnBegin
assertNull(event.turnId)
assertEquals(listOf("m-1"), event.messageIds)
assertEquals(0, event.activeTurns)
}
@Test
fun `a turn begin falls back to the session id inside the payload`() {
// 规范说会话 id 只在帧壳里(data 内不重复);真出现了也照既有口径取有值者。
val frame = """{"type":"turn_begin","data":{"sessionId":"web:abc","messageIds":["m-1"]}}"""
val event = WsEvent.parse(frame) as WsEvent.TurnBegin
assertEquals("web:abc", event.sessionId)
}
@Test
fun `blank message ids inside turn begin are dropped`() {
val frame = """{"type":"turn_begin","sessionId":"web","turnId":"t1","data":{"messageIds":["m-1"," "]}}"""
val event = WsEvent.parse(frame) as WsEvent.TurnBegin
assertEquals(listOf("m-1"), event.messageIds)
}
// —— steer_ack 的帧壳 turnId(规范 §2.5)——
@Test
fun `steer ack carries the shell turn id next to its v0_57 payload`() {
val frame = """
{"type":"steer_ack","sessionId":"web","turnId":"turn-9f3c1a2b3c4d5e6f",
"data":{"messageId":"msg-7f2a19c4e5","atStep":2}}
""".trimIndent()
val event = WsEvent.parse(frame)
assertTrue(event is WsEvent.SteerAck)
event as WsEvent.SteerAck
assertEquals("web", event.sessionId)
assertEquals("turn-9f3c1a2b3c4d5e6f", event.turnId)
assertEquals("msg-7f2a19c4e5", event.messageId)
assertEquals(2, event.atStep)
}
@Test
fun `a legacy steer ack without a turn id still parses with a null turn id`() {
// v0.57 ~ v0.64 的帧:事件照旧成形(不会被当未知帧丢掉),只是没有 turnId 可以绑定。
val frame = """{"type":"steer_ack","sessionId":"web","data":{"messageId":"m-1","atStep":1}}"""
val event = WsEvent.parse(frame) as WsEvent.SteerAck
assertNull(event.turnId)
assertEquals("m-1", event.messageId)
}
// —— turn.end(规范 §1.2 的收口事件)——
@Test
fun `turn end is parsed from the frame shell alone`() {
val frame = """{"type":"turn.end","sessionId":"web","turnId":"turn-1"}"""
val event = WsEvent.parse(frame)
assertTrue(event is WsEvent.TurnEnd)
event as WsEvent.TurnEnd
assertEquals("web", event.sessionId)
assertEquals("turn-1", event.turnId)
}
@Test
fun `turn end without any payload is not a bad frame`() {
// 后端 `streamEventData` 对 `StreamKindTurnEnd` 返回 nil(载荷天然是空的)。
// 若解析层要求 data 对象才认,这一帧就会被丢掉 —— 而那正是「灰态收不了口」的成因。
assertTrue(WsEvent.parse("""{"type":"turn.end","sessionId":"web","turnId":"t1"}""") is WsEvent.TurnEnd)
assertTrue(WsEvent.parse("""{"type":"turn.end"}""") is WsEvent.TurnEnd)
}
// —— 忙态帧的可选计数(规范 §2.6)——
@Test
fun `session busy carries active turns when the server reports them`() {
val frame = """{"type":"session_busy","sessionId":"web","data":{"sessionId":"web","mode":"executing","activeTurns":2}}"""
val event = WsEvent.parse(frame) as WsEvent.SessionBusy
assertEquals("executing", event.mode)
assertEquals(2, event.activeTurns)
}
@Test
fun `a missing active turns key stays null rather than becoming zero`() {
// 「缺席」与「0」绝不能互相顶替:拿缺席当 0 会把并行两路误报成空闲。
val old = WsEvent.parse("""{"type":"session_busy","sessionId":"web","data":{"sessionId":"web","mode":"thinking"}}""")
assertNull((old as WsEvent.SessionBusy).activeTurns)
}
@Test
fun `an explicit zero active turns is reported as zero`() {
val frame = """{"type":"session_idle","sessionId":"web","data":{"sessionId":"web","mode":"idle","activeTurns":0}}"""
assertEquals(0, (WsEvent.parse(frame) as WsEvent.SessionIdle).activeTurns)
}
@Test
fun `pending drain and interrupted carry their optional counts`() {
val drain = WsEvent.parse(
"""{"type":"pending_drain","sessionId":"web","data":{"sessionId":"web","mode":"thinking","queueDepth":1,"activeTurns":1}}""",
) as WsEvent.PendingDrain
assertEquals(1, drain.activeTurns)
val interrupted = WsEvent.parse(
"""{"type":"session_interrupted","sessionId":"web","data":{"sessionId":"web","mode":"executing","dropped":0,"cancelledTurns":1,"activeTurns":1}}""",
) as WsEvent.SessionInterrupted
assertEquals(1, interrupted.cancelledTurns)
assertEquals(1, interrupted.activeTurns)
}
@Test
fun `pending ack carries the reason code when the server gives one`() {
val frame =
"""{"type":"pending_ack","sessionId":"web","data":{"sessionId":"web","mode":"executing","queueDepth":2,"reason":"parallel_limit"}}"""
val event = WsEvent.parse(frame) as WsEvent.PendingAck
assertEquals("parallel_limit", event.reason)
assertEquals(2, event.queueDepth)
}
@Test
fun `an ordinary pending ack has no reason at all`() {
// 服务端明写「其它情况该键缺席,绝不伪造」——所以这里必须是 null,不能是 ""。
val frame = """{"type":"pending_ack","sessionId":"web","data":{"sessionId":"web","mode":"thinking","queueDepth":1}}"""
assertNull((WsEvent.parse(frame) as WsEvent.PendingAck).reason)
}
// —— 历史行的可选 turnId(规范 §2.7)——
@Test
fun `a persisted message may carry a turn id`() {
val frame = """
{"type":"message_persisted","sessionId":"web","data":{"message":
{"id":"msg-7f2a19c4e5","sessionId":"web","role":"user","content":"顺便帮我查一下明天的天气",
"createdAt":1760000000123,"turnId":"turn-9f3c1a2b3c4d5e6f"}}}
""".trimIndent()
val event = WsEvent.parse(frame) as WsEvent.MessagePersisted
assertEquals("turn-9f3c1a2b3c4d5e6f", event.message?.turnId)
}
@Test
fun `an old persisted row has no turn id`() {
val frame = """
{"type":"message_persisted","sessionId":"web","data":{"message":
{"id":"msg-1","sessionId":"web","role":"assistant","content":"好的","createdAt":1}}}
""".trimIndent()
val event = WsEvent.parse(frame) as WsEvent.MessagePersisted
assertNull(event.message?.turnId)
}
@Test
fun `history rows may carry turn ids and old rows simply do not`() {
val frame = """
{"type":"history","data":{"messages":[
{"id":"m-1","sessionId":"web","role":"user","content":"甲","createdAt":1,"turnId":"t1"},
{"id":"m-2","sessionId":"web","role":"assistant","content":"乙","createdAt":2}]}}
""".trimIndent()
val event = WsEvent.parse(frame) as WsEvent.History
assertEquals("t1", event.messages[0].turnId)
assertNull(event.messages[1].turnId)
}
}
@@ -0,0 +1,83 @@
package top.yeij.cyrene.data.ws
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Test
/**
* 忙着并行规范的上行两处**纯增量**(§2.2 `chat.parallel` / §2.3 `interrupt.turnId`)。
*
* 本层只负责一件事:**省略时整键不进线上、给值时原样发出**。本端今天一个字节都不发这两个字段
* (真并行要用户显式声明,声明入口不在本次范围内),所以这一组测试真正钉住的是
* 「老形状的那条帧仍然逐字节相同」——那是与老服务端兼容的全部依据。
*/
class WsOutboundParallelTest {
/** 加入 `parallel` 之前那条 `chat` 帧的逐字形状。 */
private val legacyChat = """{"type":"chat","data":{"sessionId":"web","text":"你好","deviceId":"d"}}"""
/** 加入 `turnId` 之前那条 `interrupt` 帧的逐字形状。 */
private val legacyInterrupt = """{"type":"interrupt","data":{"sessionId":"web","scope":"current"}}"""
// —— chat.parallel ——
@Test
fun `a chat without parallel is today's frame byte for byte`() {
assertEquals(legacyChat, WsOutbound.Chat(text = "你好", sessionId = "web", deviceId = "d").toJson())
assertEquals(
legacyChat,
WsOutbound.Chat(text = "你好", sessionId = "web", deviceId = "d", parallel = null).toJson(),
)
}
@Test
fun `omitting parallel omits the key itself rather than sending null`() {
val json = WsOutbound.Chat(text = "你好", sessionId = "web").toJson()
// `"parallel":null` 与「整键省略」在服务端不是一回事:后者才是「未声明」(规范 §2.2)。
assertFalse("省略时不该出现 parallel 键:$json", json.contains("parallel"))
}
@Test
fun `parallel true is carried as a real boolean`() {
val json = WsOutbound.Chat(text = "你好", sessionId = "web", parallel = true).toJson()
assertEquals("""{"type":"chat","data":{"sessionId":"web","text":"你好","parallel":true}}""", json)
}
@Test
fun `parallel false is sent as false and never silently dropped`() {
// 显式 false 与省略在服务端同义(都落回排队),但客户端**不改写**用户给了什么。
assertEquals(
"""{"type":"chat","data":{"sessionId":"web","text":"你好","parallel":false}}""",
WsOutbound.Chat(text = "你好", sessionId = "web", parallel = false).toJson(),
)
}
// —— interrupt.turnId ——
@Test
fun `an interrupt without a turn id is today's frame byte for byte`() {
assertEquals(legacyInterrupt, WsOutbound.Interrupt(sessionId = "web", scope = "current").toJson())
assertEquals(
legacyInterrupt,
WsOutbound.Interrupt(sessionId = "web", scope = "current", turnId = null).toJson(),
)
}
@Test
fun `a turn id is carried next to the scope`() {
assertEquals(
"""{"type":"interrupt","data":{"sessionId":"web","scope":"current","turnId":"turn-9f3c"}}""",
WsOutbound.Interrupt(sessionId = "web", scope = "current", turnId = "turn-9f3c").toJson(),
)
}
@Test
fun `the session id stays first so older servers see the same key order`() {
assertEquals(
"""{"type":"interrupt","data":{"sessionId":"web","turnId":"turn-1"}}""",
WsOutbound.Interrupt(sessionId = "web", turnId = "turn-1").toJson(),
)
}
}
@@ -0,0 +1,355 @@
package top.yeij.cyrene.domain.chat
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNull
import org.junit.Assert.assertSame
import org.junit.Assert.assertTrue
import org.junit.Test
/**
* 忙时并行的纯逻辑:**两级分桶 / 六态判定 / 灰态收口 / 消息↔回合绑定**。
*
* 这一层不碰协程也不碰 Android,所以「规范怎么说」与「代码怎么做」可以逐条对着钉死。
* 三组测试的界线也是规范 §1.7 那三件事:分桶、绑定与收口、六态。
*
* 另外有一组**降级**测试(`字段缺失`):服务的今天没有 `turn_begin` / `steer_ack.turnId`,
* 也没有 `parallel`,本模块必须做到「只按今天收到的帧走,行为与今天逐字相同」。
*/
class ParallelTurnsTest {
// —— 分桶(规范 §1.4 / §1.7 第 1 条)——
@Test
fun `a frame without a turnId lands in the anonymous bucket`() {
val bucket = turnBucketOf("web", null, null)
assertEquals(TurnBucket("web", ""), bucket)
assertTrue(bucket.isAnonymous)
}
@Test
fun `every frame without a turnId lands in the same bucket`() {
// 这一条就是「字段缺失 ⇒ 行为与今天逐字相同」的判据:老代码只有一个流式气泡,
// 分桶之后这些帧必须全部落在**同一个**桶里,否则一条回复会被劈成两半。
val a = turnBucketOf("web", null, null)
val b = turnBucketOf("web", " ", null)
val c = turnBucketOf("web", "", "")
assertEquals(a, b)
assertEquals(a, c)
}
@Test
fun `the frame turnId wins over the remembered one`() {
assertEquals(
TurnBucket("web", "turn-2"),
turnBucketOf("web", "turn-2", "turn-1"),
)
}
@Test
fun `a missing turnId falls back to the remembered one`() {
// 本会话正在跑 turn-1,某一帧没带 turnId ⇒ 它属于 turn-1(老代码里就是那唯一一个气泡)。
assertEquals(
TurnBucket("web", "turn-1"),
turnBucketOf("web", null, "turn-1"),
)
}
@Test
fun `a blank frame turnId counts as missing`() {
assertEquals(
TurnBucket("web", "turn-1"),
turnBucketOf("web", " ", "turn-1"),
)
}
@Test
fun `two turns in one session never share a bucket`() {
val leg1 = turnBucketOf("web", "turn-1", null)
val leg2 = turnBucketOf("web", "turn-2", "turn-1")
assertFalse(leg1 == leg2)
}
@Test
fun `the same turnId in another session is another bucket`() {
// 两级分桶的第二级:`turnId` 只在会话内唯一,跨会话不能合并。
assertFalse(turnBucketOf("web", "turn-1", null) == turnBucketOf("web:abc", "turn-1", null))
}
// —— 收口帧的落桶(不许新开桶)——
@Test
fun `a closing frame lands on its own bucket when it is known`() {
assertEquals(
TurnBucket("web", "turn-1"),
closingBucketOf("web", "turn-1", "turn-1", listOf("turn-1", "turn-2")),
)
}
@Test
fun `a closing frame with an unknown turnId falls back to the remembered bucket`() {
assertEquals(
TurnBucket("web", "turn-1"),
closingBucketOf("web", "turn-9", "turn-1", listOf("turn-1")),
)
}
@Test
fun `a closing frame with no recognisable id lands on the most recent bucket`() {
// 为什么这条不能省:老代码里 `stream_end` 无条件收口那唯一一个气泡。若给它新开一个桶,
// 真正在跑的那个桶就永远收不了口(光标一直转)。
assertEquals(
TurnBucket("web", "turn-2"),
closingBucketOf("web", null, null, listOf("turn-1", "turn-2")),
)
}
@Test
fun `a closing frame never invents a bucket out of nothing`() {
assertEquals(TurnBucket("web", ""), closingBucketOf("web", "turn-1", null, emptyList()))
}
// —— 六态判定(规范 §1.1)——
@Test
fun `a local bubble is sending until it is bound or queued`() {
assertEquals(
TurnPhase.Sending,
turnPhaseOf(local = true, binding = null, queued = false, rejected = false),
)
}
@Test
fun `a queued local bubble is queued not sending`() {
assertEquals(
TurnPhase.Queued,
turnPhaseOf(local = true, binding = null, queued = true, rejected = false),
)
}
@Test
fun `a persisted bubble with nothing else is sent`() {
assertEquals(
TurnPhase.Sent,
turnPhaseOf(local = false, binding = null, queued = false, rejected = false),
)
}
@Test
fun `a queued persisted bubble is queued and not grey`() {
// 规范 §1.3:`queued` **不是**灰态。排队与「已读、未答」是两件事,不许合并。
assertEquals(
TurnPhase.Queued,
turnPhaseOf(local = false, binding = null, queued = true, rejected = false),
)
}
@Test
fun `a bound message that is not closed is answering`() {
assertEquals(
TurnPhase.Answering,
turnPhaseOf(local = false, binding = binding("turn-1"), queued = false, rejected = false),
)
}
@Test
fun `a bound message that is closed is answered`() {
assertEquals(
TurnPhase.Answered,
turnPhaseOf(local = false, binding = binding("turn-1").copy(closed = true), queued = false, rejected = false),
)
}
@Test
fun `a rejection outranks every other signal`() {
// ⑥ 是最重的事实:这条根本没进去。再拿「有没有被接手」去描述它就是错的。
assertEquals(
TurnPhase.Rejected,
turnPhaseOf(local = false, binding = binding("turn-1"), queued = true, rejected = true),
)
assertEquals(
TurnPhase.Rejected,
turnPhaseOf(local = true, binding = null, queued = false, rejected = true),
)
}
@Test
fun `an unanswered binding beats the queued mark`() {
// ③ → ④ 的换态:被接手之后,「排队中」这个角标必须让位给灰态。
assertEquals(
TurnPhase.Answering,
turnPhaseOf(local = false, binding = binding("turn-1"), queued = true, rejected = false),
)
}
// —— 灰态收口(规范 §1.2)——
@Test
fun `the matching turn end closes the grey state`() {
assertTrue(closesAnswering("turn-1", "turn-1"))
}
@Test
fun `another turn never closes this one`() {
assertFalse(closesAnswering("turn-1", "turn-2"))
}
@Test
fun `two blanks are not a match`() {
// 「都为空所以相等」会让老客户端 / 老服务端上的消息被随便一帧误收口——这条是刻意的。
assertFalse(closesAnswering(null, null))
assertFalse(closesAnswering("", " "))
assertFalse(closesAnswering("turn-1", null))
assertFalse(closesAnswering(null, "turn-1"))
}
// —— 绑定表 ——
private fun binding(turnId: String, vararg messageIds: String) =
TurnBinding(turnId = turnId, messageIds = messageIds.toList())
@Test
fun `turn begin binds the messages it names`() {
val turns = emptyList<TurnBinding>().withTurnBegin("turn-1", 1, listOf("m-a"), 1)
assertEquals(1, turns.size)
assertEquals(TurnBinding("turn-1", 1, listOf("m-a"), 1), turns.single())
assertEquals("turn-1", turns.bindingOfMessage("m-a")?.turnId)
assertNull(turns.bindingOfMessage("m-b"))
}
@Test
fun `two legs keep their own messages`() {
val turns = emptyList<TurnBinding>()
.withTurnBegin("turn-1", 1, listOf("m-a"), 1)
.withTurnBegin("turn-2", 2, listOf("m-b"), 2)
assertEquals(listOf("turn-1", "turn-2"), turns.map { it.turnId })
assertEquals("turn-1", turns.bindingOfMessage("m-a")?.turnId)
assertEquals("turn-2", turns.bindingOfMessage("m-b")?.turnId)
}
@Test
fun `a repeated turn begin merges instead of replacing`() {
val turns = emptyList<TurnBinding>()
.withTurnBegin("turn-1", 1, listOf("m-a"), 1)
.withTurnBegin("turn-1", 1, listOf("m-b"), 1)
assertEquals(listOf("m-a", "m-b"), turns.single().messageIds)
}
@Test
fun `a repeated turn begin never reopens a closed leg`() {
// 重开会让气泡回到灰态,而第二次收口永远不会再来(那一帧已经发过了)。
val closed = emptyList<TurnBinding>()
.withTurnBegin("turn-1", 1, listOf("m-a"), 1)
.closingTurn("turn-1")
val again = closed.withTurnBegin("turn-1", 1, listOf("m-a"), 1)
assertTrue(again.single().closed)
}
@Test
fun `a blank turnId in turn begin changes nothing`() {
val turns = emptyList<TurnBinding>()
assertSame(turns, turns.withTurnBegin("", 1, listOf("m-a"), 1))
assertSame(turns, turns.withTurnBegin(" ", 1, listOf("m-a"), 1))
}
@Test
fun `steer ack binds a message to an existing leg`() {
val turns = emptyList<TurnBinding>()
.withTurnBegin("turn-1", 1, listOf("m-a"), 1)
.withSteeredMessage("turn-1", "m-b")
assertEquals(listOf("m-a", "m-b"), turns.single().messageIds)
assertEquals("turn-1", turns.bindingOfMessage("m-b")?.turnId)
}
@Test
fun `steer ack creates a leg when the turn is unknown`() {
// 服务端可能先发 steer_ack 再补 turn_begin;先建一条只有这一条消息的表,之后合并即可。
val turns = emptyList<TurnBinding>().withSteeredMessage("turn-1", "m-b")
assertEquals(TurnBinding("turn-1", UNKNOWN_LEG, listOf("m-b")), turns.single())
}
@Test
fun `steer ack with a blank turnId binds nothing`() {
val turns = listOf(binding("turn-1", "m-a"))
assertSame(turns, turns.withSteeredMessage("", "m-b"))
assertSame(turns, turns.withSteeredMessage("turn-1", " "))
}
@Test
fun `closing one turn leaves the other leg open`() {
val turns = emptyList<TurnBinding>()
.withTurnBegin("turn-1", 1, listOf("m-a"), 1)
.withTurnBegin("turn-2", 2, listOf("m-b"), 2)
.closingTurn("turn-2")
assertEquals(TurnPhase.Answering, phaseOfMessage(turns, "m-a"))
assertEquals(TurnPhase.Answered, phaseOfMessage(turns, "m-b"))
}
@Test
fun `closing with an unknown id closes nothing and returns the same instance`() {
val turns = listOf(binding("turn-1", "m-a"))
assertSame(turns, turns.closingTurn(null))
assertSame(turns, turns.closingTurn(""))
assertSame(turns, turns.closingTurn("turn-9"))
}
@Test
fun `closing twice is the same as closing once`() {
val once = listOf(binding("turn-1", "m-a")).closingTurn("turn-1")
assertSame(once, once.closingTurn("turn-1"))
}
@Test
fun `all closed closes every open leg and is idempotent`() {
val turns = emptyList<TurnBinding>()
.withTurnBegin("turn-1", 1, listOf("m-a"), 1)
.withTurnBegin("turn-2", 2, listOf("m-b"), 2)
val closed = turns.allClosed()
assertTrue(closed.all { it.closed })
assertSame(closed, closed.allClosed())
}
@Test
fun `a blank message id never matches a binding`() {
val turns = listOf(binding("turn-1", ""))
assertNull(turns.bindingOfMessage(""))
assertNull(turns.bindingOfMessage(" "))
}
// —— 路小标 ——
@Test
fun `only the second leg and later get a badge`() {
// 单路是常态:给唯一一路挂「并行路 1」只是给每条回复加一个新的视觉噪音。
assertNull(legBadgeOf(UNKNOWN_LEG))
assertNull(legBadgeOf(1))
assertEquals("并行路 2", legBadgeOf(2))
assertEquals("并行路 3", legBadgeOf(3))
}
private fun phaseOfMessage(turns: List<TurnBinding>, messageId: String): TurnPhase =
turnPhaseOf(
local = false,
binding = turns.bindingOfMessage(messageId),
queued = false,
rejected = false,
)
}
@@ -0,0 +1,281 @@
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.data.api.dto.HistoryMessageDto
import top.yeij.cyrene.domain.chat.TurnBinding
import top.yeij.cyrene.domain.chat.TurnBucket
import top.yeij.cyrene.domain.chat.TurnPhase
import top.yeij.cyrene.domain.chat.withTurnBegin
import top.yeij.cyrene.domain.parse.ParsedBubble
/**
* 流式气泡的**两级分桶**(规范 §1.4 / §1.7 第 1 条)在 UI 层的那一半。
*
* 这里的头号回归是「两路交错的 delta 不许糊成一条」——老代码只有一个 `StreamingState`,
* 所有 `message.delta` 都往它上面堆;分桶之后每一路各有一份。第二件事是**单路逐字不变**:
* 帧壳不带 `turnId` 时全部帧落匿名桶,`streams` 的长度恒为 1。
*/
class TurnStreamsTest {
private fun bucket(turnId: String) = TurnBucket("web", turnId)
private fun bubble(text: String) =
ParsedBubble(body = text, parts = emptyList(), head = emptyList(), tail = emptyList())
private fun streaming(vararg texts: String) = StreamingState(bubbles = texts.map(::bubble))
private fun leg(turnId: String, vararg texts: String) = StreamingLeg(bucket(turnId), streaming(*texts))
// —— 两路各长各的 ——
@Test
fun `two turns accumulate in their own legs`() {
val legs = emptyList<StreamingLeg>()
.withLeg(bucket("t1")) { it.copy(bubbles = listOf(bubble("第一路"))) }
.withLeg(bucket("t2")) { it.copy(bubbles = listOf(bubble("第二路"))) }
assertEquals(2, legs.size)
assertEquals(listOf("第一路"), bodies(legs, "t1"))
assertEquals(listOf("第二路"), bodies(legs, "t2"))
}
@Test
fun `interleaved deltas never merge into one leg`() {
// 规范 §1.4 点名的那个错:按「当前气泡」累加,两条回复会被糊成一条。
val legs = emptyList<StreamingLeg>()
.withLeg(bucket("t1")) { it.copy(bubbles = it.bubbles + bubble("A1")) }
.withLeg(bucket("t2")) { it.copy(bubbles = it.bubbles + bubble("B1")) }
.withLeg(bucket("t1")) { it.copy(bubbles = it.bubbles + bubble("A2")) }
.withLeg(bucket("t2")) { it.copy(bubbles = it.bubbles + bubble("B2")) }
assertEquals(listOf("A1", "A2"), bodies(legs, "t1"))
assertEquals(listOf("B1", "B2"), bodies(legs, "t2"))
}
@Test
fun `one turn is exactly one leg`() {
// 「字段缺失 ⇒ 行为与今天逐字相同」在 UI 层的判据:单路时长度恒为 1。
val legs = emptyList<StreamingLeg>()
.withLeg(bucket("")) { it.copy(bubbles = listOf(bubble("A"))) }
.withLeg(bucket("")) { it.copy(bubbles = it.bubbles + bubble("B")) }
assertEquals(1, legs.size)
assertEquals(listOf("A", "B"), bodies(legs, ""))
}
@Test
fun `legs keep their creation order`() {
val legs = emptyList<StreamingLeg>()
.withLeg(bucket("t1")) { it }
.withLeg(bucket("t2")) { it }
.withLeg(bucket("t1")) { it }
assertEquals(listOf("t1", "t2"), legs.map { it.bucket.turnId })
}
// —— 新一轮开始时的收口 ——
@Test
fun `beginning a turn drops the other finished legs`() {
// 老代码的 freshTurnIfFinished:上一轮已结束就把那一个气泡整个清掉(内容已被落库行取代)。
val legs = listOf(
StreamingLeg(bucket("t1"), streaming("旧内容").copy(active = false)),
StreamingLeg(bucket("t2"), streaming("还在跑")),
)
val after = legs.beginLeg(bucket("t2"))
assertEquals(listOf("t2"), after.map { it.bucket.turnId })
}
@Test
fun `beginning a turn never touches a leg that is still running`() {
val legs = listOf(
StreamingLeg(bucket("t1"), streaming("另一路")),
StreamingLeg(bucket("t2"), streaming("旧").copy(active = false)),
)
val after = legs.beginLeg(bucket("t2"))
assertEquals(listOf("t1", "t2"), after.map { it.bucket.turnId })
assertEquals(listOf("另一路"), bodies(after, "t1"))
assertEquals(emptyList<String>(), bodies(after, "t2"))
}
@Test
fun `beginning a turn on an unknown bucket adds nothing`() {
// 新的桶由紧接着的 withLeg 建;这里凭空塞一个空态只会在坏时序下多出一个看不见的空气泡。
val legs = listOf(leg("t1", "内容"))
assertEquals(legs, legs.beginLeg(bucket("t9")))
}
// —— 收口 ——
@Test
fun `finishing a leg keeps its text and stops the cursor`() {
val legs = listOf(StreamingLeg(bucket("t1"), streaming("半句")))
val after = legs.finishingLeg(bucket("t1"), listOf(bubble("整句")))
assertEquals(listOf("整句"), bodies(after, "t1"))
assertTrue(!after.single().state.active)
assertTrue(!after.single().state.thinkingActive)
}
@Test
fun `finishing without a parser keeps the text that is already on screen`() {
// 「没有新正文」不该被写成「正文是空的」——那会把已经显示出来的字擦掉。
val legs = listOf(StreamingLeg(bucket("t1"), streaming("已经显示的")))
val after = legs.finishingLeg(bucket("t1"), null)
assertEquals(listOf("已经显示的"), bodies(after, "t1"))
assertTrue(!after.single().state.active)
}
@Test
fun `finishing an unknown leg creates nothing`() {
val legs = listOf(leg("t1", "内容"))
assertSame(legs, legs.finishingLeg(bucket("t9"), emptyList()))
}
@Test
fun `dropping the persisted leg keeps the other one`() {
val legs = listOf(leg("t1", "A"), leg("t2", "B"))
val after = legs.withoutLeg(bucket("t1"))
assertEquals(listOf("t2"), after.map { it.bucket.turnId })
assertSame(after, after.withoutLeg(bucket("t1")))
}
// —— 六态在状态上的推导 ——
@Test
fun `without any turn frames every message is where it is today`() {
// 老服务端(没有 turn_begin / steer_ack.turnId):已落库 = ②,本地气泡 = ①。
val state = ChatUiState(messages = listOf(message("m-1")))
assertEquals(TurnPhase.Sent, state.phaseOf(state.messages.single()))
assertEquals(TurnPhase.Sending, state.phaseOf(PendingMessage("local-1", "你好")))
}
@Test
fun `a bound message is grey until its turn closes`() {
val turns = emptyList<TurnBinding>().withTurnBegin("t1", 2, listOf("m-1"), 2)
val bound = ChatUiState(messages = listOf(message("m-1")), turns = turns)
assertEquals(TurnPhase.Answering, bound.phaseOf(bound.messages.single()))
assertTrue(bound.phaseOf(bound.messages.single()) != TurnPhase.Queued)
}
@Test
fun `the queued mark is found through the client message id too`() {
// 队列回执先到、落库后到:那时本地认识的键是 `localId`,而消息上的键是 `clientMsgId`。
val state = ChatUiState(
messages = listOf(message("m-1", clientMsgId = "local-1")),
queuedKeys = setOf("local-1"),
)
assertEquals(TurnPhase.Queued, state.phaseOf(state.messages.single()))
}
@Test
fun `the queued mark on a local bubble survives into the composer view`() {
val state = ChatUiState(pending = listOf(PendingMessage("local-1", "你好")), queuedKeys = setOf("local-1"))
assertEquals(TurnPhase.Queued, state.phaseOf(state.pending.single()))
}
// —— 路小标与跳转 ——
@Test
fun `a single leg carries no badge`() {
val turns = emptyList<TurnBinding>().withTurnBegin("t1", 1, listOf("m-1"), 1)
val state = ChatUiState(messages = listOf(message("m-1")), turns = turns)
assertNull(state.legBadgeForMessage(state.messages.single()))
assertNull(state.legBadgeForTurn("t1"))
}
@Test
fun `the second leg carries its badge on the question and on the stream`() {
val turns = emptyList<TurnBinding>().withTurnBegin("t2", 2, listOf("m-1"), 2)
val state = ChatUiState(messages = listOf(message("m-1")), turns = turns)
assertEquals("并行路 2", state.legBadgeForMessage(state.messages.single()))
assertEquals("并行路 2", state.legBadgeForTurn("t2"))
}
@Test
fun `the answer of a leg carries the same badge through its own turn id`() {
// 回复侧:历史行的可选 turnId(规范 §2.7)指出它是哪一路产出的 ⇒ 落库之后小标仍在。
val turns = emptyList<TurnBinding>().withTurnBegin("t2", 2, listOf("m-1"), 2)
val answer = message("m-9", role = "assistant", turnId = "t2")
val state = ChatUiState(messages = listOf(message("m-1"), answer), turns = turns)
assertEquals("并行路 2", state.legBadgeForMessage(answer))
}
@Test
fun `an old answer row without a turn id carries no badge`() {
// v0.65 之前历史行没有 turnId ⇒ 挂不出标,也就一个像素都不多画。
val turns = emptyList<TurnBinding>().withTurnBegin("t2", 2, listOf("m-1"), 2)
val answer = message("m-9", role = "assistant")
val state = ChatUiState(messages = listOf(message("m-1"), answer), turns = turns)
assertNull(state.legBadgeForMessage(answer))
}
@Test
fun `jumping to a question maps to the reverse layout index`() {
// items = [已落库 m-1, m-2, m-3](升序);reverseLayout 下懒下标 = size - 1 - 升序下标。
val turns = emptyList<TurnBinding>().withTurnBegin("t1", 1, listOf("m-1"), 1)
val state = ChatUiState(
messages = listOf(message("m-1"), message("m-2"), message("m-3")),
turns = turns,
)
assertEquals(2, state.jumpIndexForTurn("t1"))
assertNull(state.jumpIndexForTurn("t9"))
}
@Test
fun `a turn whose question is not loaded has no jump target`() {
val turns = emptyList<TurnBinding>().withTurnBegin("t1", 1, listOf("m-unknown"), 1)
val state = ChatUiState(messages = listOf(message("m-1")), turns = turns)
assertNull(state.jumpIndexForTurn("t1"))
}
// —— 气泡角标文案 ——
@Test
fun `the footer only mentions queued and answering`() {
assertEquals("来自 安卓", bubbleFooter("来自 安卓", TurnPhase.Sent, null))
assertEquals("排队中 · 来自 安卓", bubbleFooter("来自 安卓", TurnPhase.Queued, null))
assertEquals("已读,正在回答… · 来自 安卓", bubbleFooter("来自 安卓", TurnPhase.Answering, null))
assertEquals("并行路 2 · 来自 安卓", bubbleFooter("来自 安卓", TurnPhase.Answered, "并行路 2"))
assertNull(bubbleFooter(null, TurnPhase.Sent, null))
}
private fun bodies(legs: List<StreamingLeg>, turnId: String): List<String> =
legs.leg(bucket(turnId))?.state?.bubbles?.map { it.body }.orEmpty()
private fun message(id: String, role: String = "user", clientMsgId: String? = null, turnId: String? = null) =
HistoryMessageDto(
id = id,
sessionId = "web",
role = role,
content = id,
clientMsgId = clientMsgId,
createdAt = 1L,
turnId = turnId,
)
}