Skip to content

Commit 2439ad9

Browse files
steipetesteipete
committed
feat(android): restore in-flight runs after reconnect
Consume the gateway's canonical reconnect snapshot (chat.history inFlightRun) to re-adopt still-streaming runs after reconnect, cold start, and seq-gap recovery in the Android ChatController. Mirrors the iOS contract from #100277. Part of #100197.
1 parent b4e0c53 commit 2439ad9

4 files changed

Lines changed: 253 additions & 1 deletion

File tree

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

Lines changed: 60 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,8 +89,13 @@ class ChatController internal constructor(
8989
private var lastHealthPollAtMs: Long? = null
9090
private var commandsAgentId: String? = null
9191

92+
// Armed on disconnect so the next health event refetches history and re-adopts
93+
// any run the gateway still reports in flight (chat.history `inFlightRun`).
94+
private var restoreRunStateOnReconnect = false
95+
9296
/** Clears transient chat state when the operator gateway session disconnects. */
9397
fun onDisconnected(message: String) {
98+
restoreRunStateOnReconnect = true
9499
_healthOk.value = false
95100
_errorText.value = null
96101
_commands.value = emptyList()
@@ -421,10 +426,19 @@ class ChatController internal constructor(
421426
"health" -> {
422427
_healthOk.value = true
423428
refreshCommandsAfterReconnect()
429+
if (restoreRunStateOnReconnect) {
430+
restoreRunStateOnReconnect = false
431+
refreshHistoryForRecovery()
432+
}
424433
}
425434
"seqGap" -> {
426-
_errorText.value = "Event stream interrupted; try refreshing."
435+
// Missed events may include deltas or the terminal state of a pending run;
436+
// refetch history and rebuild run state from the gateway snapshot.
427437
clearPendingRuns()
438+
pendingToolCallsById.clear()
439+
publishPendingToolCalls()
440+
_streamingAssistantText.value = null
441+
refreshHistoryForRecovery()
428442
}
429443
"chat" -> {
430444
if (payloadJson.isNullOrBlank()) return
@@ -448,6 +462,21 @@ class ChatController internal constructor(
448462
}
449463
}
450464

465+
/**
466+
* Reconnect/seq-gap recovery: refetch history for the current session without the
467+
* beginHistoryLoad transient-state reset. Disconnect (or the seqGap handler) already
468+
* cleared run state, and resetting healthOk here would block sends after reconnect.
469+
*/
470+
private fun refreshHistoryForRecovery() {
471+
val key = normalizeRequestedSessionKey(_sessionKey.value)
472+
val generation = historyLoadGeneration.incrementAndGet()
473+
_sessionKey.value = key
474+
_historyLoading.value = true
475+
scope.launch {
476+
bootstrap(sessionKey = key, generation = generation, forceHealth = false, refreshSessions = true)
477+
}
478+
}
479+
451480
private suspend fun bootstrap(
452481
sessionKey: String,
453482
generation: Long,
@@ -466,6 +495,7 @@ class ChatController internal constructor(
466495
prunePersistedOptimisticMessages(history.messages)
467496
_messages.value = mergeOptimisticMessages(incoming = history.messages, optimistic = optimisticMessagesByRunId.values)
468497
_sessionId.value = history.sessionId
498+
adoptInFlightRun(history.inFlightRun)
469499
_historyLoading.value = false
470500
history.thinkingLevel
471501
?.trim()
@@ -709,6 +739,28 @@ class ChatController internal constructor(
709739
pendingToolCallsById.values.sortedBy { it.startedAtMs }
710740
}
711741

742+
/**
743+
* Adopts the run the gateway reports still streaming for this session so reconnect,
744+
* cold start, and seq-gap recovery restore pending/streaming UI state. Snapshot absence
745+
* never clears local state: live terminal events and the pending-run timeout own
746+
* completion, and a snapshot fetched before our own send must not cancel that run.
747+
*/
748+
private fun adoptInFlightRun(run: ChatInFlightRun?) {
749+
if (run == null) return
750+
val runId = run.runId.trim()
751+
if (runId.isEmpty()) return
752+
synchronized(pendingRuns) {
753+
// A different locally-owned run means this snapshot predates it; ignore.
754+
if (pendingRuns.isNotEmpty() && runId !in pendingRuns) return
755+
pendingRuns.add(runId)
756+
_pendingRunCount.value = pendingRuns.size
757+
}
758+
armPendingRunTimeout(runId)
759+
if (run.text.isNotEmpty()) {
760+
_streamingAssistantText.value = run.text
761+
}
762+
}
763+
712764
private fun armPendingRunTimeout(runId: String) {
713765
pendingRunTimeoutJobs[runId]?.cancel()
714766
pendingRunTimeoutJobs[runId] =
@@ -832,9 +884,16 @@ class ChatController internal constructor(
832884
thinkingLevel = thinkingLevel,
833885
messages = reconcileMessageIds(previous = previousMessages, incoming = messages),
834886
sessionInfo = sessionInfo,
887+
inFlightRun = parseInFlightRun(root),
835888
)
836889
}
837890

891+
private fun parseInFlightRun(root: JsonObject): ChatInFlightRun? {
892+
val obj = root["inFlightRun"].asObjectOrNull() ?: return null
893+
val runId = obj["runId"].asStringOrNull()?.trim()?.takeIf { it.isNotEmpty() } ?: return null
894+
return ChatInFlightRun(runId = runId, text = obj["text"].asStringOrNull().orEmpty())
895+
}
896+
838897
private fun parseSessions(jsonString: String): List<ChatSessionEntry> {
839898
val root = json.parseToJsonElement(jsonString).asObjectOrNull() ?: return emptyList()
840899
val sessions = root["sessions"].asArrayOrNull() ?: return emptyList()

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,15 @@ data class ChatCommandEntry(
5858
)
5959

6060

61+
/**
62+
* Run still streaming on the gateway when a chat.history snapshot was captured;
63+
* [text] is the assistant text buffered so far (may be empty for runs without deltas).
64+
*/
65+
data class ChatInFlightRun(
66+
val runId: String,
67+
val text: String,
68+
)
69+
6170
/**
6271
* Snapshot of one chat session, including optional thinking level selected on the gateway.
6372
*/
@@ -67,6 +76,7 @@ data class ChatHistory(
6776
val thinkingLevel: String?,
6877
val messages: List<ChatMessage>,
6978
val sessionInfo: ChatSessionEntry? = null,
79+
val inFlightRun: ChatInFlightRun? = null,
7080
)
7181

7282
/**

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

Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,172 @@
1+
package ai.openclaw.app.chat
2+
3+
import kotlinx.coroutines.ExperimentalCoroutinesApi
4+
import kotlinx.coroutines.test.TestScope
5+
import kotlinx.coroutines.test.runCurrent
6+
import kotlinx.coroutines.test.runTest
7+
import kotlinx.serialization.json.Json
8+
import org.junit.Assert.assertEquals
9+
import org.junit.Assert.assertNull
10+
import org.junit.Assert.assertTrue
11+
import org.junit.Test
12+
13+
/**
14+
* Reconnect recovery scenarios: after a gateway disconnect, the next health event
15+
* refetches chat.history and re-adopts the run the gateway still reports in flight
16+
* (`inFlightRun`), matching the reconnect snapshot contract the TUI consumes.
17+
*/
18+
class ChatControllerReconnectRestoreTest {
19+
private val json = Json { ignoreUnknownKeys = true }
20+
21+
private fun TestScope.newController(gateway: ScriptedGateway): ChatController =
22+
ChatController(scope = this, json = json, requestGateway = gateway::request)
23+
24+
private val userTurn = ReplayHistoryMessage("user", "keep working", 1_000)
25+
26+
@Test
27+
@OptIn(ExperimentalCoroutinesApi::class)
28+
fun reconnectAdoptsInFlightRunAndConsumesLiveEvents() =
29+
runTest {
30+
val gateway = ScriptedGateway(json)
31+
gateway.respondWith("chat.history", historyResponse("session-1", listOf(userTurn)))
32+
val controller = newController(gateway)
33+
controller.load("main")
34+
runCurrent()
35+
assertEquals(0, controller.pendingRunCount.value)
36+
37+
controller.onDisconnected("Reconnecting…")
38+
gateway.respondWith(
39+
"chat.history",
40+
historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "partial reply"),
41+
)
42+
controller.handleGatewayEvent("health", null)
43+
runCurrent()
44+
45+
assertEquals(1, controller.pendingRunCount.value)
46+
assertEquals("partial reply", controller.streamingAssistantText.value)
47+
assertEquals(1, controller.messages.value.size)
48+
49+
// The adopted run keeps consuming live deltas and its terminal event.
50+
controller.handleGatewayEvent(
51+
"chat",
52+
chatDeltaPayload("main", "run-active", 5, " more", "partial reply more"),
53+
)
54+
assertEquals("partial reply more", controller.streamingAssistantText.value)
55+
gateway.respondWith(
56+
"chat.history",
57+
historyResponse(
58+
"session-1",
59+
listOf(userTurn, ReplayHistoryMessage("assistant", "partial reply more", 2_000)),
60+
),
61+
)
62+
controller.handleGatewayEvent("chat", chatTerminalPayload("main", "run-active", seq = 6))
63+
runCurrent()
64+
65+
assertEquals(0, controller.pendingRunCount.value)
66+
assertNull(controller.streamingAssistantText.value)
67+
assertEquals(2, controller.messages.value.size)
68+
}
69+
70+
@Test
71+
@OptIn(ExperimentalCoroutinesApi::class)
72+
fun reconnectWithoutInFlightRunStaysClean() =
73+
runTest {
74+
val gateway = ScriptedGateway(json)
75+
gateway.respondWith("chat.history", historyResponse("session-1", listOf(userTurn)))
76+
val controller = newController(gateway)
77+
controller.load("main")
78+
runCurrent()
79+
val historyCallsAfterLoad = gateway.callCount("chat.history")
80+
81+
controller.onDisconnected("Offline")
82+
controller.handleGatewayEvent("health", null)
83+
runCurrent()
84+
85+
// Reconnect refetched history once and restored nothing.
86+
assertEquals(historyCallsAfterLoad + 1, gateway.callCount("chat.history"))
87+
assertEquals(0, controller.pendingRunCount.value)
88+
assertNull(controller.streamingAssistantText.value)
89+
assertNull(controller.errorText.value)
90+
assertTrue(controller.healthOk.value)
91+
assertEquals(1, controller.messages.value.size)
92+
}
93+
94+
@Test
95+
@OptIn(ExperimentalCoroutinesApi::class)
96+
fun repeatedReconnectsDoNotDuplicateRunOrRows() =
97+
runTest {
98+
val gateway = ScriptedGateway(json)
99+
gateway.respondWith(
100+
"chat.history",
101+
historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "partial"),
102+
)
103+
val controller = newController(gateway)
104+
controller.load("main")
105+
runCurrent()
106+
assertEquals(1, controller.pendingRunCount.value)
107+
108+
repeat(2) {
109+
controller.onDisconnected("Reconnecting…")
110+
assertEquals(0, controller.pendingRunCount.value)
111+
controller.handleGatewayEvent("health", null)
112+
runCurrent()
113+
}
114+
115+
assertEquals(1, controller.pendingRunCount.value)
116+
assertEquals("partial", controller.streamingAssistantText.value)
117+
assertEquals(1, controller.messages.value.size)
118+
}
119+
120+
@Test
121+
@OptIn(ExperimentalCoroutinesApi::class)
122+
fun staleSnapshotRunDoesNotReplaceLocallyOwnedSend() =
123+
runTest {
124+
val gateway = ScriptedGateway(json)
125+
gateway.respondWith("chat.history", historyResponse("session-1", emptyList()))
126+
gateway.respondChatSend(status = "started")
127+
val controller = newController(gateway)
128+
controller.load("main")
129+
runCurrent()
130+
131+
// Reconnect arms a recovery refresh, then a send lands before it executes.
132+
controller.onDisconnected("Reconnecting…")
133+
gateway.respondWith(
134+
"chat.history",
135+
historyResponse("session-1", emptyList(), inFlightRun = "run-stale" to "old text"),
136+
)
137+
controller.handleGatewayEvent("health", null)
138+
assertTrue(controller.sendMessageAwaitAcceptance("new work", "off", emptyList()))
139+
val localRunId = requireNotNull(gateway.lastRunId)
140+
runCurrent()
141+
142+
assertEquals(1, controller.pendingRunCount.value)
143+
controller.handleGatewayEvent(
144+
"chat",
145+
chatDeltaPayload("main", localRunId, 1, "ours", "ours"),
146+
)
147+
assertEquals("ours", controller.streamingAssistantText.value)
148+
}
149+
150+
@Test
151+
@OptIn(ExperimentalCoroutinesApi::class)
152+
fun seqGapRefetchesHistoryAndRestoresInFlightRun() =
153+
runTest {
154+
val gateway = ScriptedGateway(json)
155+
gateway.respondWith(
156+
"chat.history",
157+
historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "still going"),
158+
)
159+
val controller = newController(gateway)
160+
controller.load("main")
161+
runCurrent()
162+
assertEquals(1, controller.pendingRunCount.value)
163+
164+
controller.handleGatewayEvent("seqGap", null)
165+
runCurrent()
166+
167+
assertEquals(1, controller.pendingRunCount.value)
168+
assertEquals("still going", controller.streamingAssistantText.value)
169+
assertNull(controller.errorText.value)
170+
assertEquals(1, controller.messages.value.size)
171+
}
172+
}

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,9 +89,20 @@ internal data class ReplayHistoryMessage(
8989
internal fun historyResponse(
9090
sessionId: String,
9191
messages: List<ReplayHistoryMessage>,
92+
inFlightRun: Pair<String, String>? = null,
9293
): String =
9394
buildJsonObject {
9495
put("sessionId", JsonPrimitive(sessionId))
96+
if (inFlightRun != null) {
97+
put(
98+
"inFlightRun",
99+
buildJsonObject {
100+
put("runId", JsonPrimitive(inFlightRun.first))
101+
put("text", JsonPrimitive(inFlightRun.second))
102+
},
103+
)
104+
put("sessionInfo", buildJsonObject { put("hasActiveRun", JsonPrimitive(true)) })
105+
}
95106
put(
96107
"messages",
97108
JsonArray(

0 commit comments

Comments
 (0)