diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/ChatDto.kt b/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/ChatDto.kt index d209995..0727e88 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/ChatDto.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/ChatDto.kt @@ -46,6 +46,15 @@ data class ChatRequest( * WS `chat` 帧的 4096 字节读上限仍在,且上行帧里没有承载附件的余地。 */ val documents: List = 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, ) /** diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/MessageDto.kt b/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/MessageDto.kt index 19e6107..7e94ac8 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/MessageDto.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/api/dto/MessageDto.kt @@ -43,6 +43,22 @@ data class HistoryMessageDto( */ val documents: List = 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, ) /** diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt index 7486246..dc6bb22 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt @@ -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 = 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(data).delta, turnId, seq, sessionId) "stream_end" -> decode(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(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(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(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(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(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(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(data) + .let { + SteerAck( + resolveSessionId(sessionId, it.sessionId), it.messageId, it.atStep, turnId, seq, + ) + } "pending_full" -> decode(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 = 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?}`。前三个键来自加入 diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsOutbound.kt b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsOutbound.kt index 3ee69a4..b746bfc 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsOutbound.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsOutbound.kt @@ -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) } } } diff --git a/app/src/main/kotlin/top/yeij/cyrene/domain/chat/ParallelTurns.kt b/app/src/main/kotlin/top/yeij/cyrene/domain/chat/ParallelTurns.kt new file mode 100644 index 0000000..22bd8ed --- /dev/null +++ b/app/src/main/kotlin/top/yeij/cyrene/domain/chat/ParallelTurns.kt @@ -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, +): 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 = emptyList(), + /** 服务端最近一次报的该会话在途路数(`turn_begin.activeTurns`;0 = 没报过)。 */ + val activeTurns: Int = 0, + /** 该路是否已收口(`turn.end` / 同 `turnId` 的助手落库行)。 */ + val closed: Boolean = false, +) + +/** 这条消息被哪一路接手;`null` = 没被任何一路接手(也是「按顺序渲染」那档的老行为)。 */ +fun List.bindingOfMessage(messageId: String): TurnBinding? = + firstOrNull { binding -> messageId.isNotBlank() && binding.messageIds.contains(messageId) } + +/** + * 收下一条 `turn_begin`(纯函数,返回新表)。 + * + * 同 [TurnBinding.turnId] 已存在时**合并**而不是新增:`messageIds` 取并集(保持原顺序在前), + * `leg` / `activeTurns` 用新报的值,**`closed` 保持原值**——一次重复到达的 `turn_begin` 不该把一条 + * 已经收口的路「重新打开」(那会让气泡回到灰态并且再也等不到第二次收口)。 + * 表里没有它时按**追加**处理:两路的顺序就是它们被启动的顺序。 + */ +fun List.withTurnBegin( + turnId: String, + leg: Int, + messageIds: List, + activeTurns: Int, +): List { + 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.withSteeredMessage(turnId: String, messageId: String): List { + 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.closingTurn(closingTurnId: String?): List { + 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.allClosed(): List { + 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() } diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatScreen.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatScreen.kt index ecb2a16..fc386dc 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatScreen.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatScreen.kt @@ -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 { "" } } /** diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt index 5df67fb..7319094 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt @@ -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 = emptyList(), val pending: List = 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 = 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 = emptyList(), + /** + * 收到过「这条进队列了」回执的键:本地乐观气泡的 `localId`,以及落库后的服务端 `messageId`。 + * + * 两种写法都要认,因为 `pending_ack`(广播、**不带消息身份**)与 `message_persisted` + * (带 `clientMsgId` = 本端 `localId`)的先后顺序不确定。它是六态里 ③ 排队中的唯一判据 + * (规范 §1.1 ③ 的进入条件是「现状帧,形状不变」——服务端没说被拒的是哪一条,所以本端只按 + * 「我发出去之后收到了入队回执」认,**不猜**)。 + */ + val queuedKeys: Set = 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() + + /** + * **每一路**「这一回合是否已经收到过结构化正文帧」(`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() + + /** + * 本会话**最近一次见到的非空 `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 { + 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, ) } diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/MessageBubble.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/MessageBubble.kt index 13298dd..7da24a1 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/MessageBubble.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/MessageBubble.kt @@ -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) { @@ -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 + }, ) } } diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/chat/TurnStreams.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/TurnStreams.kt new file mode 100644 index 0000000..5212a1c --- /dev/null +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/chat/TurnStreams.kt @@ -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.leg(bucket: TurnBucket): StreamingLeg? = firstOrNull { it.bucket == bucket } + +/** + * 改某一路:**有则改、无则新建并追加到末尾**(顺序即两路的启动顺序,界面上先起的在上)。 + * + * 新建时以 [StreamingState] 的默认值为起点,所以「一路的第一帧」与老代码里 + * `state.streaming ?: StreamingState()` 是同一条线。 + */ +fun List.withLeg( + bucket: TurnBucket, + block: (StreamingState) -> StreamingState, +): List { + 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.beginLeg(bucket: TurnBucket): List { + 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.finishingLeg( + bucket: TurnBucket, + bubbles: List?, +): List { + 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.withoutLeg(bucket: TurnBucket): List = + 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) diff --git a/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventTurnFramesTest.kt b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventTurnFramesTest.kt new file mode 100644 index 0000000..0bf9a3f --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventTurnFramesTest.kt @@ -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) + } +} diff --git a/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsOutboundParallelTest.kt b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsOutboundParallelTest.kt new file mode 100644 index 0000000..d652ddf --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsOutboundParallelTest.kt @@ -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(), + ) + } +} diff --git a/app/src/test/kotlin/top/yeij/cyrene/domain/chat/ParallelTurnsTest.kt b/app/src/test/kotlin/top/yeij/cyrene/domain/chat/ParallelTurnsTest.kt new file mode 100644 index 0000000..664ddd1 --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/domain/chat/ParallelTurnsTest.kt @@ -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().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() + .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() + .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() + .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() + + 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() + .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().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() + .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() + .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, messageId: String): TurnPhase = + turnPhaseOf( + local = false, + binding = turns.bindingOfMessage(messageId), + queued = false, + rejected = false, + ) +} diff --git a/app/src/test/kotlin/top/yeij/cyrene/ui/chat/TurnStreamsTest.kt b/app/src/test/kotlin/top/yeij/cyrene/ui/chat/TurnStreamsTest.kt new file mode 100644 index 0000000..281cbf1 --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/ui/chat/TurnStreamsTest.kt @@ -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() + .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() + .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() + .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() + .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(), 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().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().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().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().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().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().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().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, turnId: String): List = + 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, + ) +}