fix(chat): 连接重建（后端重启）时把「进行中」全部收口，不再永久卡在思考/工具调用
改了什么 - 新增纯判据 `isReconnect(previous, current, everConnected)`（ui/chat/ReconnectReset.kt）： 只有「**曾经**到过 Connected 之后又回到 Connected」才算重建，第一次接上一个字节都不动。 主判据是 VM 自己那位只置不翻的 `everConnected`（状态流会合流，`Connected → Reconnecting → Connected` 中间那一跳可能被吞掉，只置不翻的位不受影响）；第二证据是上一跳为 `Reconnecting` （只在「本 VM 是离线期间建起来的」时成立，且那一刻它手里什么都没有，收口是恒等变换）。 - 新增 `ChatUiState.settledAfterReconnect()`（纯函数）：把服务端那一轮在本端留下的五个面一次收口 —— `turns.allClosed()`（未收口绑定全标 closed，气泡 ④ 灰态 → ⑤）、`streams.allFinished()` （每一路停止生长：思维链不再「思考中…」、工具卡不再「执行 X…」）、`busyMode = null`、 `activeTurns = 0`、`queueDepth = 0`。无变化时返回**同一个实例**。 - `TurnStreams.kt` 新增 `allFinished()`：正文与思维链一个字都不删；**唯一**的删除是「本来就什么都 没有」的空路（`active` 而一个像素都不画的那种，留着就是一个占着列表末项的幽灵）。 - 工具卡收口 `List<ToolCallState>.settledAfterReconnect()`：`Running` ⇒ `phase = null` + `done = true` （渲染成「工具调用 · X」——不再有省略号、也不替服务端下「失败」的结论）；`Done`/`Error` 一个字节 都不动。`ToolPhase` 枚举**没加第四档**（它逐字对齐后端流程图 status，是契约面）。 - 接线在 `ChatViewModel.onConnectionChanged`：收口刻意放在既有的「同值早退」**之前**（见上）， 补拉那一半（`reconnectPull` / `gapCursor` / `firstPageState`）**一个字都没改**。 为什么 后端卡在回合里被重启 ⇒ 那一轮在服务端已经不存在 ⇒ 本端**永远**等不到 `turn.end` / `stream_end` / 助手落库行，而本端「进行中」的每一条闸都只认帧（灰态只认 `TurnBinding.closed`、卡片只认 `thinkingActive`/`active`/`phase`、底栏只认 `busyMode`）⇒ 三条路都不会自己收口，界面永久停在那一格。 今天唯一的兜底是 `session_idle` / `session_interrupted` 的那两个 `allClosed()` 调用点，而说那两句话的 进程已经没了。连带的第二半：那条永远 `active` 的流式路永远是 `items` 的最后一项（`reverseLayout` 下就是 屏幕最下方），而 `ChatScreen` 的「跟随底部」判的正是 `items.lastOrNull()?.key` ⇒ 末项不再变化 ⇒ 此后新消息不再把列表滚到底（用户视角就是「拉不到新消息」）。 触发时机只做「连接重建」(a)，**不做静默超时** (b)：理由、代价与风险（(b) 要多一个只能拍的时间常量、 会把正常但慢的回合误杀；但它能兜住「连接没断而编排器死锁」这一档，今天没有现场证据）都写在 `ReconnectReset.kt` 顶上，留主人拍板。 如实登记的两处代价 1. 收口**不删**正文/思维链/工具卡内容，所以那张卡本身要等一个后续事件才消失（它自己那一路的落库行， 或下一轮 `message.start` 的 `beginLeg` 把它当「已结束的路」丢掉）。若主人更想要「末项立刻让开」， 代价是那半段正文从屏幕上消失——我没选它（硬要求：不许动消息内容）。 2. `busyMode = null` 连带改变下一次发送的声明（`emitAndDeliver` 用 `busyMode != null` 算 `interruptScopeForSend` / `parallelForSend`）：重建后的**第一次**发送不再先发 `interrupt`。如实说： 重建之后本端确实不知道那一轮还在不在；服务端下一次报 `session_busy` 就恢复。 「闪动」口径：`closed` 单调（迟到的帧不会把它翻回灰态），流式卡片会被后续真帧重新拉回「在跑」 （那是如实的、且不丢字），`busyMode` 由服务端下一帧恢复。 验证 - `./gradlew.bat assembleDebug` BUILD SUCCESSFUL（app-debug.apk 已产出） - `./gradlew.bat testDebugUnitTest` 全量：**115 suites / 1405 tests / 0 failures / 0 errors** （基线 114 / 1388 ⇒ +1 suite、+17 tests，全部来自新增的 `ReconnectResetTest`） - 新增 `ReconnectResetTest`（17 例）：连接重建 ⇒ 未收口绑定全 closed（且 `phaseOf` 真的变成 Answered）； 第一次连接（`Idle`/`Connecting → Connected`）不收口；判据 5×5×2 = 50 格全表穷举；`settledAfterReconnect` 幂等且无变化时 `assertSame`；已 closed 的不会被 `withTurnBegin` 翻回；流式卡片全停且正文保留、空路被丢； 工具卡 `Running → null+done`、`Done`/`Error` 原样、后续真帧能放回 `Running`； 消息/待发/草稿/队列/error/绑定 messageIds **一律原样**。
This commit is contained in:
AskaEth committed
1 parent
53c3c5c8ee
commit
83a2a88b3c
4 files changed
+577
-2
No files matched your search
@@ -0,0 +1,329 @@
|
||||
package top.yeij.cyrene.ui.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
|
||||
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
|
||||
import top.yeij.cyrene.data.ws.WsState
|
||||
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.allClosed
|
||||
import top.yeij.cyrene.domain.chat.withTurnBegin
|
||||
import top.yeij.cyrene.domain.parse.ParsedBubble
|
||||
|
||||
/**
|
||||
* **连接重建时把「进行中」全部收口**（[isReconnect] + [settledAfterReconnect]）。
|
||||
*
|
||||
* 修的是主人 2026-10-06 报的那个现场：后端卡在回合里被重启 ⇒ 那一轮在服务端已经不存在 ⇒
|
||||
* 本端**永远**等不到 `turn.end` / `stream_end` / 助手落库行 ⇒ 气泡永久灰着、思维链永久「思考中…」、
|
||||
* 工具卡永久「执行 X…」、底栏永久「正在思考」。
|
||||
*
|
||||
* 四组测试各钉一件事：
|
||||
* 1. **收口真的发生**：连接重建 ⇒ 未收口的绑定全 `closed`（六态 ④ → ⑤）、流式卡片全部停止生长；
|
||||
* 2. **第一次连接不算重建**：`Idle/Connecting → Connected` 一个字节都不收（判据的穷举表在下面）；
|
||||
* 3. **幂等且单调**：已经收口的不会被翻回，收两次等于一次；
|
||||
* 4. **可降级**：消息 / 待发 / 草稿 / 队列 / 错误**一律原样**——收口只改「进行中的外观」。
|
||||
*/
|
||||
class ReconnectResetTest {
|
||||
|
||||
private fun bucket(turnId: String) = TurnBucket("web", turnId)
|
||||
|
||||
private fun bubble(text: String) =
|
||||
ParsedBubble(body = text, parts = emptyList(), head = emptyList(), tail = emptyList())
|
||||
|
||||
private fun message(id: String, role: String = "user", turnId: String? = null) =
|
||||
HistoryMessageDto(
|
||||
id = id,
|
||||
sessionId = "web",
|
||||
role = role,
|
||||
content = id,
|
||||
createdAt = 1L,
|
||||
turnId = turnId,
|
||||
)
|
||||
|
||||
/** 两路都还开着 + 一切「进行中」的字面承诺都挂着 —— 后端重启那一刻的本端样子。 */
|
||||
private fun stuckState() = ChatUiState(
|
||||
messages = listOf(message("m-a")),
|
||||
pending = listOf(PendingMessage("local-1", "还没发出去")),
|
||||
streams = listOf(StreamingLeg(bucket("t1"), streaming("半句回复", thinking = "她在想…"))),
|
||||
turns = emptyList<TurnBinding>()
|
||||
.withTurnBegin("t1", 1, listOf("m-a"), 2)
|
||||
.withTurnBegin("t2", 2, listOf("m-b"), 2, kind = "parallel"),
|
||||
queuedKeys = setOf("local-1"),
|
||||
queueDepth = 3,
|
||||
draft = "打到一半的字",
|
||||
busyMode = "thinking",
|
||||
activeTurns = 2,
|
||||
error = "上一次失败的原因",
|
||||
)
|
||||
|
||||
private fun streaming(text: String, thinking: String = "") = StreamingState(
|
||||
bubbles = text.takeIf { it.isNotBlank() }?.let { listOf(bubble(it)) }.orEmpty(),
|
||||
thinking = thinking,
|
||||
thinkingActive = thinking.isNotBlank(),
|
||||
active = true,
|
||||
)
|
||||
|
||||
// —— 1. 收口真的发生 ——
|
||||
|
||||
@Test
|
||||
fun `a rebuilt connection closes every binding that was still open`() {
|
||||
// 两路都还在飞（并行的两路也一样）⇒ 一次收口把**所有**未收口的绑定标 closed。
|
||||
// 六态的判据只有这一个闸（`turnPhaseOf` 第 2/3 条），所以这就是「气泡不再永远灰着」。
|
||||
val stuck = stuckState()
|
||||
assertEquals(TurnPhase.Answering, stuck.phaseOf(stuck.messages.single()))
|
||||
|
||||
val settled = stuck.settledAfterReconnect()
|
||||
|
||||
assertTrue(settled.turns.isNotEmpty())
|
||||
assertTrue("每一路都该收口", settled.turns.all { it.closed })
|
||||
assertEquals(TurnPhase.Answered, settled.phaseOf(settled.messages.single()))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a rebuilt connection stops every leg from growing`() {
|
||||
// 「思考中…」/「执行 X…」的唯一闸就是这几个位；没有卡住的那一路不在这里（它还在正常跑）。
|
||||
val stuck = stuckState().copy(
|
||||
streams = listOf(
|
||||
StreamingLeg(bucket("t1"), streaming("半句回复", thinking = "她在想…")),
|
||||
StreamingLeg(
|
||||
bucket("t2"),
|
||||
StreamingState(tools = listOf(ToolCallState("call-1", 0, tool = "search"))),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
val settled = stuck.settledAfterReconnect()
|
||||
|
||||
assertEquals(2, settled.streams.size)
|
||||
settled.streams.forEach { leg ->
|
||||
assertFalse("收口后不再生长", leg.state.active)
|
||||
assertFalse("标题不该再是「思考中…」", leg.state.thinkingActive)
|
||||
}
|
||||
assertEquals(listOf("半句回复"), settled.streams.first().state.bubbles.map { it.body })
|
||||
assertEquals("她在想…", settled.streams.first().state.thinking)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a rebuilt connection takes the busy banner down with it`() {
|
||||
// 底栏那句「正在思考…」是**服务端报的事实**（session_busy）——旧连接没了它就不可知了。
|
||||
// 与 `session_idle` / `session_interrupted` 那两处的处置逐字一致。
|
||||
val settled = stuckState().settledAfterReconnect()
|
||||
|
||||
assertNull(settled.busyMode)
|
||||
assertEquals(0, settled.activeTurns)
|
||||
assertEquals(0, settled.queueDepth)
|
||||
}
|
||||
|
||||
// —— 2. 「是不是重建」的判据（穷举） ——
|
||||
|
||||
@Test
|
||||
fun `the first connection is never a rebuild`() {
|
||||
// 冷启动：Idle → Connecting → Connected。把这一跳当重建会在用户刚打开会话的瞬间
|
||||
// 把正当在跑的路收口掉（首屏一到就可能有 turn_begin 把它绑成灰态）。
|
||||
assertFalse(isReconnect(WsState.Idle, WsState.Connected, everConnected = false))
|
||||
assertFalse(isReconnect(WsState.Connecting, WsState.Connected, everConnected = false))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `coming back to connected after a rebuild is a rebuild`() {
|
||||
// 后端重启：Connected → Reconnecting → Connected。这是本判据的主场。
|
||||
assertTrue(isReconnect(WsState.Reconnecting, WsState.Connected, everConnected = true))
|
||||
// 合流吞掉中间那一跳：收集器只看到 Connected → Connected，它**仍然是**一次真实的重建。
|
||||
assertTrue(isReconnect(WsState.Connected, WsState.Connected, everConnected = true))
|
||||
// 没记过账时的第二证据：上一跳是重连退避（只在本 VM 是离线期间建起来时才走到）。
|
||||
assertTrue(isReconnect(WsState.Reconnecting, WsState.Connected, everConnected = false))
|
||||
// 离线 → 重新 connect（`close()` 之后 attempt 归零走 Connecting）也是一次重建，
|
||||
// 只要本 VM 见过 Connected。
|
||||
assertTrue(isReconnect(WsState.Offline, WsState.Connected, everConnected = true))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `anything that is not connected is not a rebuild`() {
|
||||
// 掉下去的那一刻不收口（那一刻正是「进行中」最要紧的时候：还在等它自己跑完）。
|
||||
for (previous in WsState.entries) {
|
||||
for (ever in listOf(false, true)) {
|
||||
for (current in WsState.entries.filter { it != WsState.Connected }) {
|
||||
assertFalse(
|
||||
"previous=$previous current=$current ever=$ever",
|
||||
isReconnect(previous, current, ever),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the judgement is exhaustive over every state triple`() {
|
||||
// 5 × 5 × 2 = 50 格全表。期望值**独立**写一遍（当前不是 Connected ⇒ 假；见过 Connected ⇒
|
||||
// 真；没记过账但上一跳是重连退避 ⇒ 真；其余（第一次接上）⇒ 假）。
|
||||
for (previous in WsState.entries) {
|
||||
for (current in WsState.entries) {
|
||||
for (ever in listOf(false, true)) {
|
||||
val expected = when {
|
||||
current != WsState.Connected -> false
|
||||
ever -> true
|
||||
previous == WsState.Reconnecting -> true
|
||||
else -> false
|
||||
}
|
||||
assertEquals(
|
||||
"previous=$previous current=$current everConnected=$ever",
|
||||
expected,
|
||||
isReconnect(previous, current, ever),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// —— 3. 幂等 / 单调 ——
|
||||
|
||||
@Test
|
||||
fun `settling twice is the same as settling once`() {
|
||||
val once = stuckState().settledAfterReconnect()
|
||||
|
||||
assertSame(once, once.settledAfterReconnect())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `nothing in flight returns the very same instance`() {
|
||||
// 每次 Connected 都会调它；没有「进行中」可收时必须返回同一个实例（否则 StateFlow 白发一轮）。
|
||||
val idle = ChatUiState(messages = listOf(message("m-a")))
|
||||
|
||||
assertSame(idle, idle.settledAfterReconnect())
|
||||
assertSame(emptyList<StreamingLeg>(), emptyList<StreamingLeg>().allFinished())
|
||||
assertSame(emptyList<ToolCallState>(), emptyList<ToolCallState>().settledAfterReconnect())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a binding that was already closed is never flipped back`() {
|
||||
// `closed` 是**单调**的既成事实（`withTurnBegin` 合并时也保持原值）——收口之后
|
||||
// 一句迟到的「这一路还在跑」不许把它翻回灰态。
|
||||
val closed = emptyList<TurnBinding>()
|
||||
.withTurnBegin("t1", 1, listOf("m-a"), 1)
|
||||
.allClosed()
|
||||
val state = ChatUiState(messages = listOf(message("m-a")), turns = closed)
|
||||
|
||||
val settled = state.settledAfterReconnect()
|
||||
|
||||
assertTrue(settled.turns.all { it.closed })
|
||||
assertEquals(TurnPhase.Answered, settled.phaseOf(settled.messages.single()))
|
||||
// 一句迟到的 turn_begin 也不会把它翻回去（既有的 withTurnBegin 纪律，这里一起钉住）。
|
||||
val reopened = settled.turns.withTurnBegin("t1", 1, listOf("m-a"), 1)
|
||||
assertEquals(TurnPhase.Answered, settled.copy(turns = reopened).phaseOf(settled.messages.single()))
|
||||
}
|
||||
|
||||
// —— 4. 可降级：收口只改「进行中的外观」 ——
|
||||
|
||||
@Test
|
||||
fun `settling touches nothing the user typed or sent`() {
|
||||
val stuck = stuckState()
|
||||
|
||||
val settled = stuck.settledAfterReconnect()
|
||||
|
||||
assertSame(stuck.messages, settled.messages)
|
||||
assertSame(stuck.pending, settled.pending)
|
||||
assertSame(stuck.queuedKeys, settled.queuedKeys)
|
||||
assertEquals(stuck.draft, settled.draft)
|
||||
assertEquals(stuck.error, settled.error)
|
||||
assertFalse(settled.sessionGone)
|
||||
// 绑定表**只标 closed，不删**：并行路角标靠它，删了等于事后篡改历史。
|
||||
assertEquals(stuck.turns.map { it.turnId }, settled.turns.map { it.turnId })
|
||||
assertEquals(stuck.turns.map { it.messageIds }, settled.turns.map { it.messageIds })
|
||||
}
|
||||
|
||||
// —— 工具卡的收口 ——
|
||||
|
||||
@Test
|
||||
fun `a running tool card stops claiming to be running`() {
|
||||
val tools = listOf(
|
||||
ToolCallState(
|
||||
id = "call-1",
|
||||
index = 0,
|
||||
tool = "search",
|
||||
arguments = "{\"q\":1}",
|
||||
output = "前 3 条",
|
||||
done = true,
|
||||
phase = ToolPhase.Running,
|
||||
),
|
||||
)
|
||||
|
||||
val settled = tools.settledAfterReconnect()
|
||||
|
||||
val card = settled.single()
|
||||
assertNull("不再声称在跑", card.phase)
|
||||
assertTrue("省略号必须去掉", card.done)
|
||||
// 参数与输出一个都没动：这是「不再是进行中」，不是「清空这张卡」。
|
||||
assertEquals("{\"q\":1}", card.arguments)
|
||||
assertEquals("前 3 条", card.output)
|
||||
assertEquals("search", card.tool)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a tool call without an end frame loses its ellipsis too`() {
|
||||
// 只到了 `tool_call.*`、没到 `tool_execution.*`：标题是「调用工具…」——那个省略号
|
||||
// 同样是「还在跑」，收口之后它不再在跑。
|
||||
val settled = listOf(ToolCallState("call-1", 0, tool = "search")).settledAfterReconnect()
|
||||
|
||||
assertTrue(settled.single().done)
|
||||
assertNull(settled.single().phase)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a tool card with a verdict from the server is left alone`() {
|
||||
// Done / Error 是**服务端给的结论**，本端不许改写它，更不许把它降级成「不知道」。
|
||||
val done = ToolCallState("c1", 0, phase = ToolPhase.Done, done = true, resultSummary = "2 条")
|
||||
val failed = ToolCallState("c2", 1, phase = ToolPhase.Error, done = true)
|
||||
|
||||
val tools = listOf(done, failed).also { assertEquals(2, it.size) }
|
||||
val settled = tools.settledAfterReconnect()
|
||||
|
||||
assertSame(ToolPhase.Done, settled[0].phase)
|
||||
assertSame(ToolPhase.Error, settled[1].phase)
|
||||
assertEquals("2 条", settled[0].resultSummary)
|
||||
assertSame(tools, tools.settledAfterReconnect())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a real frame after settling can put the card back to running`() {
|
||||
// 如实登记「闪动」的方向：收口之后**真的又来了一帧**时，卡片会重新变成「执行 X…」——
|
||||
// 那是如实的（它确实又在跑了），而且是自愈的：这条帧不来的话卡片就停在「工具调用 · X」。
|
||||
val settled = listOf(ToolCallState("c1", 0, tool = "search", phase = ToolPhase.Running))
|
||||
.settledAfterReconnect()
|
||||
|
||||
val again = applyToolExecution(settled, "c1", "search", "running")
|
||||
|
||||
assertSame(ToolPhase.Running, again.single().phase)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an empty leg is dropped instead of lingering as a ghost`() {
|
||||
// `active = true` 而什么都不画时，气泡里只有一个光标；收口它等于留下一个**一个像素都不画、
|
||||
// 却仍占着列表末项**的幽灵（末项正是「跟随底部」的判据）。这一路本来就没内容，丢掉它
|
||||
// 是本函数唯一的删除动作。
|
||||
val empty = listOf(StreamingLeg(bucket("t1"), StreamingState(active = true)))
|
||||
|
||||
assertEquals(emptyList<StreamingLeg>(), empty.allFinished())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a leg with only a running tool card survives as an inert card`() {
|
||||
val leg = StreamingLeg(
|
||||
bucket("t1"),
|
||||
StreamingState(
|
||||
active = true,
|
||||
tools = listOf(ToolCallState("c1", 0, tool = "search", phase = ToolPhase.Running)),
|
||||
),
|
||||
)
|
||||
|
||||
val settled = listOf(leg).allFinished()
|
||||
|
||||
assertEquals(1, settled.size)
|
||||
assertFalse(settled.single().state.active)
|
||||
assertNull(settled.single().state.tools.single().phase)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user