merge: W50 —— 她主动说的话实时进主会话（proactive_message 消费 + 帧解码修复 timestamp 类型不符）

app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt

@@ -592,7 +592,21 @@ sealed class WsEvent {
         override val type: String get() = "voice_output"
     }
 
-    /** 自主主动推送（仅 owner/admin 连接收到）。 */
+    /**
+     * 自主主动推送（仅 owner/admin 连接收到）。
+     *
+     * v0.72 起她主动说的话**同时是主会话历史里的真消息**（`web` / `assistant` / `turn_id` 空），
+     * 载荷因此多带两个键：[id] / [createdAt] = 那条**落库行**的权威 id 与时间戳。
+     * 它与 [messageId]（IM 侧的消息号）是**两个不同的 id**，拿后者当去重键永远对不上历史行。
+     * 落库失败时这两个键**一起缺席**——那是「这条只有实时帧、没有历史行」的唯一标志，
+     * 消费方据此退化（见 [top.yeij.cyrene.ui.chat.proactiveIngestFor]）。
+     *
+     * [timestamp] 是服务端侧这条统一消息**自身**的时间。线上是 Go `time.Time` ⇒ **RFC3339 字符串**
+     * （数字形状也容忍）；它只作诊断，排序与去重一律用 [createdAt]（Unix 毫秒）。
+     *
+     * ⚠️ 帧壳**不带** `sessionId`（服务端构造这一帧时没设它，且该键是 `omitempty`）——所以
+     * [isForSession] 会把这帧**放行给每一个会话的 VM**，归属只能由消费方自己钉死（它属于主会话）。
+     */
     data class ProactiveMessage(
         val platform: String?,
         val channelId: String?,
@@ -600,8 +614,10 @@ sealed class WsEvent {
         val content: String,
         val contentType: String?,
         val sender: String?,
-        val timestamp: Long?,
+        val timestamp: String?,
         val messageId: String?,
+        val id: String? = null,
+        val createdAt: Long? = null,
         override val turnId: String? = null,
         override val seq: Long? = null,
         override val sessionId: String? = null,
@@ -839,10 +855,13 @@ sealed class WsEvent {
                 .let { MessageDeleted(it.id, turnId, seq, sessionId) }
             "voice_output" -> decode<VoiceOutputData>(data)
                 .let { VoiceOutput(it.audio, it.format, it.text, it.reply, turnId, seq, sessionId) }
+            // 她主动说的话（v0.72）。帧壳**没有** `sessionId`（服务端不设），与会话归属有关的一切
+            // 都在载荷里（落库行 id / 时间戳）+ 消费方自己钉「主会话」这条事实，见 [ProactiveMessage]。
             "proactive_message" -> decode<ProactiveData>(data).let {
                 ProactiveMessage(
                     it.platform, it.channelId, it.channelType, it.content, it.contentType,
-                    it.sender, it.timestamp, it.messageId, turnId, seq, sessionId,
+                    it.sender, it.timestamp.asStringOrNull(), it.messageId, it.id, it.createdAt,
+                    turnId, seq, sessionId,
                 )
             }
             // 定向错误帧（含队列满的伴生帧）：编码是开放集合，原样带上，未知码不做语义处置。
@@ -1044,6 +1063,16 @@ private data class VoiceOutputData(
     val reply: String? = null,
 )
 
+/**
+ * `proactive_message` 载荷（v0.72；字段名对着服务端 `gateway.proactivePayload`）。
+ *
+ * [timestamp] 声明成 [JsonElement] 而不是 `Long` 是**修一个把整帧弄没的解码缺陷**：
+ * 服务端那是 Go `time.Time`，线上是 RFC3339 **字符串**，而本端此前按 `Long?` 解 ⇒ 解码抛异常、
+ * 被 `parse` 的 `runCatching` 吞成 `null`，帧**在成形之前就没了**（`onEvent` 连 `else` 都到不了）。
+ * 收成元素之后按字面量取（数字形状也照样读出它），一个只作诊断的键从此不可能让整帧消失。
+ *
+ * [id] / [createdAt] 是落库行的权威 id 与时间戳，落库失败时**一起缺席**。
+ */
 @Serializable
 private data class ProactiveData(
     val platform: String? = null,
@@ -1052,8 +1081,10 @@ private data class ProactiveData(
     val content: String = "",
     val contentType: String? = null,
     val sender: String? = null,
-    val timestamp: Long? = null,
+    val timestamp: JsonElement? = null,
     val messageId: String? = null,
+    val id: String? = null,
+    val createdAt: Long? = null,
 )
 
 /**

app/src/main/kotlin/top/yeij/cyrene/ui/chat/ChatViewModel.kt

@@ -2561,6 +2561,10 @@ class ChatViewModel(
 
             is WsEvent.MessagePersisted -> onPersisted(event.message, event.turnId)
 
+            // 她主动说的话（v0.72）。上面那道会话闸对它是**空过**的（帧壳没有 `sessionId` ⇒
+            // `isForSession` 按「无法判定」放行给每一个会话的 VM），所以归属得由这一条自己钉。
+            is WsEvent.ProactiveMessage -> onProactiveMessage(event)
+
             // 灰态的**另一个入口**：这条消息被在飞的那一轮吃进去了（规范 §2.5）。
             // 帧壳 `turnId` 缺席时**不消费这一帧**——那样绑出来的绑定永远收不了口，
             // 宁可保持今天的样子（什么都不发生），也不留下一个永远灰着的气泡。
@@ -2812,6 +2816,33 @@ class ChatViewModel(
         }
     }
 
+    /**
+     * 她主动说的话（v0.72 `proactive_message`）：**实时**进本会话。
+     *
+     * 在这一条之前，这个帧在本端没有任何消费点 ⇒ 她说的话要等重连 / 重开 / 切会话拉历史才出现
+     * （服务端这条 Web 路由只发这一帧，**不补** `message_persisted`，所以也没有别的路能把它带进来）。
+     *
+     * 归属、身份、退化三件事全在纯函数 [proactiveIngestFor] 里（判据可单测）；这里只做它做不了的两件事：
+     * 把结果并进状态（与 [onPersisted] **同一条** [mergeMessages] 路，按 `id` upsert ⇒ 不会两条），
+     * 以及没有可并入的东西时**什么都不做**（[ProactiveIngest.Skip] 的两条成因见那里的 KDoc）。
+     *
+     * 不新造「只活在 UI 的气泡」的理由：那样刷新后她的话就没了，而刷新后仍在才是我们要的性质
+     * ——真相是服务端历史里的那一行，实时帧只是它早到一步的副本。
+     */
+    private fun onProactiveMessage(event: WsEvent.ProactiveMessage) {
+        val incoming = when (val ingest = proactiveIngestFor(event, sessionId, System.currentTimeMillis())) {
+            // 不是主会话的 VM（帧壳没有 sessionId，每个会话的 VM 都收到了这一帧）⇒ 这一帧到此为止。
+            ProactiveIngest.Skip -> return
+            // 已落库行：与 onPersisted 并入的是同一种东西，所以走同一个键（id）与同一条路。
+            is ProactiveIngest.Persisted -> ingest.message
+            // 落库失败那一档：只画一条本地气泡（刷新后不再有，见 Bubble 的 KDoc）。
+            is ProactiveIngest.Bubble -> ingest.message
+        }
+        _state.update { state ->
+            state.copy(messages = mergeMessages(state.messages, listOf(incoming)))
+        }
+    }
+
     /**
      * 落库消息到达：
      * - user：把对应的乐观气泡去掉——**首选 `clientMsgId`**（v0.56 起服务端回填），

app/src/main/kotlin/top/yeij/cyrene/ui/chat/ProactiveIngest.kt

@@ -0,0 +1,121 @@
+package top.yeij.cyrene.ui.chat
+
+import top.yeij.cyrene.data.api.dto.HistoryMessageDto
+import top.yeij.cyrene.data.ws.WsEvent
+
+/**
+ * 一帧 `proactive_message`（她主动说的话）**该怎么进本端**（纯函数 [proactiveIngestFor] 的结果）。
+ *
+ * 独立成一个类型而不是「返回一条 `HistoryMessageDto?`」，是因为两条并入路**外观相同、真相不同**：
+ * [Persisted] 那条是服务端历史里的行（按 id upsert，刷新后仍在），[Bubble] 那条只活在本端内存里。
+ * 上层要能对它们做不同的说明，测试也要能分别钉住。
+ */
+sealed interface ProactiveIngest {
+
+    /**
+     * **不并入本 VM**：这一帧不是给这个会话的（见 [proactiveIngestFor] 的归属判据），
+     * 或帧里没有任何可显示的东西。
+     */
+    data object Skip : ProactiveIngest
+
+    /**
+     * **并入一条已落库的 assistant 行**（帧带落库行的权威 id）。
+     *
+     * 这就是 v0.72 之后她这句话的真相：它既是实时帧，也是主会话历史里的一行。走 [Persisted]
+     * 而不是画一个气泡，客户端才有「刷新 / 重开 / 切会话拉历史之后她的话还在」这条性质——
+     * 而重复也**不会**发生：并入走的是以 [HistoryMessageDto.id] 为键的 upsert
+     * （`mergeMessages`），同一条（实时帧 + 握手回放 + 分页）无论到几次都只有一行。
+     */
+    data class Persisted(val message: HistoryMessageDto) : ProactiveIngest
+
+    /**
+     * **降级路径**：帧**不带** `id`（服务端那一步落库失败了，它自己就说了「这条没有历史行」）
+     * ⇒ 画一条只活在本端内存里的气泡，也好过什么都不显示。
+     *
+     * ⚠️ 它的代价必须说清：这条气泡**不在**服务端历史里，所以**刷新 / 重开 App / 切走再回来
+     * 都不会再有它**（那些时刻显示的是服务端的历史，而服务端没有这一行）。这是「如实显示一条
+     * 可能留不住的记录」而不是「假装它已落库」——正因如此它才与 [Persisted] 分开。
+     */
+    data class Bubble(val message: HistoryMessageDto) : ProactiveIngest
+}
+
+/**
+ * 本端给降级气泡铸的 id 前缀。
+ *
+ * 一眼可辨（日志 / 截图上能认出这是本地行，不是服务端 id），也**永不**与服务端的 id 撞。
+ */
+internal const val PROACTIVE_LOCAL_ID_PREFIX = "proactive-local:"
+
+/**
+ * 一帧 `proactive_message` 该不该并进 [sessionId] 这个 VM、以什么身份并入（纯函数，判据可单测）。
+ *
+ * ## 1. 归属：**只进主会话**（这一步不能省）
+ * 这一帧的帧壳**不带** `sessionId`（服务端构造时没设它，且 `ws_hub.go` 那个键是 `omitempty`），
+ * 而 [top.yeij.cyrene.data.ws.isForSession] 对空 `sessionId` 的口径是**放行**（空 = 无法判定，
+ * 不是「不属于本会话」）⇒ 同一批帧会投给**每一个**活着的 `ChatViewModel`。归属因此只能在这里
+ * 自己钉死：她主动说的话落的是**主会话**（服务端 `history.MainSessionID` = 客户端的
+ * [MAIN_SESSION_ID]）。不钉就会**在每个次级会话里都冒出一条她的话**——比「看不到」更糟。
+ *
+ * 比较是**等值**、不是前缀：主会话 `web` 与次级会话 `web:xxx` 是两个会话。
+ *
+ * ## 2. 身份：有 `id` 就是已落库行，没有就是本地气泡
+ * 服务端**先落库、再发帧**，并把落库行的 `id` / `createdAt` 放进载荷；落库失败时这两个键
+ * 一起缺席。所以 `id` 的有无就是「这条在历史里存不存在」的判据，不需要别的猜测。
+ *
+ * ## 3. `createdAt` 的退化
+ * `id` 在、`createdAt` 却缺席或非正数（形状漂移）时用 [nowMillis]：帧是刚刚到的，按「现在」
+ * 排在最末就是它该在的位置；而服务端那一行是真的存在，下一次拉历史会用同一 id 把它连同真正的
+ * 时间戳一起覆盖回来（`mergeMessages` 的 incoming 胜出）。
+ *
+ * ## 4. 无 `id` 且无正文 ⇒ [ProactiveIngest.Skip]
+ * 这种帧里没有任何可显示的东西（落库失败的空帧），画一条空气泡只会让人以为出了别的事。
+ *
+ * 时钟与本地 id 都由调用方喂（判据自己不读表、不自铸随机数），所以这里可单测、可复现。
+ */
+fun proactiveIngestFor(
+    event: WsEvent.ProactiveMessage,
+    sessionId: String,
+    nowMillis: Long,
+): ProactiveIngest {
+    if (sessionId != MAIN_SESSION_ID) return ProactiveIngest.Skip
+
+    val content = event.content
+    val id = event.id?.takeIf { it.isNotBlank() }
+    val createdAt = event.createdAt?.takeIf { it > 0 }
+
+    if (id != null) {
+        return ProactiveIngest.Persisted(
+            HistoryMessageDto(
+                id = id,
+                sessionId = MAIN_SESSION_ID,
+                role = ASSISTANT_ROLE,
+                content = content,
+                platform = event.platform?.takeIf { it.isNotBlank() },
+                createdAt = createdAt ?: nowMillis,
+            ),
+        )
+    }
+
+    if (content.isBlank()) return ProactiveIngest.Skip
+
+    // 本地 id 用**帧里已有的稳定键**（IM 侧的消息号，其次是它的时间字面量）而不是随机数：
+    // 同一个键第二次到达就是一次 upsert（仍只有一条），而且日志 / 截图上能追回它是哪一帧来的。
+    // 随机 id 做不到这两件事——同样的帧再投一次（服务端重推 / 客户端重放）就会多出一个一模一样的气泡。
+    val localKey = event.messageId?.takeIf { it.isNotBlank() }
+        ?: event.timestamp?.takeIf { it.isNotBlank() }
+        ?: nowMillis.toString()
+
+    return ProactiveIngest.Bubble(
+        HistoryMessageDto(
+            id = PROACTIVE_LOCAL_ID_PREFIX + localKey,
+            sessionId = MAIN_SESSION_ID,
+            role = ASSISTANT_ROLE,
+            content = content,
+            platform = event.platform?.takeIf { it.isNotBlank() },
+            createdAt = createdAt ?: nowMillis,
+        ),
+    )
+}
+
+/** 她主动说的话恒是 assistant（服务端 `persistProactiveToMain` 定死）。 */
+private const val ASSISTANT_ROLE = "assistant"

app/src/main/kotlin/top/yeij/cyrene/ui/sessions/SessionsViewModel.kt

@@ -457,6 +457,10 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
             // 否则「自己正看着的会话」会攒出一个未读角标，契约 §3.9 的客户端处理建议），
             // 别的会话则去问一次服务端（红点不该等用户开抽屉才出现）——两半都在 onMessagePersisted。
             is WsEvent.MessagePersisted -> onMessagePersisted(event)
+            // 她主动说的话（v0.72）。服务端**不为这条路补** `message_persisted`（那条只有 IM 路由
+            // 才广播），所以上面那一档永远不会被它触发——不在这里补一句，主会话的红点会停在她
+            // 开口之前，要等用户自己开一次抽屉（那次静默重拉）才补上。
+            is WsEvent.ProactiveMessage -> onProactiveMessage(event)
             else -> Unit
         }
     }
@@ -508,6 +512,38 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
         }
     }
 
+    /**
+     * 她主动说的话（v0.72 `proactive_message`）落在**主会话**——所以这一帧在本 VM 里问的是
+     * 「主会话那一行现在有没有未读」，与 [onMessagePersisted] 的两条出路**同一套口径**，
+     * 判据也共用 [persistedRoute]（只是会话 id 由这一帧自己钉：它不带 `sessionId`）。
+     *
+     * 为什么必须在这里补一句：服务端的主动推送 Web 路由只发这一帧，**不补** `message_persisted`
+     * （那条只有 IM 路由会广播），所以 W49 那条「非激活会话落新消息 ⇒ 节流静默重拉」根本不会被
+     * 触发——抽屉里主会话的红点就停在她开口之前，直到用户自己开一次抽屉。这里复用**同一条**路
+     * （[requestSilentRefresh] + 既有的 [SilentRefreshCause.InactiveSessionMessage]），
+     * 不新写取数路径，也不动节流判据本身。
+     *
+     * `id` 缺席 = 服务端那一刻**没有落库行** ⇒ 历史里没有这一行，未读面上没有任何东西要更新
+     * （本端也不替服务端数未读）。这一档的可见性由聊天页那条降级气泡负责，不归这里。
+     */
+    private fun onProactiveMessage(event: WsEvent.ProactiveMessage) {
+        if (event.id.isNullOrBlank()) return
+        val sessionId = SessionSummaryDto.MAIN_SESSION_ID
+        when (persistedRoute(sessionId, _state.value.activeSessionId)) {
+            PersistedRoute.InactiveSession ->
+                requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage)
+
+            // 屏幕上就是主会话：她这句话已经显示在聊天流里（ChatViewModel 那侧并入）⇒ 与
+            // message_persisted 的同一档一样「看到即已读」，免得抽屉为她刚说的话亮一个红点。
+            PersistedRoute.ActiveSession -> {
+                val before = event.createdAt?.takeIf { it > 0 }
+                before?.let { recordReadCursor(sessionId, it) }
+                _state.update { it.copy(sessions = it.sessions.markRead(sessionId)) }
+                pushReadCursor(sessionId, before)
+            }
+        }
+    }
+
     /**
      * 推进已读游标（D4）。**失败静默**：这是一次后台的「顺手记一下」，不是用户发起的操作——
      * 弹错误横幅会为一件用户没要求、也不影响使用的失败打扰他；而且服务端游标才是唯一真源，

app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventProactiveFrameTest.kt

@@ -0,0 +1,93 @@
+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
+
+/**
+ * `proactive_message` 帧的解码（v0.72：她主动说的话**同时**是主会话历史里的一行）。
+ *
+ * 这块此前**整帧解不出来**，而且症状是「静默」：载荷里的 `timestamp` 服务端是 Go `time.Time`
+ * ⇒ 线上是 RFC3339 **字符串**，本端却按 `Long?` 解 ⇒ 解码抛异常，被 [WsEvent.parse] 的
+ * `runCatching` 吞成 `null`。帧**在成形之前就没了**，所以连「加一条 `onEvent` 分支」那种修法都
+ * 碰不到它。下面第一条用的就是服务端的真实形状（`gateway.proactivePayload`）。
+ *
+ * 另外两条形状上的事实也钉在这里（它们各自都足以让这一帧走错）：
+ * - `id` / `createdAt` 与 `messageId` **不是同一个 id**：前者才是落库行（历史里那一行）的键；
+ * - 帧壳**不带** `sessionId` ⇒ [isForSession] 会对**每一个**会话的 VM 放行。
+ */
+class WsEventProactiveFrameTest {
+
+    /**
+     * 按服务端 `gateway.proactivePayload` 的**键名与类型**构造的帧（已落库那一档）：`timestamp`
+     * 是 Go `time.Time` ⇒ JSON 里的 RFC3339 字符串，`id` / `createdAt` 是落库行的权威值。
+     */
+    private val persistedFrame = """
+        {"type":"proactive_message","data":{"platform":"cyrene-backthink","channelId":"chan-1","channelType":"private","content":"记得喝水哦","contentType":"text","sender":"Cyrene","timestamp":"2026-10-06T01:20:05.123+08:00","messageId":"im-7","id":"1371","createdAt":1791212405123}}
+    """.trimIndent()
+
+    @Test
+    fun `the persisted frame decodes with the history row keys`() {
+        val event = WsEvent.parse(persistedFrame) as WsEvent.ProactiveMessage
+
+        assertEquals("proactive_message", event.type)
+        assertEquals("记得喝水哦", event.content)
+        assertEquals("cyrene-backthink", event.platform)
+        assertEquals("Cyrene", event.sender)
+        // 两个 id 是两个不同的东西：`id` 才是历史行的键，`messageId` 是 IM 侧的号。
+        assertEquals("1371", event.id)
+        assertEquals("im-7", event.messageId)
+        assertEquals(1791212405123L, event.createdAt)
+        // 时间戳按线上的字面量收（它只作诊断，排序用 createdAt）。
+        assertEquals("2026-10-06T01:20:05.123+08:00", event.timestamp)
+    }
+
+    @Test
+    fun `a numeric timestamp is tolerated instead of killing the frame`() {
+        // 一个只作诊断的键，无论线上给字符串还是数字，都不该有能力让**整帧**消失。
+        val frame = """
+            {"type":"proactive_message","data":{"content":"在的","timestamp":1791212405123}}
+        """.trimIndent()
+
+        val event = WsEvent.parse(frame) as WsEvent.ProactiveMessage
+
+        assertEquals("1791212405123", event.timestamp)
+        assertEquals("在的", event.content)
+    }
+
+    @Test
+    fun `a frame without a persisted row still forms with both row keys absent`() {
+        // 落库失败那一档：服务端**省略**这两个键（而不是给一个悬空的 id），客户端据此知道
+        // 「这条只有实时帧、没有历史行」。帧本身照旧成形——它还有正文要显示。
+        val frame = """
+            {"type":"proactive_message","data":{"content":"想你了","sender":"Cyrene","timestamp":"2026-10-06T01:20:05+08:00"}}
+        """.trimIndent()
+
+        val event = WsEvent.parse(frame) as WsEvent.ProactiveMessage
+
+        assertNull(event.id)
+        assertNull(event.createdAt)
+        assertEquals("想你了", event.content)
+    }
+
+    @Test
+    fun `the frame shell has no session id so the session gate lets every vm through`() {
+        // 这是「必须自己钉死主会话」的根据：帧壳不带 sessionId，而 isForSession 把空值当
+        // 「无法判定」**放行** ⇒ 不自己判归属，这一句话会在每个次级会话里各冒一条。
+        val event = WsEvent.parse(persistedFrame) as WsEvent.ProactiveMessage
+
+        assertNull(event.sessionId)
+        assertTrue(event.isForSession("web"))
+        assertTrue(event.isForSession("web:other"))
+    }
+
+    @Test
+    fun `an empty frame decodes to an empty message rather than throwing`() {
+        // 载荷缺席（服务端漂移）时也不抛：整帧留下、正文为空，上层按「没有可显示的东西」跳过。
+        val event = WsEvent.parse("""{"type":"proactive_message"}""") as WsEvent.ProactiveMessage
+
+        assertEquals("", event.content)
+        assertNull(event.id)
+    }
+}

app/src/test/kotlin/top/yeij/cyrene/ui/chat/ProactiveIngestTest.kt

@@ -0,0 +1,166 @@
+package top.yeij.cyrene.ui.chat
+
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import top.yeij.cyrene.data.api.dto.HistoryMessageDto
+import top.yeij.cyrene.data.repository.mergeMessages
+import top.yeij.cyrene.data.ws.WsEvent
+
+/**
+ * 「她主动说的话」该怎么进本端（W50）。
+ *
+ * 判据全在纯函数 [proactiveIngestFor] 里，这里钉住四条**每一条错了都肉眼可见**的边界：
+ * 1. **归属**：只进主会话。帧壳没有 `sessionId`（`isForSession` 因此对每个会话的 VM 放行），
+ *    所以「次级会话不并入」这条必须由我们自己成立——不成立就是她一句话在每个会话里各冒一条。
+ * 2. **身份**：带落库行 `id` ⇒ 按**已落库行**并入（不是 UI 气泡）；这正是「刷新后她的话还在」的根据。
+ * 3. **退化**：不带 `id`（服务端落库失败）⇒ 只画一条本地气泡；帧里什么可显示的都没有时不画空气泡。
+ * 4. **不重复**：同一句话（实时帧 + 重连重发 + 后来的历史页）按 `id` upsert，永远只有一行。
+ */
+class ProactiveIngestTest {
+
+    private fun frame(
+        content: String = "记得喝水哦",
+        id: String? = "1371",
+        createdAt: Long? = 1_791_212_405_123L,
+        messageId: String? = "im-7",
+        timestamp: String? = "2026-10-06T01:20:05.123+08:00",
+        platform: String? = "cyrene-backthink",
+    ) = WsEvent.ProactiveMessage(
+        platform = platform,
+        channelId = "chan-1",
+        channelType = "private",
+        content = content,
+        contentType = "text",
+        sender = "Cyrene",
+        timestamp = timestamp,
+        messageId = messageId,
+        id = id,
+        createdAt = createdAt,
+    )
+
+    // —— 归属：只进主会话 ——
+
+    @Test
+    fun `the main session folds the frame in as a persisted assistant row`() {
+        val ingest = proactiveIngestFor(frame(), MAIN_SESSION_ID, nowMillis = 999L)
+
+        val row = (ingest as ProactiveIngest.Persisted).message
+        assertEquals("1371", row.id)
+        assertEquals("web", row.sessionId)
+        assertEquals("assistant", row.role)
+        assertEquals("记得喝水哦", row.content)
+        assertEquals(1_791_212_405_123L, row.createdAt)
+        // 帧里带的平台与落库行是同一个值（服务端把 msg.Platform 同时写进帧与行）。
+        assertEquals("cyrene-backthink", row.platform)
+    }
+
+    @Test
+    fun `a secondary session never folds the frame in`() {
+        // 帧壳没有 sessionId（服务端不设）⇒ isForSession 会把这一帧放行给每一个会话的 VM。
+        // 不在这里挡住，她主动说的话就会在每个次级会话里都冒一条——比看不到更糟。
+        assertEquals(
+            ProactiveIngest.Skip,
+            proactiveIngestFor(frame(), "web:7f3a", nowMillis = 999L),
+        )
+    }
+
+    @Test
+    fun `session ids are compared verbatim, not by prefix`() {
+        // 主会话 `web` 与次级会话 `web:xxx` 是两个会话（前缀相同）：一条她的话不该进后者。
+        assertEquals(ProactiveIngest.Skip, proactiveIngestFor(frame(), "web:1", nowMillis = 999L))
+        assertTrue(proactiveIngestFor(frame(), "web", nowMillis = 999L) is ProactiveIngest.Persisted)
+    }
+
+    // —— 身份：有 id = 已落库行 ——
+
+    @Test
+    fun `a missing createdAt falls back to now and is corrected by the next history pull`() {
+        // 两个键本该一起出现；只缺时间戳时按「现在」排在最末（帧是刚到的，那正是它该在的位置），
+        // 而服务端那一行真的存在 ⇒ 下一次拉历史会用同一个 id 把它连同真时间戳一起覆盖回来。
+        val ingest = proactiveIngestFor(frame(createdAt = null), MAIN_SESSION_ID, nowMillis = 4_242L)
+
+        assertEquals(4_242L, (ingest as ProactiveIngest.Persisted).message.createdAt)
+        assertEquals("1371", ingest.message.id)
+    }
+
+    @Test
+    fun `a zero createdAt is treated as absent`() {
+        // 服务端零值 / 客户端解不出来都会走到这里：`0` 不是「1970 年」，是「没有这个键」。
+        val ingest = proactiveIngestFor(frame(createdAt = 0L), MAIN_SESSION_ID, nowMillis = 7L)
+
+        assertEquals(7L, (ingest as ProactiveIngest.Persisted).message.createdAt)
+    }
+
+    // —— 退化：落库失败那一档 ——
+
+    @Test
+    fun `without a row id the frame degrades to a local-only bubble`() {
+        val ingest = proactiveIngestFor(
+            frame(id = null, createdAt = null, messageId = "im-9"),
+            MAIN_SESSION_ID,
+            nowMillis = 1_234L,
+        )
+
+        val bubble = (ingest as ProactiveIngest.Bubble).message
+        // 绝不能拿 `messageId` 当历史行的键（它是另一个 id，永远对不上历史）——本地的键是自铸的。
+        assertEquals("proactive-local:im-9", bubble.id)
+        assertEquals("assistant", bubble.role)
+        assertEquals("web", bubble.sessionId)
+        assertEquals("记得喝水哦", bubble.content)
+        assertEquals(1_234L, bubble.createdAt)
+    }
+
+    @Test
+    fun `a redelivered bubble-less frame does not turn into two bubbles`() {
+        // 同一帧再投一次（服务端重推 / 客户端重放）不该变成两个气泡：本地键取自帧里已有的稳定键。
+        val first = proactiveIngestFor(frame(id = null, createdAt = null), MAIN_SESSION_ID, 1L)
+        val again = proactiveIngestFor(frame(id = null, createdAt = null), MAIN_SESSION_ID, 2L)
+
+        val a = (first as ProactiveIngest.Bubble).message
+        val b = (again as ProactiveIngest.Bubble).message
+        assertEquals(a.id, b.id)
+
+        val merged = mergeMessages(listOf(a), listOf(b))
+        assertEquals(1, merged.size)
+    }
+
+    @Test
+    fun `a blank local id is treated as absent`() {
+        // 形状漂移（空串）与「没有这个键」必须是同一个结论，否则会并入一行空 id 的历史行。
+        val ingest = proactiveIngestFor(frame(id = "   "), MAIN_SESSION_ID, nowMillis = 8L)
+
+        assertTrue(ingest is ProactiveIngest.Bubble)
+        assertEquals("proactive-local:im-7", (ingest as ProactiveIngest.Bubble).message.id)
+    }
+
+    @Test
+    fun `a frame with neither a row id nor any text is skipped instead of drawn empty`() {
+        // 落库失败的空帧：画一条空气泡只会让人以为出了别的事（它没有任何可显示的内容）。
+        assertEquals(
+            ProactiveIngest.Skip,
+            proactiveIngestFor(frame(content = "  ", id = null, createdAt = null), MAIN_SESSION_ID, 9L),
+        )
+    }
+
+    // —— 不重复：按 id upsert ——
+
+    @Test
+    fun `the live frame and the same row from history stay one message`() {
+        val live = (proactiveIngestFor(frame(), MAIN_SESSION_ID, nowMillis = 1L) as ProactiveIngest.Persisted)
+            .message
+        // 随后到达的历史页里是同一行（服务端 id 相同），内容与时间戳更完整。
+        val fromHistory = HistoryMessageDto(
+            id = "1371",
+            sessionId = "web",
+            role = "assistant",
+            content = "记得喝水哦",
+            createdAt = 1_791_212_405_123L,
+        )
+
+        val merged = mergeMessages(listOf(live), listOf(fromHistory))
+
+        assertEquals(1, merged.size)
+        assertEquals("1371", merged.single().id)
+    }
+}

app/src/test/kotlin/top/yeij/cyrene/ui/sessions/InactiveSessionPullTest.kt

@@ -5,6 +5,7 @@ import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNull
 import org.junit.Assert.assertTrue
 import org.junit.Test
+import top.yeij.cyrene.data.api.dto.SessionSummaryDto
 
 /**
  * 非激活会话的新消息 ⇒ 节流地静默重拉（W49 修的 S6：「停在主会话时次级会话来了新消息，
@@ -55,6 +56,24 @@ class InactiveSessionPullTest {
         )
     }
 
+    @Test
+    fun `her proactive words pull the list while another session is open`() {
+        // W50：`proactive_message` 帧壳**不带** `sessionId`（服务端构造这一帧时没设它），归属由
+        // 客户端自己钉——她主动说的话恒落**主会话**（服务端 `history.MainSessionID`）。
+        // 而这条路服务端**不补** `message_persisted`（那条只有 IM 路由会广播）⇒ W49 那条重拉
+        // 不会被它触发，抽屉里主会话的红点会停在她开口之前。SessionsViewModel 用同一个
+        // `persistedRoute` 把这一帧接进同一条路（不新写取数路径），这两条断言就是那处接线的前提：
+        // 停在次级会话 ⇒ 该拉；正看着主会话 ⇒ 不拉（她的话已经在聊天流里，看到即已读）。
+        assertEquals(
+            PersistedRoute.InactiveSession,
+            persistedRoute(sessionId = SessionSummaryDto.MAIN_SESSION_ID, activeSessionId = "web:1"),
+        )
+        assertEquals(
+            PersistedRoute.ActiveSession,
+            persistedRoute(sessionId = SessionSummaryDto.MAIN_SESSION_ID, activeSessionId = "web"),
+        )
+    }
+
     // —— 节流：该拉的那一刻拉，窗口内不重复拉 ——
 
     @Test