Skip to content

Commit c5e2cf6

Browse files
nxmxbbdnxmxbbd
committed
fix(native): clear terminal chat send acks
1 parent cd4e345 commit c5e2cf6

13 files changed

Lines changed: 455 additions & 64 deletions

File tree

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

Lines changed: 3 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import ai.openclaw.app.gateway.GatewayTlsProbeFailure
1414
import ai.openclaw.app.gateway.GatewayTlsProbeResult
1515
import ai.openclaw.app.gateway.GatewayUpdateAvailableSummary
1616
import ai.openclaw.app.gateway.normalizeGatewayTlsFingerprint
17+
import ai.openclaw.app.gateway.parseChatSendAck
1718
import ai.openclaw.app.gateway.probeGatewayTlsFingerprint
1819
import ai.openclaw.app.node.A2UIHandler
1920
import ai.openclaw.app.node.CalendarHandler
@@ -590,7 +591,8 @@ class NodeRuntime(
590591
put("idempotencyKey", JsonPrimitive(idempotencyKey))
591592
}
592593
val response = operatorSession.request("chat.send", params.toString())
593-
parseChatSendRunId(response) ?: idempotencyKey
594+
val ack = parseChatSendAck(json, response)
595+
if (ack.isTerminal) null else (ack.runId ?: idempotencyKey)
594596
},
595597
speakAssistantReply = { text ->
596598
// Voice-tab replies should speak through the dedicated reply speaker.
@@ -1775,15 +1777,6 @@ class NodeRuntime(
17751777
}
17761778
}
17771779

1778-
private fun parseChatSendRunId(response: String): String? {
1779-
return try {
1780-
val root = json.parseToJsonElement(response).asObjectOrNull() ?: return null
1781-
root["runId"].asStringOrNull()
1782-
} catch (_: Throwable) {
1783-
null
1784-
}
1785-
}
1786-
17871780
private fun parseTalkSessionId(response: String): String {
17881781
val root = json.parseToJsonElement(response).asObjectOrNull()
17891782
val sessionId =

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

Lines changed: 33 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package ai.openclaw.app.chat
22

33
import ai.openclaw.app.gateway.GatewaySession
4+
import ai.openclaw.app.gateway.parseChatSendAck
45
import kotlinx.coroutines.CoroutineScope
56
import kotlinx.coroutines.Job
67
import kotlinx.coroutines.delay
@@ -19,11 +20,21 @@ import java.util.UUID
1920
import java.util.concurrent.ConcurrentHashMap
2021
import java.util.concurrent.atomic.AtomicLong
2122

22-
class ChatController(
23+
class ChatController internal constructor(
2324
private val scope: CoroutineScope,
24-
private val session: GatewaySession,
2525
private val json: Json,
26+
private val requestGateway: suspend (method: String, paramsJson: String?) -> String,
2627
) {
28+
constructor(
29+
scope: CoroutineScope,
30+
session: GatewaySession,
31+
json: Json,
32+
) : this(
33+
scope = scope,
34+
json = json,
35+
requestGateway = { method, paramsJson -> session.request(method, paramsJson) },
36+
)
37+
2738
private var appliedMainSessionKey = "main"
2839
private val _sessionKey = MutableStateFlow("main")
2940
val sessionKey: StateFlow<String> = _sessionKey.asStateFlow()
@@ -61,9 +72,11 @@ class ChatController(
6172

6273
private val pendingRuns = mutableSetOf<String>()
6374
private val pendingRunTimeoutJobs = ConcurrentHashMap<String, Job>()
75+
6476
// Preserve sent messages locally until chat.history includes the gateway-confirmed copy.
6577
private val optimisticMessagesByRunId = LinkedHashMap<String, ChatMessage>()
6678
private val pendingRunTimeoutMs = 120_000L
79+
6780
// Drops stale history responses after session switches or refresh races.
6881
private val historyLoadGeneration = AtomicLong(0)
6982

@@ -264,8 +277,9 @@ class ChatController(
264277
)
265278
}
266279
}
267-
val res = session.request("chat.send", params.toString())
268-
val actualRunId = parseRunId(res) ?: runId
280+
val res = requestGateway("chat.send", params.toString())
281+
val ack = parseChatSendAck(json, res)
282+
val actualRunId = ack.runId ?: runId
269283
if (actualRunId != runId) {
270284
// Gateway may return a canonical run id; move all pending bookkeeping to that id.
271285
optimisticMessagesByRunId[actualRunId] = optimisticMessagesByRunId.remove(runId) ?: optimisticMessage
@@ -276,6 +290,16 @@ class ChatController(
276290
_pendingRunCount.value = pendingRuns.size
277291
}
278292
}
293+
if (ack.isTerminal) {
294+
clearPendingRun(actualRunId)
295+
removeOptimisticMessage(actualRunId)
296+
pendingToolCallsById.clear()
297+
publishPendingToolCalls()
298+
_streamingAssistantText.value = null
299+
if (ack.normalizedStatus == "error") {
300+
_errorText.value = "Chat failed before the run started; try again."
301+
}
302+
}
279303
true
280304
} catch (err: Throwable) {
281305
clearPendingRun(runId)
@@ -300,7 +324,7 @@ class ChatController(
300324
put("sessionKey", JsonPrimitive(_sessionKey.value))
301325
put("runId", JsonPrimitive(runId))
302326
}
303-
session.request("chat.abort", params.toString())
327+
requestGateway("chat.abort", params.toString())
304328
} catch (_: Throwable) {
305329
// best-effort
306330
}
@@ -344,7 +368,7 @@ class ChatController(
344368
) {
345369
try {
346370
val historyJson =
347-
session.request(
371+
requestGateway(
348372
"chat.history",
349373
buildJsonObject { put("sessionKey", JsonPrimitive(sessionKey)) }.toString(),
350374
)
@@ -377,7 +401,7 @@ class ChatController(
377401
put("includeUnknown", JsonPrimitive(false))
378402
if (limit != null && limit > 0) put("limit", JsonPrimitive(limit))
379403
}
380-
val res = session.request("sessions.list", params.toString())
404+
val res = requestGateway("sessions.list", params.toString())
381405
_sessions.value = parseSessions(res)
382406
} catch (_: Throwable) {
383407
// best-effort
@@ -390,7 +414,7 @@ class ChatController(
390414
if (!force && last != null && now - last < 10_000) return
391415
lastHealthPollAtMs = now
392416
try {
393-
session.request("health", null)
417+
requestGateway("health", null)
394418
_healthOk.value = true
395419
} catch (_: Throwable) {
396420
_healthOk.value = false
@@ -435,7 +459,7 @@ class ChatController(
435459
val currentSessionKey = _sessionKey.value
436460
val currentGeneration = historyLoadGeneration.get()
437461
val historyJson =
438-
session.request(
462+
requestGateway(
439463
"chat.history",
440464
buildJsonObject { put("sessionKey", JsonPrimitive(currentSessionKey)) }.toString(),
441465
)
@@ -623,17 +647,6 @@ class ChatController(
623647
}
624648
}
625649

626-
private fun parseRunId(resJson: String): String? =
627-
try {
628-
json
629-
.parseToJsonElement(resJson)
630-
.asObjectOrNull()
631-
?.get("runId")
632-
.asStringOrNull()
633-
} catch (_: Throwable) {
634-
null
635-
}
636-
637650
private fun normalizeThinking(raw: String): String =
638651
when (raw.trim().lowercase()) {
639652
"low" -> "low"

‎apps/android/app/src/main/java/ai/openclaw/app/gateway/ChatSendAck.kt‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
package ai.openclaw.app.gateway
2+
3+
import kotlinx.serialization.json.Json
4+
import kotlinx.serialization.json.JsonElement
5+
import kotlinx.serialization.json.JsonObject
6+
import kotlinx.serialization.json.JsonPrimitive
7+
8+
internal data class ChatSendAck(
9+
val runId: String?,
10+
val status: String?,
11+
) {
12+
val normalizedStatus: String
13+
get() = status?.trim()?.lowercase().orEmpty()
14+
15+
val isTerminal: Boolean
16+
get() = normalizedStatus == "timeout" || normalizedStatus == "error"
17+
}
18+
19+
internal fun parseChatSendAck(
20+
json: Json,
21+
responseJson: String,
22+
): ChatSendAck =
23+
try {
24+
val obj = json.parseToJsonElement(responseJson).asObjectOrNull()
25+
ChatSendAck(
26+
runId = obj?.get("runId").asStringOrNull(),
27+
status = obj?.get("status").asStringOrNull(),
28+
)
29+
} catch (_: Throwable) {
30+
ChatSendAck(runId = null, status = null)
31+
}
32+
33+
private fun JsonElement?.asObjectOrNull(): JsonObject? = this as? JsonObject
34+
35+
private fun JsonElement?.asStringOrNull(): String? = (this as? JsonPrimitive)?.takeIf { it.isString }?.content

‎apps/android/app/src/main/java/ai/openclaw/app/voice/MicCaptureManager.kt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@ class MicCaptureManager(
104104
private val messageQueue = ArrayDeque<String>()
105105
private val messageQueueLock = Any()
106106
private var flushedPartialTranscript: String? = null
107+
107108
// Correlates chat events with the idempotency key generated before sendChat returns.
108109
private var pendingRunId: String? = null
109110
private var pendingAssistantEntryId: String? = null
@@ -493,10 +494,12 @@ class MicCaptureManager(
493494
if (runId == null) {
494495
pendingRunTimeoutJob?.cancel()
495496
pendingRunTimeoutJob = null
497+
pendingRunId = null
496498
removeFirstQueuedMessage()
497499
publishQueue()
498500
_isSending.value = false
499501
pendingAssistantEntryId = null
502+
_statusText.value = if (_micEnabled.value) listeningStatus() else "Mic off"
500503
sendQueuedIfIdle()
501504
} else {
502505
armPendingRunTimeout(runId)

‎apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt‎

Lines changed: 28 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package ai.openclaw.app.voice
22

3+
import ai.openclaw.app.gateway.ChatSendAck
34
import ai.openclaw.app.gateway.GatewaySession
5+
import ai.openclaw.app.gateway.parseChatSendAck
46
import android.Manifest
57
import android.annotation.SuppressLint
68
import android.content.Context
@@ -168,6 +170,7 @@ class TalkModeManager internal constructor(
168170
@Volatile private var realtimeSessionId: String? = null
169171
private var realtimeCaptureJob: Job? = null
170172
private var realtimeAppendJob: Job? = null
173+
171174
// Realtime tool calls can complete before their chat final arrives; cache by call/run id until both sides meet.
172175
private val realtimeToolRuns = LinkedHashMap<String, RealtimeToolRun>()
173176
private val pendingRealtimeToolCalls = LinkedHashSet<String>()
@@ -380,7 +383,12 @@ class TalkModeManager internal constructor(
380383
reloadConfig()
381384
val startedAt = System.currentTimeMillis().toDouble() / 1000.0
382385
val prompt = buildPrompt(command)
383-
val runId = sendChat(prompt, session)
386+
val ack = sendChat(prompt, session)
387+
val runId = ack.runId ?: throw IllegalStateException("chat.send returned no run id")
388+
if (ack.isTerminal) {
389+
_statusText.value = if (ack.normalizedStatus == "error") "Chat error" else "Aborted"
390+
return@launch
391+
}
384392
val ok = waitForChatFinal(runId)
385393
val assistant =
386394
consumeRunText(runId)
@@ -397,8 +405,9 @@ class TalkModeManager internal constructor(
397405
}
398406
} catch (err: Throwable) {
399407
Log.w(tag, "speakWakeCommand failed: ${err.message}")
408+
} finally {
409+
onComplete()
400410
}
401-
onComplete()
402411
}
403412
}
404413

@@ -1603,8 +1612,14 @@ class TalkModeManager internal constructor(
16031612
try {
16041613
val startedAt = System.currentTimeMillis().toDouble() / 1000.0
16051614
Log.d(tag, "chat.send start sessionKey=${mainSessionKey.ifBlank { "main" }} chars=${prompt.length}")
1606-
val runId = sendChat(prompt, session)
1607-
Log.d(tag, "chat.send ok runId=$runId")
1615+
val ack = sendChat(prompt, session)
1616+
val runId = ack.runId ?: throw IllegalStateException("chat.send returned no run id")
1617+
Log.d(tag, "chat.send ok runId=$runId status=${ack.status}")
1618+
if (ack.isTerminal) {
1619+
_statusText.value = if (ack.normalizedStatus == "error") "Chat error" else "Aborted"
1620+
start()
1621+
return
1622+
}
16081623
val ok = waitForChatFinal(runId)
16091624
if (!ok) {
16101625
Log.w(tag, "chat final timeout runId=$runId; attempting history fallback")
@@ -1678,7 +1693,7 @@ class TalkModeManager internal constructor(
16781693
private suspend fun sendChat(
16791694
message: String,
16801695
session: GatewaySession,
1681-
): String {
1696+
): ChatSendAck {
16821697
val runId = UUID.randomUUID().toString()
16831698
armPendingRun(runId)
16841699
val params =
@@ -1691,11 +1706,15 @@ class TalkModeManager internal constructor(
16911706
}
16921707
try {
16931708
val res = session.request("chat.send", params.toString())
1694-
val parsed = parseRunId(res) ?: runId
1695-
if (parsed != runId) {
1696-
pendingRunId = parsed
1709+
val parsed = parseChatSendAck(json, res)
1710+
val actualRunId = parsed.runId ?: runId
1711+
if (actualRunId != runId) {
1712+
pendingRunId = actualRunId
1713+
}
1714+
if (parsed.isTerminal) {
1715+
clearPendingRun(actualRunId)
16971716
}
1698-
return parsed
1717+
return parsed.copy(runId = actualRunId)
16991718
} catch (err: Throwable) {
17001719
clearPendingRun(runId)
17011720
throw err

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

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package ai.openclaw.app.chat
2+
3+
import kotlinx.coroutines.ExperimentalCoroutinesApi
4+
import kotlinx.coroutines.test.runTest
5+
import kotlinx.serialization.json.Json
6+
import org.junit.Assert.assertEquals
7+
import org.junit.Assert.assertFalse
8+
import org.junit.Assert.assertNull
9+
import org.junit.Assert.assertTrue
10+
import org.junit.Test
11+
12+
class ChatControllerTerminalAckTest {
13+
private val json = Json { ignoreUnknownKeys = true }
14+
15+
@Test
16+
@OptIn(ExperimentalCoroutinesApi::class)
17+
fun terminalTimeoutAckRemovesOptimisticUserEchoWithoutErrorText() =
18+
runTest {
19+
var requestedMethod: String? = null
20+
val controller =
21+
ChatController(
22+
scope = this,
23+
json = json,
24+
requestGateway = { method, _ ->
25+
requestedMethod = method
26+
"""{"runId":"run-timeout","status":"timeout"}"""
27+
},
28+
)
29+
controller.handleGatewayEvent("health", null)
30+
31+
val accepted =
32+
controller.sendMessageAwaitAcceptance(
33+
message = "message that times out before start",
34+
thinkingLevel = "off",
35+
attachments = emptyList(),
36+
)
37+
38+
assertTrue(accepted)
39+
assertEquals("chat.send", requestedMethod)
40+
assertEquals(0, controller.pendingRunCount.value)
41+
assertNull(controller.errorText.value)
42+
assertFalse(controller.messages.value.hasUserText("message that times out before start"))
43+
}
44+
45+
@Test
46+
@OptIn(ExperimentalCoroutinesApi::class)
47+
fun terminalErrorAckRemovesOptimisticUserEchoAndSurfacesErrorText() =
48+
runTest {
49+
val controller =
50+
ChatController(
51+
scope = this,
52+
json = json,
53+
requestGateway = { _, _ -> """{"runId":"run-error","status":"error"}""" },
54+
)
55+
controller.handleGatewayEvent("health", null)
56+
57+
val accepted =
58+
controller.sendMessageAwaitAcceptance(
59+
message = "message that errors before start",
60+
thinkingLevel = "off",
61+
attachments = emptyList(),
62+
)
63+
64+
assertTrue(accepted)
65+
assertEquals(0, controller.pendingRunCount.value)
66+
assertEquals("Chat failed before the run started; try again.", controller.errorText.value)
67+
assertFalse(controller.messages.value.hasUserText("message that errors before start"))
68+
}
69+
70+
private fun List<ChatMessage>.hasUserText(text: String): Boolean =
71+
any { message ->
72+
message.role == "user" && message.content.any { part -> part.text == text }
73+
}
74+
}

0 commit comments

Comments
 (0)