fix(ws): 流式帧按 sessionId 收敛,修跨会话串台并退役 A1/D14
服务端从 v0.36 起就在结构化流帧(message.* / thinking.* / tool_call.* /
tool_execution.*)上带了 sessionId + turnId,而本端 WsEvent.parse 把这两个字段
**丢掉了**——所以「同一时刻只有一个会话在流式」这条长期约束(A1/D14)其实是
客户端自己造成的,不是后端限制。
而广播是发给**每条已认证连接**、不按会话订阅;每个 ChatScreen 的 VM 又各建一条
ChatSocket,切会话后旧 socket 与旧 VM 都还活着——于是旧 VM 会拿新会话的 delta
跑自己的流式状态机,思维链/工具卡/正文全线串台。
- data/ws/WsEvent.kt:基类开 open val sessionId(子类按帧壳覆盖),透传到 10 个
结构化事件类型;新增纯函数 isForSession()——**只排除「明确属于别的会话」的帧**,
空值视为「无法判定」而放行(与 historyForSession 对空值的处理一致)
- ui/chat/ChatViewModel.kt:onEvent 顶部加一道会话闸;history 帧**刻意豁免**,
它的归属由帧内每条消息的 sessionId 判定(比帧壳更细,即 2a49c25 的语义)
- data/ws/ChatSocket.kt:改掉那段描述「同一时刻只允许一个会话在流式」的过时 KDoc
顺带核实:**没有任何代码实现过 D14**(切会话时不拦截,sendInterrupt 更是零调用点),
它只是文档假设 + 解析器丢字段的副作用 → 因此 D14 退役、A1 闭环。
测试:194 全绿(178 → +16)。新增 WsEventSessionScopeTest:12 种帧类型的解码
(按真实帧壳形状)、向后兼容(缺字段 / 显式 null)、实抓的 history 帧、isForSession 规则。
未验证项:跨会话修复只由单测 + 后端自带的 ws_hub_stream_scope_test 覆盖,
没有实跑「两个会话同时流式」。
This commit is contained in:
1 parent
2a49c25e1b
commit
d12a920096
4 files changed
+322
-21
No files matched your search
@@ -41,10 +41,10 @@ import kotlin.random.Random
|
||||
* socket.close()
|
||||
* ```
|
||||
*
|
||||
* ## 会话归属约束
|
||||
* `message.*` / `thinking.*` / `tool_call.*` / `stream_chunk` 等流式帧**不带 `sessionId`**,
|
||||
* 本类不会替它们编造会话。上层必须把它们归属到「当前唯一的 active 流式会话」——
|
||||
* 也就是**同一时刻只允许一个会话在流式**(切会话时需先结束或 `interrupt`)。
|
||||
* ## 会话归属
|
||||
* 服务端自 v0.36 起在**每个结构化流帧的帧壳**上带 `sessionId`,本类把它原样透传
|
||||
* (见 [WsEvent.sessionId]),不补值。服务端广播**不按会话订阅**:同一条连接会收到**所有**
|
||||
* 会话的帧,因此收集 [events] 的上层**必须**按 [WsEvent.isForSession] 收敛到自己的会话。
|
||||
*
|
||||
* ## 心跳与重连
|
||||
* - OkHttp `pingInterval` 30s(服务端 ping 周期 54s / pong 等待 60s,客户端 30s 足够)。
|
||||
|
||||
@@ -15,12 +15,15 @@ import kotlinx.serialization.json.decodeFromJsonElement
|
||||
* 帧壳:`{type, turnId?, seq?, sessionId?, data?}`(api-reference §6.2),
|
||||
* 只有 `type` 必填,`data` 的形状由 `type` 决定。字段名逐字对齐 api-reference §6.4。
|
||||
*
|
||||
* ## 会话归属约束(重要)
|
||||
* `message.start/delta/end`、`thinking.*`、`tool_call.*`、`tool_execution.update`、
|
||||
* `stream_chunk` **不带 `sessionId`**。本层不会凭空补 `sessionId`,因此上层必须把这些事件
|
||||
* 归属到「当前唯一的 active 流式会话」——即**同一时刻只允许一个会话在流式**。
|
||||
* 带 `sessionId` 的是 `history` / `stream_end` / `message_persisted` / `channel_persisted`
|
||||
* 以及编排状态帧(`session_busy` / `pending_*` / `session_idle` / `session_interrupted`)。
|
||||
* ## 会话归属
|
||||
* 服务端自 **v0.36** 起给**每个结构化流帧的帧壳**都打上 `sessionId` + `turnId` + `streamId`
|
||||
* (打戳处 `internal/runtime/loop/stream.go`,上帧处 `gateway/ws_hub_turn.go` 的
|
||||
* `broadcastStream`),`stream_end` / `stream_chunk` / 状态帧同样带 `SessionID` + `TurnID`。
|
||||
* 本层把帧壳上的 `sessionId` **原样透传**到每个事件类([WsEvent.sessionId]),**不补值**。
|
||||
*
|
||||
* 服务端广播**不按会话订阅**(hub 把帧发给每条已认证连接),因此「这一帧属于哪个会话」只能由
|
||||
* 客户端按 `sessionId` 收敛——见 [isForSession]。该字段可能为 `null`(帧壳省略,或该类型本就
|
||||
* 无会话归属),此时是「无法判定」而非「不属于本会话」,上层应放行并自行兜底。
|
||||
*
|
||||
* ## 未知事件与坏帧
|
||||
* [parse] 对未知 `type`、非 JSON、结构不匹配的帧一律返回 `null`(安全忽略),绝不抛异常。
|
||||
@@ -35,7 +38,10 @@ sealed class WsEvent {
|
||||
/** hub 级单调递增序号,供排序关联流式突发。 */
|
||||
open val seq: Long? = null
|
||||
|
||||
/** 能确定会话的事件才有;流式正文/思维链/工具帧恒为 `null`。 */
|
||||
/**
|
||||
* 本帧归属的会话(帧壳上的 `sessionId`,原样透传)。
|
||||
* `null` = 帧壳没给,**不代表不属于本会话**——见 [isForSession]。
|
||||
*/
|
||||
open val sessionId: String? = null
|
||||
|
||||
/** 连接首帧:主会话最近 50 条历史(`thinking` 已被服务端剥离)。 */
|
||||
@@ -53,6 +59,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "message.start"
|
||||
}
|
||||
@@ -63,6 +70,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "message.delta"
|
||||
}
|
||||
@@ -72,6 +80,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "message.end"
|
||||
}
|
||||
@@ -81,6 +90,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "thinking.start"
|
||||
}
|
||||
@@ -91,6 +101,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "thinking.delta"
|
||||
}
|
||||
@@ -100,6 +111,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "thinking.end"
|
||||
}
|
||||
@@ -110,6 +122,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_call.start"
|
||||
}
|
||||
@@ -121,6 +134,7 @@ sealed class WsEvent {
|
||||
val delta: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_call.delta"
|
||||
}
|
||||
@@ -131,6 +145,7 @@ sealed class WsEvent {
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_call.end"
|
||||
}
|
||||
@@ -142,6 +157,7 @@ sealed class WsEvent {
|
||||
val delta: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_execution.update"
|
||||
}
|
||||
@@ -329,17 +345,22 @@ sealed class WsEvent {
|
||||
data: JsonElement?,
|
||||
): WsEvent? = when (type) {
|
||||
"history" -> History(decode<HistoryData>(data).messages, turnId, seq, sessionId)
|
||||
"message.start" -> MessageStart(decode<IndexData>(data).index, turnId, seq)
|
||||
"message.delta" -> decode<DeltaIndexData>(data).let { MessageDelta(it.delta, it.index, turnId, seq) }
|
||||
"message.end" -> MessageEnd(decode<IndexData>(data).index, turnId, seq)
|
||||
"thinking.start" -> ThinkingStart(decode<IndexData>(data).index, turnId, seq)
|
||||
"thinking.delta" -> decode<DeltaIndexData>(data).let { ThinkingDelta(it.delta, it.index, turnId, seq) }
|
||||
"thinking.end" -> ThinkingEnd(decode<IndexData>(data).index, turnId, seq)
|
||||
"tool_call.start" -> decode<ToolCallIdData>(data).let { ToolCallStart(it.id, it.index, turnId, seq) }
|
||||
"tool_call.delta" -> decode<ToolCallData>(data).let { ToolCallDelta(it.id, it.index, it.delta, turnId, seq) }
|
||||
"tool_call.end" -> decode<ToolCallIdData>(data).let { ToolCallEnd(it.id, it.index, turnId, seq) }
|
||||
"message.start" -> MessageStart(decode<IndexData>(data).index, turnId, seq, sessionId)
|
||||
"message.delta" -> decode<DeltaIndexData>(data)
|
||||
.let { MessageDelta(it.delta, it.index, turnId, seq, sessionId) }
|
||||
"message.end" -> MessageEnd(decode<IndexData>(data).index, turnId, seq, sessionId)
|
||||
"thinking.start" -> ThinkingStart(decode<IndexData>(data).index, turnId, seq, sessionId)
|
||||
"thinking.delta" -> decode<DeltaIndexData>(data)
|
||||
.let { ThinkingDelta(it.delta, it.index, turnId, seq, sessionId) }
|
||||
"thinking.end" -> ThinkingEnd(decode<IndexData>(data).index, turnId, seq, sessionId)
|
||||
"tool_call.start" -> decode<ToolCallIdData>(data)
|
||||
.let { ToolCallStart(it.id, it.index, turnId, seq, sessionId) }
|
||||
"tool_call.delta" -> decode<ToolCallData>(data)
|
||||
.let { ToolCallDelta(it.id, it.index, it.delta, turnId, seq, sessionId) }
|
||||
"tool_call.end" -> decode<ToolCallIdData>(data)
|
||||
.let { ToolCallEnd(it.id, it.index, turnId, seq, sessionId) }
|
||||
"tool_execution.update" -> decode<ToolCallData>(data)
|
||||
.let { ToolExecutionUpdate(it.id, it.index, it.delta, turnId, seq) }
|
||||
.let { ToolExecutionUpdate(it.id, it.index, it.delta, turnId, seq, sessionId) }
|
||||
"stream_chunk" -> StreamChunk(decode<StreamChunkData>(data).delta, turnId, seq, sessionId)
|
||||
"stream_end" -> decode<StreamEndData>(data)
|
||||
.let { StreamEnd(it.stopReason, it.steps, it.error, turnId, seq, sessionId) }
|
||||
@@ -389,6 +410,25 @@ sealed class WsEvent {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 这一帧是否属于 [sessionId] 这个会话(纯函数)。
|
||||
*
|
||||
* 服务端广播不按会话订阅,**每条**连接都会收到**每个**会话的帧,所以调用方必须先过这一闸,
|
||||
* 否则别的会话的 `message.delta` 会被自己的流式状态机吃掉。
|
||||
*
|
||||
* 判定原则是**只排除「明确属于别的会话」的帧**:
|
||||
* - 帧壳上的 `sessionId` 为空(`null` / 空白)→ **放行**。空值是「无法判定」而不是「不属于本会话」,
|
||||
* 丢掉会让旧服务端(或不带该字段的帧类型)整类事件消失——这与
|
||||
* [top.yeij.cyrene.data.repository.historyForSession] 对空 `sessionId` 的处理一致。
|
||||
* - 有值且相等 → 放行;有值且不等 → 丢弃。
|
||||
*
|
||||
* 不修改入参,无副作用。
|
||||
*/
|
||||
fun WsEvent.isForSession(sessionId: String): Boolean {
|
||||
val frameSession = this.sessionId
|
||||
return frameSession.isNullOrBlank() || frameSession == sessionId
|
||||
}
|
||||
|
||||
private fun JsonElement?.asStringOrNull(): String? {
|
||||
val primitive = this as? JsonPrimitive ?: return null
|
||||
if (primitive is JsonNull) return null
|
||||
|
||||
@@ -16,6 +16,7 @@ import top.yeij.cyrene.data.repository.mergeMessages
|
||||
import top.yeij.cyrene.data.ws.ChatSocket
|
||||
import top.yeij.cyrene.data.ws.WsEvent
|
||||
import top.yeij.cyrene.data.ws.WsState
|
||||
import top.yeij.cyrene.data.ws.isForSession
|
||||
import top.yeij.cyrene.domain.parse.ParsedBubble
|
||||
import top.yeij.cyrene.domain.parse.StreamingContentParser
|
||||
|
||||
@@ -93,6 +94,9 @@ data class ChatUiState(
|
||||
* 2. WS 首帧 `history`
|
||||
* 3. WS 增量 `message_persisted`(含乐观消息归一)
|
||||
* 4. WS 流式 `thinking.* / tool_call.* / message.* / stream_end`(临时气泡,落库后由 3 取代)
|
||||
*
|
||||
* 本 VM 只收**属于 [sessionId] 的帧**(见 [onEvent] 的会话闸)。服务端广播不按会话订阅,
|
||||
* 且每个会话各有一个 VM(各自的 socket 都还活着),没有这道闸就会串台。
|
||||
*/
|
||||
class ChatViewModel(
|
||||
private val container: AppContainer,
|
||||
@@ -172,6 +176,13 @@ class ChatViewModel(
|
||||
}
|
||||
|
||||
private fun onEvent(event: WsEvent) {
|
||||
// 广播是发给每条已认证连接的(服务端没有按会话订阅),所以每个 VM 都会收到**所有**会话的帧。
|
||||
// 帧壳自服务端 v0.36 起带 `sessionId`,先按它收敛:明确属于别的会话 → 整帧丢掉,
|
||||
// 否则上一会话的 VM 会拿新会话的 delta 跑自己的流式状态机(思维链/工具卡/正文串台)。
|
||||
// `history` 不走这道闸——它的归属由帧内**每条消息**的 sessionId 判定(见下面的
|
||||
// `historyForSession`),逐条过滤比帧壳更细,也更耐「将来服务端给次级会话回放」。
|
||||
if (event !is WsEvent.History && !event.isForSession(sessionId)) return
|
||||
|
||||
when (event) {
|
||||
is WsEvent.History -> {
|
||||
// 这一帧是**主会话**的回放,却会被投给任意 ChatViewModel——必须先收敛到本会话,
|
||||
|
||||
Reference in new issue
Block a user