11package ai.openclaw.app.chat
2233import ai.openclaw.app.gateway.GatewaySession
4+import ai.openclaw.app.gateway.parseChatSendAck
45import kotlinx.coroutines.CoroutineScope
56import kotlinx.coroutines.Job
67import kotlinx.coroutines.delay
@@ -19,11 +20,21 @@ import java.util.UUID
1920import java.util.concurrent.ConcurrentHashMap
2021import java.util.concurrent.atomic.AtomicLong
212222-class ChatController(
23+class ChatController internal constructor(
2324private val scope: CoroutineScope,
24-private val session: GatewaySession,
2525private 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+2738private var appliedMainSessionKey = "main"
2839private val _sessionKey = MutableStateFlow("main")
2940val sessionKey: StateFlow<String> = _sessionKey.asStateFlow()
@@ -267,8 +278,9 @@ class ChatController(
267278 )
268279 }
269280 }
270-val res = session.request("chat.send", params.toString())
271-val actualRunId = parseRunId(res) ?: runId
281+val res = requestGateway("chat.send", params.toString())
282+val ack = parseChatSendAck(json, res)
283+val actualRunId = ack.runId ?: runId
272284if (actualRunId != runId) {
273285// Gateway may return a canonical run id; move all pending bookkeeping to that id.
274286 optimisticMessagesByRunId[actualRunId] = optimisticMessagesByRunId.remove(runId) ?: optimisticMessage
@@ -279,7 +291,24 @@ class ChatController(
279291_pendingRunCount.value = pendingRuns.size
280292 }
281293 }
282-true
294+if (ack.isTerminal) {
295+ clearPendingRun(actualRunId)
296+ removeOptimisticMessage(actualRunId)
297+ pendingToolCallsById.clear()
298+ publishPendingToolCalls()
299+_streamingAssistantText.value = null
300+if (ack.isTerminalSuccess) {
301+ refreshCurrentHistoryBestEffort()
302+true
303+ } else {
304+// Terminal timeout/error means the gateway did not accept a runnable turn.
305+// Surface failed acceptance instead of letting a cleared composer look successful.
306+_errorText.value = "Chat failed before the run started; try again."
307+false
308+ }
309+ } else {
310+true
311+ }
283312 } catch (err: Throwable) {
284313 clearPendingRun(runId)
285314 removeOptimisticMessage(runId)
@@ -303,7 +332,7 @@ class ChatController(
303332 put("sessionKey", JsonPrimitive(_sessionKey.value))
304333 put("runId", JsonPrimitive(runId))
305334 }
306-session.request("chat.abort", params.toString())
335+requestGateway("chat.abort", params.toString())
307336 } catch (_: Throwable) {
308337// best-effort
309338 }
@@ -356,7 +385,7 @@ class ChatController(
356385 ) {
357386try {
358387val historyJson =
359-session.request(
388+requestGateway(
360389"chat.history",
361390 buildJsonObject { put("sessionKey", JsonPrimitive(sessionKey)) }.toString(),
362391 )
@@ -391,7 +420,7 @@ class ChatController(
391420 put("includeUnknown", JsonPrimitive(false))
392421if (limit != null && limit > 0) put("limit", JsonPrimitive(limit))
393422 }
394-val res = session.request("sessions.list", params.toString())
423+val res = requestGateway("sessions.list", params.toString())
395424_sessions.value = parseSessions(res)
396425 } catch (_: Throwable) {
397426// best-effort
@@ -408,7 +437,7 @@ class ChatController(
408437if (!force && last != null && now - last < 10_000) return
409438 lastHealthPollAtMs = now
410439try {
411-session.request("health", null)
440+requestGateway("health", null)
412441_healthOk.value = true
413442 } catch (_: Throwable) {
414443_healthOk.value = false
@@ -451,7 +480,7 @@ class ChatController(
451480val currentSessionKey = _sessionKey.value
452481val currentGeneration = historyLoadGeneration.get()
453482val historyJson =
454-session.request(
483+requestGateway(
455484"chat.history",
456485 buildJsonObject { put("sessionKey", JsonPrimitive(currentSessionKey)) }.toString(),
457486 )
@@ -632,6 +661,45 @@ class ChatController(
632661 optimisticMessagesByRunId.entries.removeAll { entry -> entry.value !in retained }
633662 }
634663664+private fun refreshCurrentHistoryBestEffort() {
665+ scope.launch {
666+try {
667+val currentSessionKey = _sessionKey.value
668+val currentGeneration = historyLoadGeneration.get()
669+val historyJson =
670+ requestGateway(
671+"chat.history",
672+ buildJsonObject { put("sessionKey", JsonPrimitive(currentSessionKey)) }.toString(),
673+ )
674+if (
675+!isCurrentHistoryLoad(
676+ currentSessionKey,
677+_sessionKey.value,
678+ currentGeneration,
679+ historyLoadGeneration.get(),
680+ )
681+ ) {
682+return@launch
683+ }
684+val history =
685+ parseHistory(
686+ historyJson,
687+ sessionKey = currentSessionKey,
688+ previousMessages = _messages.value,
689+ )
690+ prunePersistedOptimisticMessages(history.messages)
691+_messages.value = mergeOptimisticMessages(incoming = history.messages, optimistic = optimisticMessagesByRunId.values)
692+_sessionId.value = history.sessionId
693+ history.thinkingLevel
694+ ?.trim()
695+ ?.takeIf { it.isNotEmpty() }
696+ ?.let { _thinkingLevel.value = it }
697+ } catch (_: Throwable) {
698+// best-effort
699+ }
700+ }
701+ }
702+635703private fun parseHistory(
636704historyJson: String,
637705sessionKey: String,
@@ -728,17 +796,6 @@ class ChatController(
728796_sessions.value = _sessions.value.filterNot { it.key == key }
729797 }
730798731-private fun parseRunId(resJson: String): String? =
732-try {
733- json
734- .parseToJsonElement(resJson)
735- .asObjectOrNull()
736- ?.get("runId")
737- .asStringOrNull()
738- } catch (_: Throwable) {
739-null
740- }
741-742799private fun normalizeThinking(raw: String): String =
743800when (raw.trim().lowercase()) {
744801"low" -> "low"
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。