diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/ws/ChatSocket.kt b/app/src/main/kotlin/top/yeij/cyrene/data/ws/ChatSocket.kt index e86c1e2..67d1fd9 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/ws/ChatSocket.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/ws/ChatSocket.kt @@ -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 足够)。 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 941cceb..0ffd546 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 @@ -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(data).messages, turnId, seq, sessionId) - "message.start" -> MessageStart(decode(data).index, turnId, seq) - "message.delta" -> decode(data).let { MessageDelta(it.delta, it.index, turnId, seq) } - "message.end" -> MessageEnd(decode(data).index, turnId, seq) - "thinking.start" -> ThinkingStart(decode(data).index, turnId, seq) - "thinking.delta" -> decode(data).let { ThinkingDelta(it.delta, it.index, turnId, seq) } - "thinking.end" -> ThinkingEnd(decode(data).index, turnId, seq) - "tool_call.start" -> decode(data).let { ToolCallStart(it.id, it.index, turnId, seq) } - "tool_call.delta" -> decode(data).let { ToolCallDelta(it.id, it.index, it.delta, turnId, seq) } - "tool_call.end" -> decode(data).let { ToolCallEnd(it.id, it.index, turnId, seq) } + "message.start" -> MessageStart(decode(data).index, turnId, seq, sessionId) + "message.delta" -> decode(data) + .let { MessageDelta(it.delta, it.index, turnId, seq, sessionId) } + "message.end" -> MessageEnd(decode(data).index, turnId, seq, sessionId) + "thinking.start" -> ThinkingStart(decode(data).index, turnId, seq, sessionId) + "thinking.delta" -> decode(data) + .let { ThinkingDelta(it.delta, it.index, turnId, seq, sessionId) } + "thinking.end" -> ThinkingEnd(decode(data).index, turnId, seq, sessionId) + "tool_call.start" -> decode(data) + .let { ToolCallStart(it.id, it.index, turnId, seq, sessionId) } + "tool_call.delta" -> decode(data) + .let { ToolCallDelta(it.id, it.index, it.delta, turnId, seq, sessionId) } + "tool_call.end" -> decode(data) + .let { ToolCallEnd(it.id, it.index, turnId, seq, sessionId) } "tool_execution.update" -> decode(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(data).delta, turnId, seq, sessionId) "stream_end" -> decode(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 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 6ff0419..d7ee304 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 @@ -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——必须先收敛到本会话, diff --git a/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventSessionScopeTest.kt b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventSessionScopeTest.kt new file mode 100644 index 0000000..80b8bdc --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventSessionScopeTest.kt @@ -0,0 +1,250 @@ +package top.yeij.cyrene.data.ws + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * A1 顺带项:**结构化流帧的 `sessionId` 必须被解码出来**,并按会话过滤。 + * + * 帧形状不是手写猜的: + * - 帧壳(`type` / `turnId` / `seq` / `sessionId` / `data`)取自 `gateway/wsEvent` 的 + * JSON tag 与 `ws_hub_stream.go` 的 `broadcast`;`history` 那一帧是**线上抓的** + * (`http://127.0.0.1:10942`,只连不写)。 + * - 每类 `data` 取自 `gateway/ws_bridge.go:streamEventData`(`streamId` 只在非空时并进 + * `data`,这里保留它,因为真实线上就是带的)。 + * - 「结构化流帧带 sessionId」由后端端到端测试钉死 + * (`gateway/ws_hub_stream_scope_test.go:TestStreamFramesCarrySessionScope`, + * 它解的就是连接上的原始 JSON)。 + */ +class WsEventSessionScopeTest { + + /** 帧壳上的一轮 id,形如 `turn-` + 8 字节 hex(`ws_hub_turn.go:newTurnID`)。 */ + private val turnId = "turn-9f2c1a4b7d0e3355" + + private fun frame( + type: String, + data: String? = null, + sessionId: String? = "web", + turnId: String? = this.turnId, + seq: Int? = 41, + ): String = buildString { + append("""{"type":"$type"""") + if (turnId != null) append(""","turnId":"$turnId"""") + if (seq != null) append(""","seq":$seq""") + if (sessionId != null) append(""","sessionId":"$sessionId"""") + if (data != null) append(""","data":$data""") + append("}") + } + + private fun parse(raw: String): WsEvent = + requireNotNull(WsEvent.parse(raw)) { "帧没解出来:$raw" } + + // —— 正文 —— + + @Test + fun `message start carries the session from the frame shell`() { + val ev = parse(frame("message.start", """{"index":0,"streamId":"s-6b1f"}""")) + as WsEvent.MessageStart + + assertEquals(0, ev.index) + assertEquals("web", ev.sessionId) + assertEquals(turnId, ev.turnId) + assertEquals(41L, ev.seq) + } + + @Test + fun `message delta carries the session from the frame shell`() { + val ev = parse(frame("message.delta", """{"delta":"Hel","index":0,"streamId":"s-6b1f"}""", seq = 42)) + as WsEvent.MessageDelta + + assertEquals("Hel", ev.delta) + assertEquals("web", ev.sessionId) + } + + @Test + fun `message end carries the session from the frame shell`() { + val ev = parse(frame("message.end", """{"index":0,"streamId":"s-6b1f"}""", seq = 43)) + as WsEvent.MessageEnd + + assertEquals("web", ev.sessionId) + } + + // —— 思维链 —— + + @Test + fun `thinking frames carry the session from the frame shell`() { + val start = parse(frame("thinking.start", """{"index":0,"streamId":"s-6b1f"}""")) + as WsEvent.ThinkingStart + val delta = parse(frame("thinking.delta", """{"delta":"嗯","index":0,"streamId":"s-6b1f"}""")) + as WsEvent.ThinkingDelta + val end = parse(frame("thinking.end", """{"index":0,"streamId":"s-6b1f"}""")) + as WsEvent.ThinkingEnd + + assertEquals("web", start.sessionId) + assertEquals("web", delta.sessionId) + assertEquals("web", end.sessionId) + } + + // —— 工具调用 —— + + @Test + fun `tool call frames carry the session from the frame shell`() { + val start = parse(frame("tool_call.start", """{"id":"call_1","index":0,"streamId":"s-6b1f"}""")) + as WsEvent.ToolCallStart + val delta = parse( + frame("tool_call.delta", """{"id":"call_1","index":0,"delta":"{\"q\":","streamId":"s-6b1f"}"""), + ) as WsEvent.ToolCallDelta + val end = parse(frame("tool_call.end", """{"id":"call_1","index":0,"streamId":"s-6b1f"}""")) + as WsEvent.ToolCallEnd + + assertEquals("web", start.sessionId) + assertEquals("web", delta.sessionId) + assertEquals("web", end.sessionId) + } + + @Test + fun `tool execution update carries the session from the frame shell`() { + // tool_execution.* 的 data 用 `tool`/`phase` 而不是 `index`(ws_bridge.go:toolExecutionData)。 + val ev = parse( + frame( + "tool_execution.update", + """{"tool":"read_file","phase":"running","id":"call_1","delta":"12 行"}""", + ), + ) as WsEvent.ToolExecutionUpdate + + assertEquals("call_1", ev.id) + assertEquals("12 行", ev.delta) + assertEquals("web", ev.sessionId) + } + + // —— 整段文本 / 收尾 —— + + @Test + fun `stream chunk and stream end carry the session from the frame shell`() { + val chunk = parse(frame("stream_chunk", """{"delta":"整段"}""", seq = 44)) as WsEvent.StreamChunk + val end = parse(frame("stream_end", """{"stopReason":"stop","steps":1}""", seq = 45)) as WsEvent.StreamEnd + + assertEquals("web", chunk.sessionId) + assertEquals("整段", chunk.delta) + assertEquals("web", end.sessionId) + assertEquals("stop", end.stopReason) + } + + @Test + fun `every structured stream event type exposes the same frame session`() { + // 一次扫完全套:漏一个类型就会在真机上表现为「那一类帧串台」。 + val cases = mapOf( + "message.start" to """{"index":0}""", + "message.delta" to """{"delta":"a","index":0}""", + "message.end" to """{"index":0}""", + "thinking.start" to """{"index":0}""", + "thinking.delta" to """{"delta":"a","index":0}""", + "thinking.end" to """{"index":0}""", + "tool_call.start" to """{"id":"call_1","index":0}""", + "tool_call.delta" to """{"id":"call_1","index":0,"delta":"a"}""", + "tool_call.end" to """{"id":"call_1","index":0}""", + "tool_execution.update" to """{"tool":"read_file","phase":"running","id":"call_1","delta":"x"}""", + "stream_chunk" to """{"delta":"a"}""", + "stream_end" to """{"stopReason":"stop"}""", + ) + + for ((type, data) in cases) { + val ev = parse(frame(type, data, sessionId = "sess-9c3")) + assertEquals("$type 的 sessionId 透了没有?", "sess-9c3", ev.sessionId) + } + } + + // —— 向后兼容:帧壳不带 sessionId —— + + @Test + fun `frames without a session on the shell decode to a null session`() { + // 旧服务端(v0.36 之前)/ 连接级错误帧就是这样:字段整个缺席。 + val ev = parse(frame("message.delta", """{"delta":"hi","index":0}""", sessionId = null)) + as WsEvent.MessageDelta + + assertNull(ev.sessionId) + assertEquals("hi", ev.delta) + } + + @Test + fun `an explicit null session on the shell decodes to a null session`() { + val ev = parse("""{"type":"message.delta","sessionId":null,"data":{"delta":"hi","index":0}}""") + as WsEvent.MessageDelta + + assertNull(ev.sessionId) + } + + // —— 线上抓到的 history 帧(逐字照抄形状) —— + + @Test + fun `the captured live history frame decodes with its shell session`() { + // 2026-10-04 从 http://127.0.0.1:10942/ws/chat 只连不写抓下来的首帧, + // 内容换成合成值;`seq` / `turnId` 在真实帧里**根本不存在**。 + val live = """ + {"type":"history","sessionId":"web","data":{"messages":[ + {"id":"c78a10cdc3f141fdde4a4bb14179a55a","sessionId":"web","role":"assistant", + "content":"因为那是你呀,亲爱的♪","platform":"weixin-ilink", + "deviceId":"weixin-ilink:probe","adapterId":"weixin-ilink:05487d7f", + "createdAt":1788854366025}]}} + """.trimIndent() + + val ev = parse(live) as WsEvent.History + + assertEquals("web", ev.sessionId) + assertNull("真实 history 帧不带 seq", ev.seq) + assertNull("真实 history 帧不带 turnId", ev.turnId) + assertEquals(1, ev.messages.size) + assertEquals("web", ev.messages.first().sessionId) + } + + // —— 过滤规则(纯函数) —— + + @Test + fun `a frame for this session is accepted`() { + val ev = parse(frame("message.delta", """{"delta":"hi","index":0}""", sessionId = "sess-9c3")) + + assertTrue(ev.isForSession("sess-9c3")) + } + + @Test + fun `a frame for another session is rejected`() { + val ev = parse(frame("message.delta", """{"delta":"hi","index":0}""", sessionId = "web")) + + assertFalse(ev.isForSession("sess-9c3")) + // 主会话的 VM 反过来也不该吃次级会话的帧。 + assertTrue(ev.isForSession("web")) + } + + @Test + fun `a frame without a session is accepted by everyone`() { + // 空值 = 「无法判定」,不是「不属于本会话」:放行才能与修复前的行为一致。 + val ev = parse(frame("message.delta", """{"delta":"hi","index":0}""", sessionId = null)) + + assertTrue(ev.isForSession("web")) + assertTrue(ev.isForSession("sess-9c3")) + } + + @Test + fun `a blank session on the shell is treated as unknown and accepted`() { + val ev = parse(frame("message.delta", """{"delta":"hi","index":0}""", sessionId = "")) + + assertTrue(ev.isForSession("web")) + } + + @Test + fun `the session rule also covers the orchestration and error frames`() { + val busy = parse( + """{"type":"session_busy","sessionId":"sess-9c3","data":{"mode":"thinking"}}""", + ) as WsEvent.SessionBusy + val foreign = parse( + """{"type":"error","sessionId":"sess-9c3","data":{"message":"boom"}}""", + ) as WsEvent.Error + + assertTrue(busy.isForSession("sess-9c3")) + assertFalse(busy.isForSession("web")) + assertFalse(foreign.isForSession("web")) + } +}