Skip to content

Commit 5921f31

Browse files
steipetesteipete
committed
fix(android): harden reconnect history cursor
1 parent 33524a6 commit 5921f31

3 files changed

Lines changed: 144 additions & 22 deletions

File tree

‎apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt‎

Lines changed: 54 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
package ai.openclaw.app.chat
22

3-
import ai.openclaw.app.resolveAgentIdFromMainSessionKey
43
import ai.openclaw.app.gateway.GatewaySession
54
import ai.openclaw.app.gateway.parseChatSendAck
5+
import ai.openclaw.app.resolveAgentIdFromMainSessionKey
66
import kotlinx.coroutines.CoroutineScope
77
import kotlinx.coroutines.Job
88
import kotlinx.coroutines.delay
@@ -93,6 +93,10 @@ class ChatController internal constructor(
9393
// reconnect catch-up can fetch only missed rows; resets wherever messages reset.
9494
private var lastAppliedTranscriptSeq: Long? = null
9595

96+
// Visible session state clears offline; this identity stays paired with the cursor so
97+
// a replacement transcript cannot append onto the prior session after reconnect.
98+
private var lastAppliedTranscriptSessionId: String? = null
99+
96100
// Armed on disconnect, consumed on the next healthy transition; keeps routine health
97101
// polls from re-running catch-up while connected.
98102
private var reconnectCatchUpPending = false
@@ -250,6 +254,7 @@ class ChatController internal constructor(
250254
if (clearMessages) {
251255
_messages.value = emptyList()
252256
lastAppliedTranscriptSeq = null
257+
lastAppliedTranscriptSessionId = null
253258
}
254259
return generation
255260
}
@@ -261,8 +266,7 @@ class ChatController internal constructor(
261266
return key
262267
}
263268

264-
private fun resolveAgentIdForSessionKey(parentKey: String): String =
265-
resolveAgentIdFromMainSessionKey(parentKey) ?: "main"
269+
private fun resolveAgentIdForSessionKey(parentKey: String): String = resolveAgentIdFromMainSessionKey(parentKey) ?: "main"
266270

267271
/** Queues a chat send without waiting for gateway acceptance. */
268272
fun sendMessage(
@@ -491,7 +495,10 @@ class ChatController internal constructor(
491495
val historyJson =
492496
requestGateway(
493497
"chat.history",
494-
buildJsonObject { put("sessionKey", JsonPrimitive(sessionKey)) }.toString(),
498+
buildJsonObject {
499+
put("sessionKey", JsonPrimitive(sessionKey))
500+
put("offset", JsonPrimitive(0))
501+
}.toString(),
495502
)
496503
if (!isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get())) return null
497504
val history = parseHistory(historyJson, sessionKey = sessionKey, previousMessages = _messages.value)
@@ -517,6 +524,7 @@ class ChatController internal constructor(
517524
// A full page is the authoritative baseline; a pending reconnect catch-up would only
518525
// re-fetch rows this page already delivered. Null seq (legacy gateway) disables catch-up.
519526
lastAppliedTranscriptSeq = history.maxTranscriptSeq
527+
lastAppliedTranscriptSessionId = history.sessionId
520528
reconnectCatchUpPending = false
521529
}
522530

@@ -588,11 +596,15 @@ class ChatController internal constructor(
588596
// Fresh sessions and legacy gateways have no baseline; the existing full-fetch
589597
// paths (load/refresh/terminal events) stay the only history source for them.
590598
val afterSeq = lastAppliedTranscriptSeq ?: return
591-
scope.launch { catchUpHistoryAfterReconnect(afterSeq) }
599+
val sessionId = lastAppliedTranscriptSessionId ?: return
600+
scope.launch { catchUpHistoryAfterReconnect(afterSeq, sessionId) }
592601
}
593602

594603
/** Fetches only transcript rows missed while disconnected by paging the afterSeq cursor. */
595-
private suspend fun catchUpHistoryAfterReconnect(startAfterSeq: Long) {
604+
private suspend fun catchUpHistoryAfterReconnect(
605+
startAfterSeq: Long,
606+
expectedSessionId: String,
607+
) {
596608
val sessionKey = _sessionKey.value
597609
val generation = historyLoadGeneration.get()
598610
var afterSeq = startAfterSeq
@@ -609,11 +621,7 @@ class ChatController internal constructor(
609621
} catch (_: Throwable) {
610622
// Version-skew fallback: older gateways reject the unknown afterSeq param,
611623
// so fall back to the plain full-history refetch.
612-
try {
613-
fetchAndApplyCurrentHistory(sessionKey, generation, updateSessionInfo = false)
614-
} catch (_: Throwable) {
615-
// best-effort; a re-disconnect re-arms the catch-up
616-
}
624+
refetchCurrentHistoryBestEffort(sessionKey, generation)
617625
return
618626
}
619627
if (!isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get())) return
@@ -625,14 +633,34 @@ class ChatController internal constructor(
625633
applyWholesaleHistory(history, updateSessionInfo = false)
626634
return
627635
}
628-
applyHistoryDeltaPage(history, cursor)
629636
val nextAfterSeq = cursor.nextAfterSeq
630-
// Advance-only guard: a non-advancing cursor echo must not loop forever.
631-
if (!cursor.hasMore || nextAfterSeq == null || nextAfterSeq <= afterSeq) return
637+
if (
638+
cursor.afterSeq != afterSeq ||
639+
history.sessionId != expectedSessionId ||
640+
nextAfterSeq == null ||
641+
nextAfterSeq < afterSeq ||
642+
(cursor.hasMore && nextAfterSeq == afterSeq)
643+
) {
644+
refetchCurrentHistoryBestEffort(sessionKey, generation)
645+
return
646+
}
647+
applyHistoryDeltaPage(history, cursor)
648+
if (!cursor.hasMore) return
632649
afterSeq = nextAfterSeq
633650
}
634651
}
635652

653+
private suspend fun refetchCurrentHistoryBestEffort(
654+
sessionKey: String,
655+
generation: Long,
656+
) {
657+
try {
658+
fetchAndApplyCurrentHistory(sessionKey, generation, updateSessionInfo = false)
659+
} catch (_: Throwable) {
660+
// Best-effort; a re-disconnect re-arms catch-up.
661+
}
662+
}
663+
636664
/** Appends one catch-up page, deduping rows the client already shows by message identity. */
637665
private fun applyHistoryDeltaPage(
638666
history: ChatHistory,
@@ -891,11 +919,13 @@ class ChatController internal constructor(
891919

892920
// Max over ALL rows (including role-less rows dropped above): the cursor consumes raw
893921
// transcript rows, so the baseline must cover rows this page delivered but did not render.
894-
val maxTranscriptSeq =
922+
val transcriptSeqs =
895923
array
896-
.mapNotNull { item -> item.asObjectOrNull()?.get("__openclaw").asObjectOrNull()?.get("seq").asLongOrNull() }
897-
.filter { it > 0L }
898-
.maxOrNull()
924+
.mapNotNull { item ->
925+
val metadata = item.asObjectOrNull()?.get("__openclaw").asObjectOrNull()
926+
metadata?.get("seq").asLongOrNull()
927+
}
928+
val maxTranscriptSeq = transcriptSeqs.filter { it > 0L }.maxOrNull()
899929
val cursor =
900930
root["afterSeq"].asLongOrNull()?.let { echoedAfterSeq ->
901931
ChatHistoryCursor(
@@ -1056,9 +1086,14 @@ internal fun parseChatMessageContents(obj: JsonObject): List<ChatMessageContent>
10561086
return emptyList()
10571087
}
10581088

1059-
private fun parseCreatedSessionKey(json: Json, sessionJson: String): String? {
1089+
private fun parseCreatedSessionKey(
1090+
json: Json,
1091+
sessionJson: String,
1092+
): String? {
10601093
val root = runCatching { json.parseToJsonElement(sessionJson).asObjectOrNull() }.getOrNull() ?: return null
1094+
10611095
fun clean(value: String?): String? = value?.trim()?.takeIf { it.isNotEmpty() }
1096+
10621097
return clean(root["key"].asStringOrNull())
10631098
?: clean(root["sessionKey"].asStringOrNull())
10641099
?: root["session"].asObjectOrNull()?.let { session ->

‎apps/android/app/src/main/java/ai/openclaw/app/chat/ChatModels.kt‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,6 @@ data class ChatCommandEntry(
5757
val acceptsArgs: Boolean = false,
5858
)
5959

60-
6160
/**
6261
* Snapshot of one chat session, including optional thinking level selected on the gateway.
6362
*

‎apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectCatchUpTest.kt‎

Lines changed: 90 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,16 +27,20 @@ class ChatControllerReconnectCatchUpTest {
2727
return """{"role":"$role","content":[{"type":"text","text":"$text"}],"timestamp":$timestamp$meta}"""
2828
}
2929

30-
private fun fullHistory(vararg rows: String): String = """{"sessionId":"sess-1","messages":[${rows.joinToString(",")}]}"""
30+
private fun fullHistory(
31+
vararg rows: String,
32+
sessionId: String = "sess-1",
33+
): String = """{"sessionId":"$sessionId","messages":[${rows.joinToString(",")}]}"""
3134

3235
private fun cursorPage(
3336
afterSeq: Long,
3437
nextAfterSeq: Long,
3538
hasMore: Boolean,
3639
vararg rows: String,
40+
sessionId: String = "sess-1",
3741
): String =
3842
"""
39-
{"sessionId":"sess-1","afterSeq":$afterSeq,"nextAfterSeq":$nextAfterSeq,"hasMore":$hasMore,
43+
{"sessionId":"$sessionId","afterSeq":$afterSeq,"nextAfterSeq":$nextAfterSeq,"hasMore":$hasMore,
4044
"totalMessages":$nextAfterSeq,"messages":[${rows.joinToString(",")}]}
4145
""".trimIndent()
4246

@@ -92,6 +96,7 @@ class ChatControllerReconnectCatchUpTest {
9296

9397
val deltaParams = calls.historyParams().last()
9498
assertTrue(deltaParams, deltaParams.contains(""""afterSeq":2"""))
99+
assertTrue(calls.historyParams().first().contains(""""offset":0"""))
95100
assertEquals(listOf("hello", "world", "next"), controller.visibleTexts())
96101
}
97102

@@ -212,9 +217,92 @@ class ChatControllerReconnectCatchUpTest {
212217
assertEquals(3, historyParams.size)
213218
assertTrue(historyParams[1].contains("afterSeq"))
214219
assertTrue(!historyParams[2].contains("afterSeq"))
220+
assertTrue(historyParams[2].contains(""""offset":0"""))
215221
assertEquals(listOf("hello", "world"), controller.visibleTexts())
216222
}
217223

224+
@Test
225+
fun changedSessionIdRebasesFromFullHistory() =
226+
runTest {
227+
val calls = mutableListOf<RecordedCall>()
228+
var fullFetches = 0
229+
val controller =
230+
ChatController(
231+
scope = this,
232+
json = json,
233+
requestGateway = { method, params ->
234+
calls += RecordedCall(method, params)
235+
when {
236+
method != "chat.history" -> "{}"
237+
params.orEmpty().contains("afterSeq") ->
238+
cursorPage(
239+
afterSeq = 2,
240+
nextAfterSeq = 3,
241+
hasMore = false,
242+
historyRow("assistant", "stale", 3000, seq = 3),
243+
sessionId = "sess-replaced",
244+
)
245+
fullFetches++ == 0 ->
246+
fullHistory(
247+
historyRow("user", "old question", 1000, seq = 1),
248+
historyRow("assistant", "old answer", 2000, seq = 2),
249+
)
250+
else ->
251+
fullHistory(
252+
historyRow("assistant", "replacement", 4000, seq = 1),
253+
sessionId = "sess-replaced",
254+
)
255+
}
256+
},
257+
)
258+
259+
controller.load("main")
260+
advanceUntilIdle()
261+
controller.onDisconnected("gateway closed")
262+
controller.handleGatewayEvent("health", null)
263+
advanceUntilIdle()
264+
265+
assertEquals(listOf("replacement"), controller.visibleTexts())
266+
assertEquals(3, calls.historyParams().size)
267+
}
268+
269+
@Test
270+
fun backwardCursorRebasesFromFullHistory() =
271+
runTest {
272+
var fullFetches = 0
273+
val controller =
274+
ChatController(
275+
scope = this,
276+
json = json,
277+
requestGateway = { method, params ->
278+
when {
279+
method != "chat.history" -> "{}"
280+
params.orEmpty().contains("afterSeq") ->
281+
cursorPage(
282+
afterSeq = 2,
283+
nextAfterSeq = 1,
284+
hasMore = false,
285+
historyRow("assistant", "stale", 3000, seq = 1),
286+
)
287+
fullFetches++ == 0 ->
288+
fullHistory(
289+
historyRow("user", "old question", 1000, seq = 1),
290+
historyRow("assistant", "old answer", 2000, seq = 2),
291+
)
292+
else -> fullHistory(historyRow("assistant", "replacement", 4000, seq = 3))
293+
}
294+
},
295+
)
296+
297+
controller.load("main")
298+
advanceUntilIdle()
299+
controller.onDisconnected("gateway closed")
300+
controller.handleGatewayEvent("health", null)
301+
advanceUntilIdle()
302+
303+
assertEquals(listOf("replacement"), controller.visibleTexts())
304+
}
305+
218306
@Test
219307
fun sessionSwitchResetsSeqBaselineToNewSession() =
220308
runTest {

0 commit comments

Comments
 (0)