merge: W57 —— 把「她主动说的话」的通知消费点提到进程级(新增 ProactiveFrameIntake;修 Activity 被划掉后收不到帧/不弹通知)

This commit is contained in:
AskaEth committed 2026-10-06 22:22:26 +08:00
commit 900d191b8d
6 files changed
+352 -38

No files matched your search

@@ -65,8 +65,8 @@ class AppContainer(context: Context) {
/**
* 「她主动说的话」的通知通路(W53N):判据、去重与落地都在里面,Android 那一步从这里注入。
*
* 它**不订阅任何东西**(帧由 `SessionsViewModel` 那条既有订阅喂进来),因此与连接的所有权
* 完全无关——本端那条共享连接的生死仍然只在登录态那一处表达。
* 它**不订阅任何东西**(帧由 [proactiveFrameIntake] 喂进来),因此与连接的所有权完全无关
* ——本端那条共享连接的生死仍然只在登录态那一处表达。
*/
val proactiveNotifier: ProactiveNotifier = ProactiveNotifier(
presence = appPresence,
@@ -287,4 +287,22 @@ class AppContainer(context: Context) {
onAuthFailed = authManager::onSessionExpired,
)
}
/**
* 「她主动说的话 ⇒ 弹通知」那条通路在**帧这一层的消费点**(W57,**进程级**):把 [chatSocket]
* 上那两种帧交给 [proactiveNotifier]。构造即订阅,此后与进程同寿(没有 start / stop 的理由
* 见 [ProactiveFrameIntake] 的类注释)。
*
* 改前这份订阅挂在 `SessionsViewModel` 的 `viewModelScope` 上(它随 Activity 一起结束),于是
* 「从最近任务划掉 App、进程还被 `KeepAliveService` 撑着」那一档:帧到了 [chatSocket]、
* 一个订阅者都没有,而 `events` 是 `replay = 0` 的 `SharedFlow` ⇒ 当场丢掉,通知永远不弹。
*
* 声明在这里(类的末尾)有一个顺序上的理由:它要同时用到 [proactiveNotifier] 与 [chatSocket],
* 而 [chatSocket] 是本类最后声明的那个(Kotlin 按声明顺序初始化)。读 [ChatSocket.events] 会把
* 那条连接对象提前建出来——本容器与它同时诞生、同时活着,这一步没有任何代价。
*/
val proactiveFrameIntake: ProactiveFrameIntake = ProactiveFrameIntake(
events = chatSocket.events,
notifier = proactiveNotifier,
)
}
@@ -0,0 +1,136 @@
package top.yeij.cyrene.app
import android.util.Log
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.launch
import top.yeij.cyrene.BuildConfig
import top.yeij.cyrene.core.notify.ProactiveNotifier
import top.yeij.cyrene.data.ws.WsEvent
private const val LOG_TAG = "ProactiveFrameIntake"
/** 一帧「她主动说的话」剥出来的东西:[ProactiveNotifier.onProactiveFrame] 的四个入参 + 会话归属。 */
internal data class ProactiveFrame(
val sessionId: String?,
val id: String?,
val messageId: String?,
val timestamp: String?,
val content: String,
)
/**
* 这一帧算不算「她主动说的话」,以及那四个入参各取帧上的哪个键——**纯函数**(离线可穷举)。
*
* 只有两种帧是她主动说的话,它们对应服务端两条路(见 [ProactiveNotifier]):
* 1. [WsEvent.ProactiveMessage](实时帧 `proactive_message`)——`id` 取落库行键 `data.id`,`messageId`
* 取 IM 侧消息号,两者**不是同一个 id**(见那个事件的 KDoc);
* 2. [WsEvent.MessagePersisted] **且** `proactive == true`(IM 桥那条路补发的落库帧)——这一档只有
* 落库行的键(`message.id`),另外两个键帧上就没有,所以恒给 `null`,一个字都不许编。
*
* 其余帧一律 `null`:普通落库广播、流式增量、别的事件都不是「她主动找我说话」。判据错在这一层的
* 两个方向都很难在真机上发现——要么她开口了却什么都不弹(要等下一次主动消息才看得见),要么普通
* 消息也弹(每一次她正常回话都响)。
*/
internal fun proactiveFrameOf(event: WsEvent): ProactiveFrame? = when (event) {
is WsEvent.ProactiveMessage -> ProactiveFrame(
sessionId = event.sessionId,
id = event.id,
messageId = event.messageId,
timestamp = event.timestamp,
content = event.content,
)
is WsEvent.MessagePersisted -> {
if (!event.proactive) {
// 普通落库广播(她回话、别的端发的消息都走这一帧):主人只批了 proactive 这一档。
null
} else {
ProactiveFrame(
sessionId = event.sessionId,
id = event.message?.id,
messageId = null,
timestamp = null,
content = event.message?.content.orEmpty(),
)
}
}
else -> null
}
/**
* 「她主动说的话 ⇒ 弹系统通知」那条通路的**帧消费点**(W57,**进程级**)。
*
* ## 为什么必须提到这一层(改前那一档是坏的)
* 这条通路的判据与去重一直是对的,错的是**订阅挂在哪**。改前它是 `SessionsViewModel` 里那句
* `viewModelScope.launch { socket.events.collect(::onEvent) }`。连接倒是进程级的——划掉 Activity
* 之后它照样连着、照样在收帧——但 `viewModelScope` 随 Activity 一起结束,而 `ChatSocket.events`
* 是 `replay = 0` 的 `SharedFlow`:**没有订阅者的 `tryEmit` 就是当场丢掉**(它不替将来的订阅者
* 留任何东西)。于是「从最近任务把 App 划掉、进程还被前台服务撑着」那一档,她的话进了连接、
* 然后没了。
*
* 本类因此只做一件事:**在进程级订那份帧流**,把 [proactiveFrameOf] 认出的帧交给 [ProactiveNotifier]。
* 它不建连接、不关连接、不认识登录态——连接的所有权一个字节都没动,仍然只在登录态那一处表达
* (`SessionsViewModel` 的 `LoggedIn → connect()` / `LoggedOut → close()`)。
*
* ## 为什么没有 start() / stop()
* 它的生命周期**就是进程**:`AppContainer` 在 `Application.onCreate` 里构造它,此后一直活着。
* 刻意不提供 `stop()` 正是这次要修的那个 bug——**唯一会想到去调 stop 的调用方(某个 ViewModel)
* 会随 Activity 一起死**,那等于把这条通路又绑回 Activity 上;留一个可以被忘掉调用的 `start()`
* 也只是多一种坏法(忘调 = 通知整个哑掉,且没有任何报错)。
*
* 它不需要停:帧只有连接活着才可能到,而连接在登出(含令牌失效那条经 `onAuthFailed` 的路径)时
* 由登录态关掉。「登出期间收不到帧」这件事由**连接**保证,不靠这里取消订阅。去重账本也**刻意不**
* 在这里清:它是 `ProactiveNotifier` 里的进程级账本,而换了账号之后同一条消息再推一次,仍是
* 同一个人的同一句话,不该弹第二遍。
*
* ## 它现在是这条通路上唯一的帧消费点
* 改前 `SessionsViewModel` 里那两处转交已经删掉(留两处就等于靠去重去替一次重复消费擦屁股,
* 而且「谁在喂通知」会有两个答案)。所以这条订阅一旦死掉,通知就**整个**哑掉——兜底见 [deliver]。
*
* @param events 进程内唯一那条连接的帧流(`ChatSocket.events`)。
* @param notifier 判据 + 去重 + 落地那一处。本类只喂帧,不替它判断任何事(那两个入参是它自己的
* 事实与动作,见它的 KDoc)。
* @param dispatcher 收集用的调度器;单测传 `Dispatchers.Unconfined`,让帧在同一次 `emit` 里走完。
*/
class ProactiveFrameIntake(
events: SharedFlow<WsEvent>,
private val notifier: ProactiveNotifier,
dispatcher: CoroutineDispatcher = Dispatchers.Default,
) {
/**
* 订阅本身那个 scope:私有,没有第二个人拿得到它,也就没有人能取消这条订阅——本类的全部意义
* 就在这里。`SupervisorJob` 只为 [deliver] 那条兜底留余地。
*/
private val scope = CoroutineScope(SupervisorJob() + dispatcher)
/** 构造即订阅(理由见类注释:启动时刻只有一个,留个能忘调的 `start()` 只会多一种坏法)。 */
init {
scope.launch { events.collect { event -> proactiveFrameOf(event)?.let(::deliver) } }
}
private fun deliver(frame: ProactiveFrame) {
try {
notifier.onProactiveFrame(
sessionId = frame.sessionId,
id = frame.id,
messageId = frame.messageId,
timestamp = frame.timestamp,
content = frame.content,
)
} catch (e: CancellationException) {
throw e
} catch (e: Throwable) {
// 落地那一步(建渠道 / notify)万一抛,**这条订阅必须活下来**:它是进程级那唯一一处、
// 不会再被重建,而不接住的话异常会打到线程的未捕获处理器(Android 上就是进程当场挂)。
// 代价封顶在「这一条通知没弹」——去重账本已经把它记下了,不会为了补它再弹一次。
if (BuildConfig.DEBUG) Log.d(LOG_TAG, "这条主动消息的通知没能落地", e)
}
}
}
@@ -21,7 +21,8 @@ import kotlinx.coroutines.flow.asStateFlow
* 一个布尔量在那里必然被旧实例的 `onStop` 记成「后台」,于是用户正看着聊天页,她一句话却弹出
* 通知来。按实例登记就没有这个方向性问题——新实例先加进来、旧实例后移出去,集合始终非空。
* 集合按 `ConcurrentHashMap` 语义(`Activity` 不重写 `equals` ⇒ 就是**同一实例**),
* 且读写可能不在同一线程上(帧在主线程的收集里读),所以用并发容器而不是普通 `MutableSet`。
* 且读写可能不在同一线程上(她的话那条通路是在**进程级订阅自己的调度器**上读它的,见
* `ProactiveFrameIntake`),所以用并发容器而不是普通 `MutableSet`。
*
* ## 它**不是**「App 有没有活着」
* 后台保活(前台服务)让进程活着、连接继续收帧,而本类在这里恒报「不在前台」——那正是通知要
@@ -32,10 +32,12 @@ enum class ProactiveNotifyOutcome {
/**
* 「她主动说的话」的通路(W53N):**判据 + 去重 + 落地**收在一处,Android 那一步由 [post] 注入。
*
* ## 它凭什么能看见两条路
* ## 谁把帧喂进来(W57 起)
* 后端两条路(实时帧 `proactive_message`、落库帧 `message_persisted{proactive:true}`)在客户端
* 汇于一处:`SessionsViewModel` 的帧订阅(它持有整个会话列表,两条帧它本来就都在消费)。本类
* 因此只是一个**纯逻辑 + 一个回调**,不订阅任何东西、不碰连接、不认识 Android。
* 汇于一处:**进程级**的帧消费点 `app.ProactiveFrameIntake` 订着那条共享连接,把这两种帧折成下面
* [onProactiveFrame] 的四个入参喂过来(同一条消息无论走哪条路、到几次,都只经过这一处)。
* 本类因此只是一个**纯逻辑 + 一个回调**,不订阅任何东西、不碰连接、不认识 Android——测试里传个
* 假的 [post] 就能把它整条驱动起来。
*
* ## 它绝不碰连接
* 连接的所有权在登录态(`SessionsViewModel` 那处 `LoggedIn → connect()`)。本类只在帧**已经到达
@@ -506,10 +506,11 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
* 帧、不碰下面的分支,所以两条既有路径(W49 的节流静默重拉、[persistedRoute] 的两档)语义
* 一个字节都没变。
*
* **她主动说的话还有第三个落点**(W53N):两条路(实时帧 / 带 `proactive` 的落库帧)**都**要
* 喂给 [top.yeij.cyrene.core.notify.ProactiveNotifier]——通知该不该弹、是不是重复由它判。
* 选在这里而不是新起一个订阅,是因为本 VM 是**唯一**同时看着这两帧、又活过整个登录态的
* 消费者;而它只把帧**转交**出去,不碰连接(连接的所有权仍在登录态那一处)。
* **通知不在这里**(W57 起):她主动说的话那两条路(实时帧 / 带 `proactive` 的落库帧)**曾经**
* 在本 VM 的帧订阅里转交给 `core.notify.ProactiveNotifier`,而这份订阅挂在 `viewModelScope` 上
* ——从最近任务把 Activity 划掉时它就没了,进程被前台服务保活着也收不到帧。那条通路现在由
* **进程级**的消费点喂(`app.ProactiveFrameIntake`,与那条共享连接同一层、同寿),本类**不再
* 判断通知的任何一件事**:这里只剩列表自己的两件事(未读、「谁在跑」)。
*/
private fun onEvent(event: WsEvent) {
onBusyFrame(event)
@@ -519,37 +520,11 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
// 落了新消息。哪一个会话都有事做:正在看的那个推进游标(消息已经显示在屏幕上了,
// 否则「自己正看着的会话」会攒出一个未读角标,契约 §3.9 的客户端处理建议),
// 别的会话则去问一次服务端(红点不该等用户开抽屉才出现)——两半都在 onMessagePersisted。
is WsEvent.MessagePersisted -> {
onMessagePersisted(event)
// 她主动说的话走 **IM 桥那条路由**时,服务端把它补发在这个落库帧上(帧里的
// `proactive` 判据,见 WsEvent.MessagePersisted)。那条路**不发** `proactive_message`,
// 所以不在这一档看一眼,通知就只剩一半入口(W53N)。
if (event.proactive) {
container.proactiveNotifier.onProactiveFrame(
sessionId = event.sessionId,
id = event.message?.id,
messageId = null,
timestamp = null,
content = event.message?.content.orEmpty(),
)
}
}
is WsEvent.MessagePersisted -> onMessagePersisted(event)
// 她主动说的话(v0.72)。服务端**不为这条路补** `message_persisted`(那条只有 IM 路由
// 才广播),所以上面那一档永远不会被它触发——不在这里补一句,主会话的红点会停在她
// 开口之前,要等用户自己开一次抽屉(那次静默重拉)才补上。
//
// 同一帧也是「要不要弹通知」的入口之一(另一条是上面那个落库帧):同一条消息两条路都
// 可能到,去重由 `core.notify.ProactiveNotifier` 按 id 管,这里不做任何判断。
is WsEvent.ProactiveMessage -> {
onProactiveMessage(event)
container.proactiveNotifier.onProactiveFrame(
sessionId = event.sessionId,
id = event.id,
messageId = event.messageId,
timestamp = event.timestamp,
content = event.content,
)
}
is WsEvent.ProactiveMessage -> onProactiveMessage(event)
else -> Unit
}
}
@@ -0,0 +1,182 @@
package top.yeij.cyrene.app
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.MutableSharedFlow
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNull
import org.junit.Test
import top.yeij.cyrene.core.notify.AppPresence
import top.yeij.cyrene.core.notify.ProactiveNotifier
import top.yeij.cyrene.data.api.dto.SessionSummaryDto
import top.yeij.cyrene.data.ws.WsEvent
/**
* 「她主动说的话」在**帧这一层**的消费点(W57)。
*
* 这里钉的是**接线**,不是判据(判据与去重仍在 `ProactiveNotifierTest`):
* 1. **它不需要任何 Activity 活着就能收帧**——改前那份订阅挂在 `SessionsViewModel` 的
* `viewModelScope` 上,从最近任务划掉 Activity 就没了,而 `ChatSocket.events` 是 `replay = 0`
* 的 `SharedFlow` ⇒ 没有订阅者的帧**当场丢掉**。本类的第一条测试就是那个 bug 的现场;
* 2. 哪些帧算「她主动说的话」、四个入参各取帧上的哪个键([proactiveFrameOf])——错在这两个方向
* 都是静默的(她开口了却不弹、普通消息也弹),所以逐键钉住;
* 3. 判据与去重在新的消费点上**没有被放松**:两条路仍只弹一次、看着主会话仍不弹、别的会话仍不弹。
*
* 收集用的调度器传 `Dispatchers.Unconfined`:`launch` 与 `tryEmit` 都在当前线程上把 collect 体
* 走完,于是测试是同步的(不需要等、也不需要 `runBlocking`)。
* **测不了的**:真·进程存活(Activity 没了而进程还在)这件事只有真机能造,本层只能钉住「订阅不挂
* 在任何会随 UI 结束的 scope 上」这个可测的部分。
*/
class ProactiveFrameIntakeTest {
private val presence = AppPresence()
private val posted = mutableListOf<String>()
/** 与 `ChatSocket` 同一档的订阅方式:无 replay、有缓冲、溢出丢最旧。 */
private val events = MutableSharedFlow<WsEvent>(
replay = 0,
extraBufferCapacity = EVENT_BUFFER,
onBufferOverflow = BufferOverflow.DROP_OLDEST,
)
/** 真实的帧(键名与类型都对服务端):实时帧带落库行 `id` + IM 号 + RFC3339 时间戳。 */
private val liveJson = """
{"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()
/** 同一条消息的另一条路:IM 桥那条路由补发的落库帧(判据键在载荷里)。 */
private val persistedProactiveJson = """
{"type":"message_persisted","sessionId":"web","data":{"proactive":true,"message":{"id":"1371","sessionId":"web","role":"assistant","content":"记得喝水哦","createdAt":1791212405123}}}
""".trimIndent()
/** 普通消息的落库广播(她正常回话):**不是**她主动说的话。 */
private val ordinaryPersistedJson = """
{"type":"message_persisted","sessionId":"web","data":{"message":{"id":"1372","sessionId":"web","role":"assistant","content":"在的","createdAt":1791212405123}}}
""".trimIndent()
/** 她主动说的、但不属主会话:绝不弹(宁可漏,也不要把别处的消息说成是她在找我说话)。 */
private val liveJsonForAnotherSession = """
{"type":"proactive_message","sessionId":"sess-abc","data":{"content":"在吗","messageId":"im-9","id":"1373"}}
""".trimIndent()
private fun intake() = ProactiveFrameIntake(
events = events,
notifier = ProactiveNotifier(presence = presence, post = { text, _ -> posted += text }),
dispatcher = Dispatchers.Unconfined,
)
private fun parse(json: String): WsEvent = requireNotNull(WsEvent.parse(json)) { "帧本身要能成形" }
// —— 这一档就是那个 bug 的现场 ——
@Test
fun `her words still notify when no activity is alive`() {
// Activity 已经从最近任务里划掉(没有任何实例在 started 档、聊天页也不在),进程被
// KeepAliveService 撑着、连接照旧在收帧。改前这里是空的:订阅随 VM 一起没了。
assertFalse("这一档的前提就是前台没有任何 Activity", presence.appInForeground)
intake()
events.tryEmit(parse(liveJson))
assertEquals(listOf("记得喝水哦"), posted)
}
@Test
fun `the notification id comes from the notifier not from this layer`() {
// 号必须按消息派生(同一条 ⇒ 同一个号、不同的句子各占一条)。这一层只转交帧,不许自己
// 造号——这里用一个把号也记下来的假 post 确认它拿到的确实是「按消息派生」的那个值。
val ids = mutableListOf<Int>()
ProactiveFrameIntake(
events = events,
notifier = ProactiveNotifier(presence = presence, post = { _, id -> ids += id }),
dispatcher = Dispatchers.Unconfined,
)
events.tryEmit(parse(liveJson))
events.tryEmit(parse(ordinaryPersistedJson))
assertEquals(1, ids.size)
}
// —— 判据与去重在新的消费点上没有被放松 ——
@Test
fun `the live frame and the persisted frame of one message notify once`() {
intake()
events.tryEmit(parse(liveJson))
// 同一条的落库帧(另一条路):认得出是同一条 ⇒ 不再弹。
events.tryEmit(parse(persistedProactiveJson))
assertEquals(listOf("记得喝水哦"), posted)
}
@Test
fun `a frame that lands on screen while the main session is open does not notify`() {
// 用户正看着她说话的那个会话:话就出现在他眼前,不打扰(判据本身在 ProactiveNotify.kt)。
presence.onActivityStarted(Any())
presence.onChatSurface(SessionSummaryDto.MAIN_SESSION_ID)
intake()
events.tryEmit(parse(liveJson))
assertEquals(emptyList<String>(), posted)
}
@Test
fun `a frame for another session never notifies`() {
intake()
events.tryEmit(parse(liveJsonForAnotherSession))
assertEquals(emptyList<String>(), posted)
}
@Test
fun `an ordinary persisted frame never notifies`() {
intake()
events.tryEmit(parse(ordinaryPersistedJson))
assertEquals(emptyList<String>(), posted)
}
// —— 哪些帧算她的话、四个入参各取自帧上的哪里 ——
@Test
fun `only the two proactive shapes are frames of her words`() {
assertNull("普通落库广播不是", proactiveFrameOf(parse(ordinaryPersistedJson)))
assertNull("流式增量不是", proactiveFrameOf(WsEvent.MessageDelta(delta = "在的", index = 0)))
assertNull("开始帧不是", proactiveFrameOf(WsEvent.MessageStart(index = 0)))
}
@Test
fun `the live frame takes the history row id and the im id apart`() {
val frame = requireNotNull(proactiveFrameOf(parse(liveJson)))
// 帧壳不带会话归属 ⇒「无法判定」,由「她主动说的话落主会话」这条事实兜住(见 ProactiveNotifier)。
assertNull(frame.sessionId)
assertEquals("1371", frame.id)
assertEquals("im-7", frame.messageId)
assertEquals("2026-10-06T01:20:05.123+08:00", frame.timestamp)
assertEquals("记得喝水哦", frame.content)
}
@Test
fun `the persisted frame has only the row id and never invents the other two`() {
val frame = requireNotNull(proactiveFrameOf(parse(persistedProactiveJson)))
assertEquals(SessionSummaryDto.MAIN_SESSION_ID, frame.sessionId)
assertEquals("1371", frame.id)
// 帧上就没有这两个键。编一个出来(比如把 `id` 也当 IM 号)就是伪造服务端键,去重会跟着漂。
assertNull(frame.messageId)
assertNull(frame.timestamp)
assertEquals("记得喝水哦", frame.content)
}
private companion object {
/** 与 `ChatSocket.EVENT_BUFFER` 同档:订阅方不设上限、发射方不阻塞。 */
const val EVENT_BUFFER = 8
}
}