From 2b48e98148405441b3aa7812ca643a114228dace Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 6 Jul 2026 03:50:09 +0100 Subject: [PATCH] fix(android): preserve chat sends across reconnect recovery (#100551) * fix(android): preserve chat recovery ownership * fix(android): retry reconnect recovery on ticks * chore(android): refresh native i18n inventory --- CHANGELOG.md | 1 + apps/.i18n/native-source.json | 62 +- .../java/ai/openclaw/app/MainViewModel.kt | 31 +- .../main/java/ai/openclaw/app/NodeRuntime.kt | 4 +- .../ai/openclaw/app/chat/ChatController.kt | 561 +++++++-- .../ai/openclaw/app/gateway/GatewaySession.kt | 27 +- .../ai/openclaw/app/ui/chat/ChatScreen.kt | 19 +- .../openclaw/app/ui/chat/ChatSheetContent.kt | 38 +- .../ChatControllerReconnectRestoreTest.kt | 1051 ++++++++++++++++- .../chat/ChatControllerStreamReplayTest.kt | 137 ++- .../app/chat/ChatControllerTerminalAckTest.kt | 99 ++ .../ai/openclaw/app/chat/ChatReplayHarness.kt | 40 +- .../app/gateway/GatewaySessionInvokeTest.kt | 54 +- .../app/ui/chat/ChatSheetContentTest.kt | 41 - 14 files changed, 1941 insertions(+), 224 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f7e5651e2f0..2d9f8e6d0a5c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,7 @@ Docs: https://docs.openclaw.ai - **Android microphone capture:** treat negative `AudioRecord.read` results as fatal shared-session errors so both transcription and Talk capture stop cleanly after device loss. (#100028) Thanks @NianJiuZst. - **Lean local-model tools:** trim media generation, TTS, and PDF tools from lean agent surfaces while preserving explicit config and runtime opt-ins. (#88881) Thanks @vincentkoc. - **iOS development app identity:** keep the development app labeled OpenClaw while using its distinct debug icon to differentiate it from release builds. +- **Android chat recovery:** preserve optimistic user messages and locally owned runs while reconnect and sequence-gap history snapshots catch up, preventing sent messages from disappearing or stale runs from taking ownership. (#100197) - **iOS QR gateway handoff:** stop VisionKit before delivering scanned setup codes, and keep deferred auth, approval, Watch, and foreground-node work bound to its originating gateway across reconnects. (#99572) Thanks @PollyBot13. - **Agent terminal failures:** surface a safe interactive reply when an agent run ends without visible output, while preserving completed message-tool delivery and heartbeat-specific guidance. (#99304) Thanks @moeedahmed. - **MCP loopback tool results:** preserve schema-valid text, image, and embedded-resource content through HTTP tool calls while rendering malformed or protocol-incompatible blocks as safe text. (#100336) Thanks @tzy-17. diff --git a/apps/.i18n/native-source.json b/apps/.i18n/native-source.json index 25f163d0191a..7af30e86de55 100644 --- a/apps/.i18n/native-source.json +++ b/apps/.i18n/native-source.json @@ -123,7 +123,7 @@ }, { "kind": "ui-state-text", - "line": 101, + "line": 103, "path": "apps/android/app/src/main/java/ai/openclaw/app/MainViewModel.kt", "source": "Searching…", "surface": "android", @@ -131,7 +131,7 @@ }, { "kind": "ui-state-text", - "line": 190, + "line": 192, "path": "apps/android/app/src/main/java/ai/openclaw/app/MainViewModel.kt", "source": "Mic off", "surface": "android", @@ -139,7 +139,7 @@ }, { "kind": "ui-state-text", - "line": 200, + "line": 202, "path": "apps/android/app/src/main/java/ai/openclaw/app/MainViewModel.kt", "source": "Off", "surface": "android", @@ -371,7 +371,7 @@ }, { "kind": "conditional-branch", - "line": 1185, + "line": 1200, "path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt", "source": "Connecting…", "surface": "android", @@ -379,7 +379,7 @@ }, { "kind": "conditional-branch", - "line": 1185, + "line": 1200, "path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt", "source": "Reconnecting…", "surface": "android", @@ -5203,7 +5203,7 @@ }, { "kind": "ui-named-argument", - "line": 225, + "line": 226, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Chat needs attention", "surface": "android", @@ -5211,7 +5211,7 @@ }, { "kind": "ui-named-argument", - "line": 366, + "line": 367, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "All", "surface": "android", @@ -5219,7 +5219,7 @@ }, { "kind": "ui-named-argument", - "line": 420, + "line": 421, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "OpenClaw", "surface": "android", @@ -5227,7 +5227,7 @@ }, { "kind": "ui-named-argument", - "line": 441, + "line": 442, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "New chat", "surface": "android", @@ -5235,7 +5235,7 @@ }, { "kind": "ui-named-argument", - "line": 442, + "line": 443, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Refresh chat", "surface": "android", @@ -5243,7 +5243,7 @@ }, { "kind": "ui-named-argument", - "line": 445, + "line": 446, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Chat", "surface": "android", @@ -5251,7 +5251,7 @@ }, { "kind": "ui-named-argument", - "line": 582, + "line": 583, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Loading session", "surface": "android", @@ -5259,7 +5259,7 @@ }, { "kind": "ui-named-argument", - "line": 609, + "line": 610, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Jump to latest", "surface": "android", @@ -5267,7 +5267,7 @@ }, { "kind": "conditional-branch", - "line": 629, + "line": 630, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Ready when you are", "surface": "android", @@ -5275,7 +5275,7 @@ }, { "kind": "ui-named-argument", - "line": 660, + "line": 661, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Fix connection", "surface": "android", @@ -5283,7 +5283,7 @@ }, { "kind": "ui-named-argument", - "line": 661, + "line": 662, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Copy diagnostics", "surface": "android", @@ -5291,7 +5291,7 @@ }, { "kind": "ui-named-argument", - "line": 799, + "line": 800, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Tools running", "surface": "android", @@ -5299,7 +5299,7 @@ }, { "kind": "ui-named-argument", - "line": 804, + "line": 805, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "+${toolCalls.size - 4} more", "surface": "android", @@ -5307,7 +5307,7 @@ }, { "kind": "ui-named-argument", - "line": 814, + "line": 815, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Thinking", "surface": "android", @@ -5315,7 +5315,7 @@ }, { "kind": "ui-named-argument", - "line": 815, + "line": 816, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "OpenClaw is preparing a response.", "surface": "android", @@ -5323,7 +5323,7 @@ }, { "kind": "ui-named-argument", - "line": 934, + "line": 935, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Stop", "surface": "android", @@ -5331,7 +5331,7 @@ }, { "kind": "ui-named-argument", - "line": 951, + "line": 952, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "No commands found", "surface": "android", @@ -5339,7 +5339,7 @@ }, { "kind": "ui-named-argument", - "line": 1013, + "line": 1014, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Gateway offline", "surface": "android", @@ -5347,7 +5347,7 @@ }, { "kind": "ui-named-argument", - "line": 1103, + "line": 1104, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Attach image", "surface": "android", @@ -5355,7 +5355,7 @@ }, { "kind": "ui-named-argument", - "line": 1118, + "line": 1119, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Message OpenClaw", "surface": "android", @@ -5363,7 +5363,7 @@ }, { "kind": "ui-named-argument", - "line": 1133, + "line": 1134, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Voice", "surface": "android", @@ -5371,7 +5371,7 @@ }, { "kind": "ui-named-argument", - "line": 1171, + "line": 1172, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Remove attachment", "surface": "android", @@ -5379,7 +5379,7 @@ }, { "kind": "conditional-branch", - "line": 1192, + "line": 1193, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Main", "surface": "android", @@ -5387,7 +5387,7 @@ }, { "kind": "ui-named-argument", - "line": 1248, + "line": 1249, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "Send", "surface": "android", @@ -5395,7 +5395,7 @@ }, { "kind": "conditional-branch", - "line": 1287, + "line": 1288, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt", "source": "$contextLabel · ${contextMeterThinkingLabel(thinkingLevel)}", "surface": "android", @@ -5403,7 +5403,7 @@ }, { "kind": "ui-named-argument", - "line": 287, + "line": 269, "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatSheetContent.kt", "source": "CHAT ERROR", "surface": "android", diff --git a/apps/android/app/src/main/java/ai/openclaw/app/MainViewModel.kt b/apps/android/app/src/main/java/ai/openclaw/app/MainViewModel.kt index 5c677936e326..563dc7934cf8 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/MainViewModel.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/MainViewModel.kt @@ -51,6 +51,8 @@ class MainViewModel( val chatDraft: StateFlow = _chatDraft private val _pendingAssistantAutoSend = MutableStateFlow(null) val pendingAssistantAutoSend: StateFlow = _pendingAssistantAutoSend + private val _assistantAutoSendInFlight = MutableStateFlow(false) + val assistantAutoSendInFlight: StateFlow = _assistantAutoSendInFlight /** * Lazily starts NodeRuntime and preserves the current foreground bit across startup. @@ -428,8 +430,33 @@ class MainViewModel( _chatDraft.value = null } - fun clearPendingAssistantAutoSend() { - _pendingAssistantAutoSend.value = null + /** Claims an assistant prompt before sending so Compose effect restarts cannot dispatch it twice. */ + fun dispatchPendingAssistantAutoSend( + pendingPrompt: String, + thinking: String, + ) { + val prompt = pendingPrompt.trim().ifEmpty { return } + if (!chatHealthOk.value || pendingRunCount.value > 0) return + if (!_assistantAutoSendInFlight.compareAndSet(false, true)) return + if (_pendingAssistantAutoSend.value != pendingPrompt) { + _assistantAutoSendInFlight.value = false + return + } + viewModelScope.launch { + try { + sendChatAwaitAcceptance( + message = prompt, + thinking = thinking, + attachments = emptyList(), + ) + // A definitive rejection is surfaced by chatError; it must not strand the + // one-shot assistant prompt or overwrite text typed into the composer. + _pendingAssistantAutoSend.compareAndSet(pendingPrompt, null) + } finally { + // Observable release wakes a newer prompt queued while this send was in flight. + _assistantAutoSendInFlight.value = false + } + } } fun setMicEnabled(enabled: Boolean) { diff --git a/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt b/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt index e241ca616ba9..ee3d398fcf6b 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt @@ -717,8 +717,8 @@ class NodeRuntime private constructor( _gatewayUpdateAvailable.value = hello.updateAvailable _seamColorArgb.value = DEFAULT_SEAM_COLOR_ARGB syncMainSessionKey(resolveAgentIdFromMainSessionKey(hello.mainSessionKey)) - // Every successful connection refreshes history, including reconnects whose main key did not change. - chat.refresh() + // Every successful connection refreshes history; reconnects preserve local run ownership. + chat.onGatewayConnected() refreshGatewayControlPage() updateStatus { operatorConnectionProblem = null diff --git a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt index 40d36cc04793..c72156d784cf 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt @@ -1,5 +1,7 @@ package ai.openclaw.app.chat +import ai.openclaw.app.gateway.GatewayRequestDefinitiveFailure +import ai.openclaw.app.gateway.GatewayRequestOutcomeUnknown import ai.openclaw.app.gateway.GatewaySession import ai.openclaw.app.gateway.parseChatSendAck import ai.openclaw.app.resolveAgentIdFromMainSessionKey @@ -99,15 +101,29 @@ class ChatController internal constructor( val commands: StateFlow> = _commands.asStateFlow() private val pendingRuns = mutableSetOf() + private val disconnectedPendingRunIds = mutableSetOf() + private val timedOutRunIds = ConcurrentHashMap.newKeySet() + private val terminalWithoutReplyRunIds = ConcurrentHashMap.newKeySet() + private val unknownOutcomeRunIds = ConcurrentHashMap.newKeySet() private val pendingRunTimeoutJobs = ConcurrentHashMap() // Preserve sent messages locally until chat.history includes the gateway-confirmed copy. - private val optimisticMessagesByRunId = LinkedHashMap() + private val optimisticMessagesByRunId = ConcurrentHashMap() + // Keep reply ownership after the user row persists; the assistant row can land later. + private val unresolvedRepliesByRunId = ConcurrentHashMap() private val pendingRunTimeoutMs = 120_000L + private val recoveryHistoryRetryDelayMs = 750L + private var recoveryHistoryReconciliationGeneration = -1L + private var recoveryHistoryReconciliationJob: Job? = null // Drops stale history responses after session switches or refresh races. private val historyLoadGeneration = AtomicLong(0) + private val historyRequestSequence = AtomicLong(0) private val gatewayScopeApplyLock = Any() + private var latestAppliedHistoryRequest = 0L + private var latestAppliedInFlightRunId: String? = null + private var lastHandledTerminalRunId: String? = null + private var historyLoadErrorGeneration: Long? = null private val newChatCreateInFlight = AtomicBoolean(false) private var lastHealthPollAtMs: Long? = null @@ -116,6 +132,15 @@ class ChatController internal constructor( // Armed on disconnect so the next health event refetches history and re-adopts // any run the gateway still reports in flight (chat.history `inFlightRun`). private var restoreRunStateOnReconnect = false + private var reconnectRecoveryGeneration: Long? = null + + private fun updateErrorText( + message: String?, + historyGeneration: Long? = null, + ) { + _errorText.value = message + historyLoadErrorGeneration = historyGeneration + } private val _outboxItems = MutableStateFlow>(emptyList()) val outboxItems: StateFlow> = _outboxItems.asStateFlow() @@ -138,12 +163,22 @@ class ChatController internal constructor( /** Clears transient chat state when the operator gateway session disconnects. */ fun onDisconnected(message: String) { + historyLoadGeneration.incrementAndGet() restoreRunStateOnReconnect = true + reconnectRecoveryGeneration = null _healthOk.value = false - _errorText.value = null + updateErrorText(null) _commands.value = emptyList() commandsAgentId = null - clearPendingRuns() + synchronized(pendingRuns) { + disconnectedPendingRunIds.addAll(pendingRuns) + } + // History can lag the accepted send. Keep the optimistic echo available for the + // reconnect snapshot to reconcile instead of dropping the user's message. + clearPendingRuns( + clearOptimisticMessages = false, + preserveDisconnectedOwnership = true, + ) pendingToolCallsById.clear() publishPendingToolCalls() _streamingAssistantText.value = null @@ -151,6 +186,16 @@ class ChatController internal constructor( _sessionId.value = null } + /** Refreshes the connected gateway while preserving recovery ownership after a disconnect. */ + fun onGatewayConnected() { + if (!restoreRunStateOnReconnect) { + refresh() + return + } + updateErrorText(null) + refreshHistoryForRecovery(forceHealth = true, completesReconnectRecovery = true) + } + /** Invalidates and clears gateway-bound UI state before a target switch can race old responses. */ fun onGatewayScopeChanging() { synchronized(gatewayScopeApplyLock) { @@ -177,7 +222,11 @@ class ChatController internal constructor( /** Loads a chat session, normalizing "main" to the current gateway-provided main session key. */ fun load(sessionKey: String) { val key = normalizeRequestedSessionKey(sessionKey) - val generation = beginHistoryLoad(key, clearMessages = key != _sessionKey.value) + if (key == _sessionKey.value) { + refresh() + return + } + val generation = beginHistoryLoad(key, clearMessages = true) scope.launch { bootstrap(sessionKey = key, generation = generation, forceHealth = true, refreshSessions = true) } @@ -208,11 +257,8 @@ class ChatController internal constructor( /** Refreshes current chat history and session list without clearing optimistic messages first. */ fun refresh() { - val key = normalizeRequestedSessionKey(_sessionKey.value) - val generation = beginHistoryLoad(key, clearMessages = false) - scope.launch { - bootstrap(sessionKey = key, generation = generation, forceHealth = true, refreshSessions = true) - } + updateErrorText(null) + refreshHistoryForRecovery(forceHealth = true) } fun refreshSessions(limit: Int? = null) { @@ -229,14 +275,14 @@ class ChatController internal constructor( val parentKey = normalizeRequestedSessionKey(_sessionKey.value) if (parentKey.isEmpty()) return false if (_pendingRunCount.value > 0) { - _errorText.value = "Wait for the current response to finish before starting a new chat." + updateErrorText("Wait for the current response to finish before starting a new chat.") return false } if (!newChatCreateInFlight.compareAndSet(false, true)) { return false } val requestGeneration = historyLoadGeneration.get() - _errorText.value = null + updateErrorText(null) _historyLoading.value = true return try { val label = nextNewChatSessionLabel(_sessions.value) @@ -259,7 +305,7 @@ class ChatController internal constructor( bootstrap(sessionKey = createdKey, generation = generation, forceHealth = true, refreshSessions = true) true } catch (err: Throwable) { - _errorText.value = err.message + updateErrorText(err.message) _historyLoading.value = false false } finally { @@ -297,12 +343,13 @@ class ChatController internal constructor( ): Long { val generation = historyLoadGeneration.incrementAndGet() _sessionKey.value = key + lastHandledTerminalRunId = null val nextAgentId = resolveAgentIdForSessionKey(key) if (commandsAgentId != nextAgentId) { _commands.value = emptyList() commandsAgentId = null } - _errorText.value = null + updateErrorText(null) _healthOk.value = false clearPendingRuns() pendingToolCallsById.clear() @@ -353,7 +400,7 @@ class ChatController internal constructor( // Offline capture: text-only commands become durable outbox rows and flush on reconnect. // Attachments stay blocked (text-only v1) so large payloads never sit in the database. if (commandOutbox == null || attachments.isNotEmpty()) { - _errorText.value = "Gateway health not OK; cannot send" + updateErrorText("Gateway health not OK; cannot send") return false } return enqueueOfflineCommand(text = trimmed, thinkingLevel = normalizeThinking(thinkingLevel)) @@ -386,8 +433,9 @@ class ChatController internal constructor( content = userContent, timestampMs = System.currentTimeMillis(), idempotencyKey = "$runId:user", - ) + ) optimisticMessagesByRunId[runId] = optimisticMessage + unresolvedRepliesByRunId[runId] = optimisticMessage _messages.value = _messages.value + optimisticMessage armPendingRunTimeout(runId) @@ -396,7 +444,7 @@ class ChatController internal constructor( _pendingRunCount.value = pendingRuns.size } - _errorText.value = null + updateErrorText(null) _streamingAssistantText.value = null pendingToolCallsById.clear() publishPendingToolCalls() @@ -429,14 +477,7 @@ class ChatController internal constructor( val ack = parseChatSendAck(json, res) val actualRunId = ack.runId ?: runId if (actualRunId != runId) { - // Gateway may return a canonical run id; move all pending bookkeeping to that id. - optimisticMessagesByRunId[actualRunId] = optimisticMessagesByRunId.remove(runId) ?: optimisticMessage - clearPendingRun(runId) - armPendingRunTimeout(actualRunId) - synchronized(pendingRuns) { - pendingRuns.add(actualRunId) - _pendingRunCount.value = pendingRuns.size - } + transferRunOwnership(runId, actualRunId, optimisticMessage) } if (ack.isTerminal) { clearPendingRun(actualRunId) @@ -445,21 +486,40 @@ class ChatController internal constructor( publishPendingToolCalls() _streamingAssistantText.value = null if (ack.isTerminalSuccess) { - refreshCurrentHistoryBestEffort() + unresolvedRepliesByRunId.remove(actualRunId) + refreshCurrentHistoryBestEffort(runIdsToReconcile = setOf(actualRunId)) true } else { // Terminal timeout/error means the gateway did not accept a runnable turn. // Surface failed acceptance instead of letting a cleared composer look successful. - _errorText.value = "Chat failed before the run started; try again." + unresolvedRepliesByRunId.remove(actualRunId) + updateErrorText("Chat failed before the run started; try again.") false } } else { true } + } catch (err: CancellationException) { + throw err + } catch (err: GatewayRequestDefinitiveFailure) { + clearPendingRun(runId) + removeOptimisticMessage(runId) + unresolvedRepliesByRunId.remove(runId) + updateErrorText(err.message) + false + } catch (_: GatewayRequestOutcomeUnknown) { + // A transport failure cannot distinguish rejection from an accepted send whose + // ACK was lost. Keep the idempotency-key-backed row to prevent a duplicate retry. + unknownOutcomeRunIds.add(runId) + if (_healthOk.value) { + refreshCurrentHistoryBestEffort(runIdsToReconcile = setOf(runId)) + } + true } catch (err: Throwable) { clearPendingRun(runId) removeOptimisticMessage(runId) - _errorText.value = err.message + unresolvedRepliesByRunId.remove(runId) + updateErrorText(err.message) false } } @@ -493,20 +553,23 @@ class ChatController internal constructor( ) { when (event) { "tick" -> { - scope.launch { pollHealthIfNeeded(force = false) } + if (restoreRunStateOnReconnect) { + refreshHistoryForRecovery(forceHealth = true, completesReconnectRecovery = true) + } else { + scope.launch { pollHealthIfNeeded(force = false) } + } } "health" -> { - markHealthOk() - refreshCommandsAfterReconnect() if (restoreRunStateOnReconnect) { - restoreRunStateOnReconnect = false - refreshHistoryForRecovery() + refreshHistoryForRecovery(forceHealth = true, completesReconnectRecovery = true) + } else { + markHealthOk() + refreshCommandsAfterReconnect() } } "seqGap" -> { // Missed events may include deltas or the terminal state of a pending run; - // refetch history and rebuild run state from the gateway snapshot. - clearPendingRuns() + // retain local ownership until the recovery snapshot can reconcile it. pendingToolCallsById.clear() publishPendingToolCalls() _streamingAssistantText.value = null @@ -536,16 +599,44 @@ class ChatController internal constructor( /** * Reconnect/seq-gap recovery: refetch history for the current session without the - * beginHistoryLoad transient-state reset. Disconnect (or the seqGap handler) already - * cleared run state, and resetting healthOk here would block sends after reconnect. + * beginHistoryLoad transient-state reset. Runs pending when the request begins stay + * owned until that authoritative snapshot resolves them; resetting healthOk here + * would block sends after reconnect. */ - private fun refreshHistoryForRecovery() { + private fun refreshHistoryForRecovery( + forceHealth: Boolean = false, + completesReconnectRecovery: Boolean = false, + ) { val key = normalizeRequestedSessionKey(_sessionKey.value) val generation = historyLoadGeneration.incrementAndGet() + if (completesReconnectRecovery) { + synchronized(gatewayScopeApplyLock) { + reconnectRecoveryGeneration = generation + } + } + val restoredRunIds = + synchronized(pendingRuns) { + val restored = disconnectedPendingRunIds.toSet() + pendingRuns.addAll(restored) + disconnectedPendingRunIds.clear() + _pendingRunCount.value = pendingRuns.size + restored + } + restoredRunIds.forEach(::armPendingRunTimeout) + val runIdsToReconcile = + synchronized(pendingRuns) { + pendingRuns + optimisticMessagesByRunId.keys + unresolvedRepliesByRunId.keys + } _sessionKey.value = key _historyLoading.value = true scope.launch { - bootstrap(sessionKey = key, generation = generation, forceHealth = false, refreshSessions = true) + bootstrap( + sessionKey = key, + generation = generation, + forceHealth = forceHealth, + refreshSessions = true, + runIdsToReconcile = runIdsToReconcile, + ) } } @@ -554,22 +645,43 @@ class ChatController internal constructor( generation: Long, forceHealth: Boolean, refreshSessions: Boolean, + runIdsToReconcile: Set = emptySet(), ) { + val ownsReconnectRecovery = + synchronized(gatewayScopeApplyLock) { + reconnectRecoveryGeneration == generation + } // Cache-first cold open: prime before the live request so ordering is deterministic and the // live chat.history response always replaces cached rows wholesale. primeFromCache(sessionKey, generation) try { - if (!fetchAndApplyHistory(sessionKey, generation, updateSessionInfo = true)) return - _historyLoading.value = false + val historyApplied = + fetchAndApplyHistory( + sessionKey, + generation, + updateSessionInfo = true, + runIdsToReconcile = runIdsToReconcile, + ) + if (!historyApplied) return - pollHealthIfNeeded(force = forceHealth) + if (!ownsReconnectRecovery) { + pollHealthIfNeeded(force = forceHealth) + } if (refreshSessions) { fetchSessions(limit = 50) } } catch (err: Throwable) { if (!isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get())) return - _errorText.value = err.message + updateErrorText(err.message, historyGeneration = generation) _historyLoading.value = false + } finally { + if (isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get())) { + scheduleRecoveryHistoryReconciliation( + sessionKey = sessionKey, + generation = generation, + runIds = runIdsToReconcile, + ) + } } } @@ -581,29 +693,83 @@ class ChatController internal constructor( sessionKey: String, generation: Long, updateSessionInfo: Boolean, + runIdsToReconcile: Set = emptySet(), ): Boolean { + val requestSequence = historyRequestSequence.incrementAndGet() val requestCacheScope = currentCacheScope() - val historyJson = - requestGateway( - "chat.history", - buildJsonObject { put("sessionKey", JsonPrimitive(sessionKey)) }.toString(), - ) - val history = parseHistory(historyJson, sessionKey = sessionKey, previousMessages = _messages.value) + val history = + try { + val historyJson = + requestGateway( + "chat.history", + buildJsonObject { put("sessionKey", JsonPrimitive(sessionKey)) }.toString(), + ) + parseHistory(historyJson, sessionKey = sessionKey, previousMessages = _messages.value) + } catch (err: CancellationException) { + throw err + } catch (err: Throwable) { + val superseded = + synchronized(gatewayScopeApplyLock) { + !isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get()) || + requestCacheScope != currentCacheScope() || + requestSequence < latestAppliedHistoryRequest + } + if (superseded) return false + throw err + } val applied = synchronized(gatewayScopeApplyLock) { if ( !isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get()) || - requestCacheScope != currentCacheScope() + requestCacheScope != currentCacheScope() || + requestSequence < latestAppliedHistoryRequest ) { return@synchronized false } + latestAppliedHistoryRequest = requestSequence if (updateSessionInfo) { updateSessionFromHistory(history) } + transferLostAckOwnershipFromHistory(history) + resolvePersistedReplies(history.messages) + val snapshotRunId = history.inFlightRun?.runId?.trim()?.takeIf { it.isNotEmpty() } + latestAppliedInFlightRunId = snapshotRunId + val optimisticRunIds = runIdsToReconcile.filterTo(mutableSetOf()) { optimisticMessagesByRunId.containsKey(it) } prunePersistedOptimisticMessages(history.messages) + if (snapshotRunId == null) { + optimisticRunIds + .filterNot { runId -> + unknownOutcomeRunIds.contains(runId) && unresolvedRepliesByRunId.containsKey(runId) + } + .filterNotTo(mutableSetOf()) { optimisticMessagesByRunId.containsKey(it) } + .forEach(::clearPendingRun) + } + if (snapshotRunId != null) { + runIdsToReconcile + .filterTo(mutableSetOf()) { + it != snapshotRunId && + !optimisticMessagesByRunId.containsKey(it) && + !unresolvedRepliesByRunId.containsKey(it) + } + .forEach(::clearPendingRun) + } _messagesFromCache.value = false _messages.value = mergeOptimisticMessages(incoming = history.messages, optimistic = optimisticMessagesByRunId.values) _sessionId.value = history.sessionId + _historyLoading.value = false + if (historyLoadErrorGeneration == generation) { + updateErrorText(null) + } + if (history.inFlightRun == null) { + // Empty history is terminal proof for acknowledged runs. An unknown-outcome + // send stays owned until its reply persists, a terminal arrives, or it expires. + runIdsToReconcile + .filterNot { runId -> + unknownOutcomeRunIds.contains(runId) && unresolvedRepliesByRunId.containsKey(runId) + } + .forEach(::clearPendingRun) + } + clearTransientRunUiIfIdle() // All live history paths (bootstrap, reconnect recovery, cache-first // replace) adopt the gateway's in-flight run snapshot so restored // runs keep their pending state and streaming text. @@ -615,10 +781,35 @@ class ChatController internal constructor( true } if (!applied) return false + completeReconnectRecoveryIfOwned(sessionKey, generation) persistTranscript(requestCacheScope, sessionKey, history.messages) return true } + /** Lets whichever same-generation history request wins finish reconnect health recovery. */ + private suspend fun completeReconnectRecoveryIfOwned( + sessionKey: String, + generation: Long, + ) { + val ownsRecovery = + synchronized(gatewayScopeApplyLock) { + reconnectRecoveryGeneration == generation && + isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get()) + } + if (!ownsRecovery) return + pollHealthIfNeeded(force = true) + synchronized(gatewayScopeApplyLock) { + if ( + reconnectRecoveryGeneration == generation && + isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get()) && + _healthOk.value + ) { + reconnectRecoveryGeneration = null + restoreRunStateOnReconnect = false + } + } + } + /** Emits cached transcript/session rows for instant cold open; live data replaces them wholesale. */ private suspend fun primeFromCache( sessionKey: String, @@ -770,7 +961,7 @@ class ChatController internal constructor( val outbox = commandOutbox ?: return false val outboxScope = currentCacheScope() ?: run { - _errorText.value = "Gateway health not OK; cannot send" + updateErrorText("Gateway health not OK; cannot send") return false } val result = @@ -783,21 +974,21 @@ class ChatController internal constructor( nowMs = System.currentTimeMillis(), ) } catch (_: Throwable) { - _errorText.value = "Could not queue message for later delivery." + updateErrorText("Could not queue message for later delivery.") return false } return when (result) { is ChatOutboxEnqueueResult.Queued -> { - _errorText.value = null + updateErrorText(null) publishOutbox() true } ChatOutboxEnqueueResult.QueueFull -> { - _errorText.value = "Offline queue is full ($OUTBOX_MAX_QUEUED messages); delete queued items first." + updateErrorText("Offline queue is full ($OUTBOX_MAX_QUEUED messages); delete queued items first.") false } ChatOutboxEnqueueResult.Unavailable -> { - _errorText.value = "Gateway health not OK; cannot send" + updateErrorText("Gateway health not OK; cannot send") false } } @@ -1001,6 +1192,7 @@ class ChatController internal constructor( val runId = payload["runId"].asStringOrNull() val isPending = if (runId != null) synchronized(pendingRuns) { pendingRuns.contains(runId) } else true + val isOwned = isPending || (runId != null && unresolvedRepliesByRunId.containsKey(runId)) val state = payload["state"].asStringOrNull() when (state) { @@ -1013,28 +1205,62 @@ class ChatController internal constructor( } } "final", "aborted", "error" -> { + val terminalHasAssistantMessage = + state == "final" && payload["message"].asObjectOrNull()?.get("role").asStringOrNull() == "assistant" + val resolvesWithoutReply = state != "final" || !terminalHasAssistantMessage + val wasTimedOut = runId != null && timedOutRunIds.remove(runId) + if (runId != null && runId == lastHandledTerminalRunId) return + if (runId != null && !isOwned && !wasTimedOut) { + val hasLocalRun = + synchronized(pendingRuns) { pendingRuns.isNotEmpty() } || unresolvedRepliesByRunId.isNotEmpty() + if (!hasLocalRun) { + // Another client or chat.inject can finish the open session. Refresh + // idle history without allowing its terminal state to own local UI. + lastHandledTerminalRunId = runId + refreshCurrentHistoryBestEffort(updateSessionInfo = true) + } + return + } + if (runId != null) lastHandledTerminalRunId = runId + if (wasTimedOut) { + val hasNewerRun = + synchronized(pendingRuns) { pendingRuns.isNotEmpty() } || unresolvedRepliesByRunId.isNotEmpty() + if (!hasNewerRun) { + pendingToolCallsById.clear() + publishPendingToolCalls() + _streamingAssistantText.value = null + updateErrorText(if (state == "error") payload["errorMessage"].asStringOrNull() ?: "Chat failed" else null) + } + refreshCurrentHistoryBestEffort(updateSessionInfo = true) + return + } + if (runId != null && !isPending) { + if (resolvesWithoutReply) terminalWithoutReplyRunIds.add(runId) + refreshCurrentHistoryBestEffort( + runIdsToReconcile = setOf(runId), + updateSessionInfo = true, + ) + return + } if (state == "error") { - _errorText.value = payload["errorMessage"].asStringOrNull() ?: "Chat failed" + updateErrorText(payload["errorMessage"].asStringOrNull() ?: "Chat failed") } if (runId != null) { clearPendingRun(runId) + if (resolvesWithoutReply) { + terminalWithoutReplyRunIds.add(runId) + } } else { clearPendingRuns(clearOptimisticMessages = false) } pendingToolCallsById.clear() publishPendingToolCalls() _streamingAssistantText.value = null - scope.launch { - try { - fetchAndApplyHistory( - sessionKey = _sessionKey.value, - generation = historyLoadGeneration.get(), - updateSessionInfo = true, - ) - } catch (_: Throwable) { - // best-effort - } - } + val terminalRunIds = runId?.let(::setOf) ?: unresolvedRepliesByRunId.keys.toSet() + refreshCurrentHistoryBestEffort( + runIdsToReconcile = terminalRunIds, + updateSessionInfo = true, + ) } } } @@ -1067,6 +1293,14 @@ class ChatController internal constructor( val payload = json.parseToJsonElement(payloadJson).asObjectOrNull() ?: return val sessionKey = payload["sessionKey"].asStringOrNull()?.trim() if (!sessionKey.isNullOrEmpty() && sessionKey != _sessionKey.value) return + val runId = payload["runId"].asStringOrNull() + if ( + runId != null && + synchronized(pendingRuns) { runId !in pendingRuns } && + !unresolvedRepliesByRunId.containsKey(runId) + ) { + return + } val stream = payload["stream"].asStringOrNull() val data = payload["data"].asObjectOrNull() @@ -1102,7 +1336,7 @@ class ChatController internal constructor( } } "error" -> { - _errorText.value = "Event stream interrupted; try refreshing." + updateErrorText("Event stream interrupted; try refreshing.") clearPendingRuns() pendingToolCallsById.clear() publishPendingToolCalls() @@ -1144,6 +1378,7 @@ class ChatController internal constructor( synchronized(pendingRuns) { // A different locally-owned run means this snapshot predates it; ignore. if (pendingRuns.isNotEmpty() && runId !in pendingRuns) return + if (pendingRuns.isEmpty() && unresolvedRepliesByRunId.isNotEmpty() && !unresolvedRepliesByRunId.containsKey(runId)) return pendingRuns.add(runId) _pendingRunCount.value = pendingRuns.size } @@ -1158,34 +1393,72 @@ class ChatController internal constructor( pendingRunTimeoutJobs[runId] = scope.launch { delay(pendingRunTimeoutMs) + refreshHistorySnapshotBestEffort( + sessionKey = _sessionKey.value, + generation = historyLoadGeneration.get(), + runIdsToReconcile = emptySet(), + ) + val runStillInFlight = synchronized(gatewayScopeApplyLock) { latestAppliedInFlightRunId == runId } + val replyStillUnresolved = unresolvedRepliesByRunId.containsKey(runId) + if (!runStillInFlight) { + clearPendingRun(runId) + clearTransientRunUiIfIdle() + if (!replyStillUnresolved) return@launch + } val stillPending = synchronized(pendingRuns) { pendingRuns.contains(runId) } - if (!stillPending) return@launch + if (!stillPending && !replyStillUnresolved) return@launch clearPendingRun(runId) + clearTransientRunUiIfIdle() removeOptimisticMessage(runId) - _errorText.value = "Timed out waiting for a reply; try again or refresh." + unresolvedRepliesByRunId.remove(runId) + terminalWithoutReplyRunIds.remove(runId) + timedOutRunIds.add(runId) + updateErrorText("Timed out waiting for a reply; try again or refresh.") } } private fun clearPendingRun(runId: String) { pendingRunTimeoutJobs.remove(runId)?.cancel() + unknownOutcomeRunIds.remove(runId) synchronized(pendingRuns) { + disconnectedPendingRunIds.remove(runId) pendingRuns.remove(runId) _pendingRunCount.value = pendingRuns.size } } - private fun clearPendingRuns(clearOptimisticMessages: Boolean = true) { + private fun clearTransientRunUiIfIdle() { + if (synchronized(pendingRuns) { pendingRuns.isNotEmpty() }) return + pendingToolCallsById.clear() + publishPendingToolCalls() + _streamingAssistantText.value = null + } + + private fun clearPendingRuns( + clearOptimisticMessages: Boolean = true, + preserveDisconnectedOwnership: Boolean = false, + ) { for ((_, job) in pendingRunTimeoutJobs) { job.cancel() } pendingRunTimeoutJobs.clear() if (clearOptimisticMessages) { + recoveryHistoryReconciliationJob?.cancel() + recoveryHistoryReconciliationGeneration = -1L + recoveryHistoryReconciliationJob = null optimisticMessagesByRunId.clear() + unresolvedRepliesByRunId.clear() + timedOutRunIds.clear() + terminalWithoutReplyRunIds.clear() + unknownOutcomeRunIds.clear() } synchronized(pendingRuns) { + if (!preserveDisconnectedOwnership) { + disconnectedPendingRunIds.clear() + } pendingRuns.clear() _pendingRunCount.value = 0 } @@ -1196,6 +1469,61 @@ class ChatController internal constructor( _messages.value = _messages.value.filterNot { it.id == message.id } } + private fun transferRunOwnership( + oldRunId: String, + newRunId: String, + fallbackMessage: ChatMessage, + messageIdempotencyKey: String? = fallbackMessage.idempotencyKey, + ) { + if (oldRunId == newRunId) return + val optimistic = optimisticMessagesByRunId.remove(oldRunId) + val unresolved = unresolvedRepliesByRunId.remove(oldRunId) + val terminalWithoutReply = terminalWithoutReplyRunIds.remove(oldRunId) + unknownOutcomeRunIds.remove(oldRunId) + val original = optimistic ?: unresolved ?: fallbackMessage + // Run ownership can change independently of the client key persisted on the + // user row. Only history proof may replace that transcript identity. + val rekeyed = original.copy(idempotencyKey = messageIdempotencyKey) + if (optimistic != null) optimisticMessagesByRunId[newRunId] = rekeyed + if (unresolved != null) unresolvedRepliesByRunId[newRunId] = rekeyed + if (terminalWithoutReply) terminalWithoutReplyRunIds.add(newRunId) + _messages.value = _messages.value.map { if (it.id == original.id) rekeyed else it } + clearPendingRun(oldRunId) + synchronized(pendingRuns) { + pendingRuns.add(newRunId) + _pendingRunCount.value = pendingRuns.size + } + armPendingRunTimeout(newRunId) + } + + private fun transferLostAckOwnershipFromHistory(history: ChatHistory) { + val snapshotRunId = history.inFlightRun?.runId?.trim()?.takeIf { it.isNotEmpty() } ?: return + if (unresolvedRepliesByRunId.containsKey(snapshotRunId)) return + val localRunId = + synchronized(pendingRuns) { + (pendingRuns + disconnectedPendingRunIds).singleOrNull() + } ?: return + if (!unknownOutcomeRunIds.contains(localRunId)) return + val optimistic = unresolvedRepliesByRunId[localRunId] ?: return + val canonicalUserKey = "$snapshotRunId:user" + val optimisticUserKey = optimistic.idempotencyKey?.trim() + val optimisticContentKey = messageContentIdentityKey(optimistic) + val persistedUser = + history.messages.firstOrNull { message -> + val persistedUserKey = message.idempotencyKey?.trim() + (persistedUserKey == optimisticUserKey || persistedUserKey == canonicalUserKey) && + messageContentIdentityKey(message) == optimisticContentKey + } + if (persistedUser != null) { + transferRunOwnership( + oldRunId = localRunId, + newRunId = snapshotRunId, + fallbackMessage = optimistic, + messageIdempotencyKey = persistedUser.idempotencyKey, + ) + } + } + private fun prunePersistedOptimisticMessages(incoming: List) { val retained = retainUnmatchedOptimisticMessages( @@ -1205,18 +1533,99 @@ class ChatController internal constructor( optimisticMessagesByRunId.entries.removeAll { entry -> entry.value !in retained } } - private fun refreshCurrentHistoryBestEffort() { + private fun resolvePersistedReplies(incoming: List) { + val resolvedRunIds = + unresolvedRepliesByRunId + .filter { (runId, optimistic) -> + val userIndex = incoming.indexOfFirst { message -> incomingMessageConsumesOptimistic(message, optimistic) } + if (userIndex < 0) return@filter false + terminalWithoutReplyRunIds.contains(runId) || + incoming + .drop(userIndex + 1) + .takeWhile { it.role.trim().lowercase() != "user" } + .any { it.role.trim().lowercase() == "assistant" } + }.keys + .toList() + resolvedRunIds.forEach(unresolvedRepliesByRunId::remove) + resolvedRunIds.forEach(terminalWithoutReplyRunIds::remove) + } + + private fun scheduleRecoveryHistoryReconciliation( + sessionKey: String, + generation: Long, + runIds: Set, + ) { + val reconciliationRunIds = runIds + unresolvedRepliesByRunId.keys + if (reconciliationRunIds.isEmpty()) return + val hasPendingRun = synchronized(pendingRuns) { reconciliationRunIds.any { it in pendingRuns } } + if (!hasPendingRun && reconciliationRunIds.none(unresolvedRepliesByRunId::containsKey)) return + if (generation < recoveryHistoryReconciliationGeneration) return + recoveryHistoryReconciliationJob?.cancel() + recoveryHistoryReconciliationGeneration = generation + recoveryHistoryReconciliationJob = + scope.launch { + delay(recoveryHistoryRetryDelayMs) + if (!isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get())) return@launch + if (!_healthOk.value) return@launch + refreshHistorySnapshotBestEffort(sessionKey, generation, reconciliationRunIds) + if (synchronized(pendingRuns) { reconciliationRunIds.any { it in pendingRuns } }) return@launch + if (reconciliationRunIds.none(unresolvedRepliesByRunId::containsKey)) return@launch + + // A persisted user row is not terminal proof: the assistant row can lag + // behind it even after the run disappears from the history snapshot. + delay(pendingRunTimeoutMs - recoveryHistoryRetryDelayMs) + if (!isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get())) return@launch + if (!_healthOk.value) return@launch + refreshHistorySnapshotBestEffort(sessionKey, generation, reconciliationRunIds) + if (synchronized(pendingRuns) { reconciliationRunIds.any { it in pendingRuns } }) return@launch + val unresolvedRunIds = reconciliationRunIds.filter(unresolvedRepliesByRunId::containsKey) + if (unresolvedRunIds.isEmpty()) return@launch + unresolvedRunIds.forEach(::removeOptimisticMessage) + unresolvedRunIds.forEach(unresolvedRepliesByRunId::remove) + unresolvedRunIds.forEach(terminalWithoutReplyRunIds::remove) + updateErrorText("Timed out confirming the sent message; refresh to check delivery.") + } + } + + private suspend fun refreshHistorySnapshotBestEffort( + sessionKey: String, + generation: Long, + runIdsToReconcile: Set, + ) { + try { + fetchAndApplyHistory( + sessionKey, + generation, + updateSessionInfo = true, + runIdsToReconcile = runIdsToReconcile, + ) + } catch (err: CancellationException) { + throw err + } catch (_: Throwable) { + // The bounded expiry below remains the final reconciliation path. + } + } + + private fun refreshCurrentHistoryBestEffort( + runIdsToReconcile: Set = emptySet(), + updateSessionInfo: Boolean = false, + ) { + val sessionKey = _sessionKey.value + val generation = historyLoadGeneration.get() scope.launch { try { fetchAndApplyHistory( - sessionKey = _sessionKey.value, - generation = historyLoadGeneration.get(), - // Intentionally skips session-info upserts: post-send refreshes should not reorder the - // session list; sessions.changed events own that. - updateSessionInfo = false, + sessionKey = sessionKey, + generation = generation, + updateSessionInfo = updateSessionInfo, + runIdsToReconcile = runIdsToReconcile, ) } catch (_: Throwable) { // best-effort + } finally { + if (isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get())) { + scheduleRecoveryHistoryReconciliation(sessionKey, generation, runIdsToReconcile) + } } } } diff --git a/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt b/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt index 9c7f6385b8e6..d83803ad1d58 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt @@ -129,7 +129,20 @@ private class GatewayConnectFailure( val gatewayError: GatewaySession.ErrorShape, ) : IllegalStateException(gatewayError.message) -private class GatewayRequestNotEnqueued( +internal sealed class GatewayRequestDefinitiveFailure( + message: String, +) : IllegalStateException(message) + +internal class GatewayRequestNotEnqueued( + message: String, +) : GatewayRequestDefinitiveFailure(message) + +internal class GatewayRequestRejected( + val gatewayError: GatewaySession.ErrorShape, +) : GatewayRequestDefinitiveFailure("${gatewayError.code}: ${gatewayError.message}") + +/** Request frame was sent, but no response proved whether the gateway applied it. */ +internal class GatewayRequestOutcomeUnknown( message: String, ) : IllegalStateException(message) @@ -380,8 +393,7 @@ class GatewaySession( ): String { val res = requestDetailed(method = method, paramsJson = paramsJson, timeoutMs = timeoutMs) if (res.ok) return res.payloadJson ?: "" - val err = res.error - throw IllegalStateException("${err?.code ?: "UNAVAILABLE"}: ${err?.message ?: "request failed"}") + throw GatewayRequestRejected(res.error ?: ErrorShape("UNAVAILABLE", "request failed")) } /** Sends an RPC request and returns the structured success/error payload. */ @@ -390,7 +402,7 @@ class GatewaySession( paramsJson: String?, timeoutMs: Long = 15_000, ): RpcResult { - val conn = readyConnection() ?: throw IllegalStateException("not connected") + val conn = readyConnection() ?: throw GatewayRequestNotEnqueued("not connected") val params = if (paramsJson.isNullOrBlank()) { null @@ -506,7 +518,7 @@ class GatewaySession( sendJson(buildRequestFrame(id = id, method = method, params = params)) return withTimeout(timeoutMs) { deferred.await() } } catch (err: TimeoutCancellationException) { - throw IllegalStateException("request timeout") + throw GatewayRequestOutcomeUnknown("request timeout") } finally { pending.remove(id) if (connectRequestId == id) connectRequestId = null @@ -537,6 +549,9 @@ class GatewaySession( return@launch } catch (_: CancellationException) { return@launch + } catch (err: GatewayRequestOutcomeUnknown) { + onError(ErrorShape("UNAVAILABLE", err.message ?: "request outcome unknown")) + return@launch } if (!response.ok) { onError(response.error ?: ErrorShape("UNAVAILABLE", "request failed")) @@ -1161,7 +1176,7 @@ class GatewaySession( pending.values.toList().also { pending.clear() } } for (waiter in waiters) { - waiter.cancel() + waiter.completeExceptionally(GatewayRequestOutcomeUnknown("Gateway disconnected before response")) } } } diff --git a/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt b/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt index 2d4e78a9588b..a01706bf6fab 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt @@ -110,6 +110,7 @@ fun ChatScreen( val chatCommands by viewModel.chatCommands.collectAsState() val chatDraft by viewModel.chatDraft.collectAsState() val pendingAssistantAutoSend by viewModel.pendingAssistantAutoSend.collectAsState() + val assistantAutoSendInFlight by viewModel.assistantAutoSendInFlight.collectAsState() val remoteAddress by viewModel.remoteAddress.collectAsState() val outboxItems by viewModel.chatOutboxItems.collectAsState() val manualHost by viewModel.manualHost.collectAsState() @@ -151,18 +152,18 @@ fun ChatScreen( viewModel.refreshChatCommands() } - LaunchedEffect(pendingAssistantAutoSend, healthOk, pendingRunCount, thinkingLevel) { - val accepted = - dispatchPendingAssistantAutoSend( + LaunchedEffect(pendingAssistantAutoSend, assistantAutoSendInFlight, healthOk, pendingRunCount, thinkingLevel) { + if (!healthOk) return@LaunchedEffect + val prompt = + resolvePendingAssistantAutoSend( pendingPrompt = pendingAssistantAutoSend, healthOk = healthOk, pendingRunCount = pendingRunCount, - ) { prompt -> - viewModel.sendChatAwaitAcceptance(message = prompt, thinking = thinkingLevel, attachments = emptyList()) - } - if (accepted) { - viewModel.clearPendingAssistantAutoSend() - } + ) ?: return@LaunchedEffect + viewModel.dispatchPendingAssistantAutoSend( + pendingPrompt = prompt, + thinking = thinkingLevel, + ) } var input by rememberSaveable { mutableStateOf("") } diff --git a/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatSheetContent.kt b/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatSheetContent.kt index 45005faecd7d..2296d4abb7b1 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatSheetContent.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatSheetContent.kt @@ -57,22 +57,6 @@ internal fun resolvePendingAssistantAutoSend( return prompt } -/** Dispatches a pending assistant prompt once and reports whether it was accepted. */ -internal suspend fun dispatchPendingAssistantAutoSend( - pendingPrompt: String?, - healthOk: Boolean, - pendingRunCount: Int, - dispatch: suspend (String) -> Boolean, -): Boolean { - val prompt = - resolvePendingAssistantAutoSend( - pendingPrompt = pendingPrompt, - healthOk = healthOk, - pendingRunCount = pendingRunCount, - ) ?: return false - return dispatch(prompt) -} - /** Chooses the session key to load for initial chat hydration, if any. */ internal fun resolveInitialChatLoadSessionKey( sessionKey: String, @@ -101,6 +85,7 @@ fun ChatSheetContent(viewModel: MainViewModel) { val chatCommands by viewModel.chatCommands.collectAsState() val chatDraft by viewModel.chatDraft.collectAsState() val pendingAssistantAutoSend by viewModel.pendingAssistantAutoSend.collectAsState() + val assistantAutoSendInFlight by viewModel.assistantAutoSendInFlight.collectAsState() val outboxItems by viewModel.chatOutboxItems.collectAsState() LaunchedEffect(Unit) { @@ -111,23 +96,20 @@ fun ChatSheetContent(viewModel: MainViewModel) { viewModel.refreshChatCommands() } - LaunchedEffect(pendingAssistantAutoSend, healthOk, pendingRunCount, thinkingLevel) { + LaunchedEffect(pendingAssistantAutoSend, assistantAutoSendInFlight, healthOk, pendingRunCount, thinkingLevel) { // Assistant-launch prompts should wait for a healthy idle chat so they do // not race an already-running turn. - val accepted = - dispatchPendingAssistantAutoSend( + if (!healthOk) return@LaunchedEffect + val prompt = + resolvePendingAssistantAutoSend( pendingPrompt = pendingAssistantAutoSend, healthOk = healthOk, pendingRunCount = pendingRunCount, - ) { prompt -> - viewModel.sendChatAwaitAcceptance( - message = prompt, - thinking = thinkingLevel, - attachments = emptyList(), - ) - } - if (!accepted) return@LaunchedEffect - viewModel.clearPendingAssistantAutoSend() + ) ?: return@LaunchedEffect + viewModel.dispatchPendingAssistantAutoSend( + pendingPrompt = prompt, + thinking = thinkingLevel, + ) } val context = LocalContext.current diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt index ecb2c0ce5b5d..b0d2ed3e5516 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt @@ -1,11 +1,18 @@ package ai.openclaw.app.chat +import ai.openclaw.app.gateway.GatewayRequestOutcomeUnknown +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.Json +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse import org.junit.Assert.assertNull import org.junit.Assert.assertTrue import org.junit.Test @@ -18,8 +25,7 @@ import org.junit.Test class ChatControllerReconnectRestoreTest { private val json = Json { ignoreUnknownKeys = true } - private fun TestScope.newController(gateway: ScriptedGateway): ChatController = - ChatController(scope = this, json = json, requestGateway = gateway::request) + private fun TestScope.newController(gateway: ScriptedGateway): ChatController = ChatController(scope = this, json = json, requestGateway = gateway::request) private val userTurn = ReplayHistoryMessage("user", "keep working", 1_000) @@ -39,7 +45,7 @@ class ChatControllerReconnectRestoreTest { "chat.history", historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "partial reply"), ) - controller.handleGatewayEvent("health", null) + controller.onGatewayConnected() runCurrent() assertEquals(1, controller.pendingRunCount.value) @@ -59,7 +65,10 @@ class ChatControllerReconnectRestoreTest { listOf(userTurn, ReplayHistoryMessage("assistant", "partial reply more", 2_000)), ), ) - controller.handleGatewayEvent("chat", chatTerminalPayload("main", "run-active", seq = 6)) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", "run-active", seq = 6, assistantText = "partial reply more"), + ) runCurrent() assertEquals(0, controller.pendingRunCount.value) @@ -77,13 +86,15 @@ class ChatControllerReconnectRestoreTest { controller.load("main") runCurrent() val historyCallsAfterLoad = gateway.callCount("chat.history") + val commandCallsAfterLoad = gateway.callCount("commands.list") controller.onDisconnected("Offline") - controller.handleGatewayEvent("health", null) + controller.onGatewayConnected() runCurrent() // Reconnect refetched history once and restored nothing. assertEquals(historyCallsAfterLoad + 1, gateway.callCount("chat.history")) + assertEquals(commandCallsAfterLoad + 1, gateway.callCount("commands.list")) assertEquals(0, controller.pendingRunCount.value) assertNull(controller.streamingAssistantText.value) assertNull(controller.errorText.value) @@ -91,6 +102,313 @@ class ChatControllerReconnectRestoreTest { assertEquals(1, controller.messages.value.size) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun reconnectStaysUnhealthyUntilRecoveryHistoryApplies() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + val recoveryHistory = CompletableDeferred() + gateway.respond("chat.history") { recoveryHistory.await() } + controller.onDisconnected("Reconnecting…") + controller.onGatewayConnected() + runCurrent() + + assertFalse(controller.healthOk.value) + val healthCallsDuringRecovery = gateway.callCount("health") + val historyCallsDuringRecovery = gateway.callCount("chat.history") + controller.handleGatewayEvent("tick", null) + runCurrent() + assertFalse(controller.healthOk.value) + assertEquals(healthCallsDuringRecovery, gateway.callCount("health")) + assertEquals(historyCallsDuringRecovery + 1, gateway.callCount("chat.history")) + + recoveryHistory.complete(historyResponse("session-1", emptyList())) + runCurrent() + assertTrue(controller.healthOk.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun newerSameGenerationHistoryRequestCompletesReconnectHealth() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith( + "chat.history", + historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "working"), + ) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + val reconnectHistoryStarted = CompletableDeferred() + val releaseReconnectHistory = CompletableDeferred() + var recoveryHistoryCalls = 0 + gateway.respond("chat.history") { + recoveryHistoryCalls += 1 + if (recoveryHistoryCalls == 1) { + reconnectHistoryStarted.complete(Unit) + releaseReconnectHistory.await() + } else { + historyResponse( + "session-1", + listOf(userTurn, ReplayHistoryMessage("assistant", "done", 2_000)), + ) + } + } + + controller.onDisconnected("Reconnecting…") + controller.onGatewayConnected() + runCurrent() + reconnectHistoryStarted.await() + assertFalse(controller.healthOk.value) + + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", "run-active", seq = 2, assistantText = "done"), + ) + runCurrent() + + assertTrue(controller.healthOk.value) + assertEquals(listOf("keep working", "done"), controller.messages.value.map { it.content.single().text }) + + releaseReconnectHistory.complete(historyResponse("session-1", emptyList())) + runCurrent() + assertTrue(controller.healthOk.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun recoveredPendingRunRefreshesHistoryBeforeTimingOut() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith( + "chat.history", + historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "working"), + ) + val controller = newController(gateway) + controller.load("main") + runCurrent() + assertEquals(1, controller.pendingRunCount.value) + controller.handleGatewayEvent( + "agent", + """{"sessionKey":"main","runId":"run-active","seq":2,"ts":10,"stream":"tool","data":{"phase":"start","name":"exec","toolCallId":"tool-1"}}""", + ) + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf(userTurn, ReplayHistoryMessage("assistant", "completed while offline", 2_000)), + ), + ) + advanceTimeBy(120_000) + runCurrent() + + assertEquals(0, controller.pendingRunCount.value) + assertEquals( + listOf("keep working", "completed while offline"), + controller.messages.value.map { it.content.single().text }, + ) + assertNull(controller.errorText.value) + assertNull(controller.streamingAssistantText.value) + assertTrue(controller.pendingToolCalls.value.isEmpty()) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun explicitRefreshClearsPriorHistoryError() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + gateway.respond("chat.history") { error("history unavailable") } + controller.refresh() + runCurrent() + assertEquals("history unavailable", controller.errorText.value) + + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + controller.refresh() + assertNull(controller.errorText.value) + runCurrent() + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun disconnectInvalidatesLateHistoryError() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + val pendingHistory = CompletableDeferred() + gateway.respond("chat.history") { pendingHistory.await() } + controller.refresh() + runCurrent() + controller.onDisconnected("Reconnecting…") + pendingHistory.completeExceptionally(IllegalStateException("socket closed")) + runCurrent() + assertNull(controller.errorText.value) + + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + controller.onGatewayConnected() + assertNull(controller.errorText.value) + runCurrent() + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun disconnectInvalidatesOlderHistorySnapshotBeforeOwnershipRestore() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + assertTrue(controller.sendMessageAwaitAcceptance("keep ownership", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + + val staleHistory = CompletableDeferred() + gateway.respond("chat.history") { staleHistory.await() } + controller.refresh() + runCurrent() + controller.onDisconnected("Reconnecting…") + staleHistory.complete(historyResponse("session-1", emptyList())) + runCurrent() + + gateway.respondWith( + "chat.history", + historyResponse("session-1", emptyList(), inFlightRun = runId to "working"), + ) + controller.onGatewayConnected() + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("working", controller.streamingAssistantText.value) + assertEquals(listOf("keep ownership"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun disconnectAfterGatewayAcceptancePreservesSendWhenAckIsLost() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + val sendStarted = CompletableDeferred() + val releaseSend = CompletableDeferred() + gateway.respond("chat.send") { + sendStarted.complete(Unit) + releaseSend.await() + } + val sendResult = async { controller.sendMessageAwaitAcceptance("accepted before drop", "off", emptyList()) } + sendStarted.await() + val runId = + json + .parseToJsonElement(requireNotNull(gateway.calls.last { it.method == "chat.send" }.paramsJson)) + .jsonObject + .getValue("idempotencyKey") + .jsonPrimitive + .content + + controller.onDisconnected("Reconnecting…") + releaseSend.completeExceptionally(GatewayRequestOutcomeUnknown("socket closed before ACK")) + assertTrue(sendResult.await()) + assertEquals(listOf("accepted before drop"), controller.messages.value.map { it.content.single().text }) + assertNull(controller.errorText.value) + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "accepted before drop", 1_000, idempotencyKey = "$runId:user"), + ReplayHistoryMessage("assistant", "completed once", 2_000), + ), + ), + ) + controller.onGatewayConnected() + runCurrent() + + assertEquals(0, controller.pendingRunCount.value) + assertEquals( + listOf("accepted before drop", "completed once"), + controller.messages.value.map { it.content.single().text }, + ) + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun lostAckAdoptsCanonicalRunWhilePreservingClientHistoryIdentity() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + gateway.respond("chat.send") { throw GatewayRequestOutcomeUnknown("ACK lost") } + var clientRunId: String? = null + var recoveryHistoryCalls = 0 + gateway.respond("chat.history") { + recoveryHistoryCalls += 1 + clientRunId = + json + .parseToJsonElement(requireNotNull(gateway.calls.last { it.method == "chat.send" }.paramsJson)) + .jsonObject + .getValue("idempotencyKey") + .jsonPrimitive + .content + if (recoveryHistoryCalls == 1) { + historyResponse("session-1", emptyList()) + } else { + historyResponse( + "session-1", + listOf(ReplayHistoryMessage("user", "canonical recovery", 1_000, idempotencyKey = "$clientRunId:user")), + inFlightRun = "canonical-run" to "working", + ) + } + } + assertTrue(controller.sendMessageAwaitAcceptance("canonical recovery", "off", emptyList())) + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertNull(controller.streamingAssistantText.value) + + advanceTimeBy(750) + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("working", controller.streamingAssistantText.value) + assertEquals( + "$clientRunId:user", + controller.messages.value + .single { it.role == "user" } + .idempotencyKey, + ) + controller.handleGatewayEvent( + "chat", + chatDeltaPayload("main", "canonical-run", 1, " now", "working now"), + ) + assertEquals("working now", controller.streamingAssistantText.value) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun repeatedReconnectsDoNotDuplicateRunOrRows() = @@ -108,7 +426,7 @@ class ChatControllerReconnectRestoreTest { repeat(2) { controller.onDisconnected("Reconnecting…") assertEquals(0, controller.pendingRunCount.value) - controller.handleGatewayEvent("health", null) + controller.onGatewayConnected() runCurrent() } @@ -119,7 +437,7 @@ class ChatControllerReconnectRestoreTest { @Test @OptIn(ExperimentalCoroutinesApi::class) - fun staleSnapshotRunDoesNotReplaceLocallyOwnedSend() = + fun reconnectKeepsOptimisticUserWhileHistoryPersistenceLags() = runTest { val gateway = ScriptedGateway(json) gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) @@ -128,23 +446,732 @@ class ChatControllerReconnectRestoreTest { controller.load("main") runCurrent() - // Reconnect arms a recovery refresh, then a send lands before it executes. + assertTrue(controller.sendMessageAwaitAcceptance("survive reconnect", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) controller.onDisconnected("Reconnecting…") gateway.respondWith( "chat.history", - historyResponse("session-1", emptyList(), inFlightRun = "run-stale" to "old text"), + historyResponse("session-1", emptyList(), inFlightRun = runId to "working"), ) - controller.handleGatewayEvent("health", null) - assertTrue(controller.sendMessageAwaitAcceptance("new work", "off", emptyList())) - val localRunId = requireNotNull(gateway.lastRunId) + controller.onGatewayConnected() runCurrent() assertEquals(1, controller.pendingRunCount.value) + assertEquals("working", controller.streamingAssistantText.value) + assertEquals(listOf("survive reconnect"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun reconnectStaleSnapshotCannotReplaceDisconnectedLocalRun() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("local work", "off", emptyList())) + val localRunId = requireNotNull(gateway.lastRunId) + controller.onDisconnected("Reconnecting…") + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf(ReplayHistoryMessage("user", "local work", 1_000, idempotencyKey = "$localRunId:user")), + inFlightRun = "run-stale" to "old text", + ), + ) + controller.onGatewayConnected() + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertNull(controller.streamingAssistantText.value) controller.handleGatewayEvent( "chat", chatDeltaPayload("main", localRunId, 1, "ours", "ours"), ) assertEquals("ours", controller.streamingAssistantText.value) + controller.handleGatewayEvent( + "agent", + """{"sessionKey":"main","runId":"run-stale","seq":2,"stream":"assistant","data":{"text":"stale agent"}}""", + ) + controller.handleGatewayEvent( + "agent", + """{"sessionKey":"main","runId":"run-stale","seq":3,"ts":10,"stream":"tool","data":{"phase":"start","name":"exec","toolCallId":"stale-tool"}}""", + ) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", "run-stale", seq = 4, state = "error"), + ) + assertEquals(1, controller.pendingRunCount.value) + assertEquals("ours", controller.streamingAssistantText.value) + assertTrue(controller.pendingToolCalls.value.isEmpty()) + assertNull(controller.errorText.value) + assertEquals(listOf("local work"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun reconnectRetiresPersistedLocalRunBeforeAdoptingOtherRun() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("local work", "off", emptyList())) + val localRunId = requireNotNull(gateway.lastRunId) + controller.onDisconnected("Reconnecting…") + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "local work", 1_000, idempotencyKey = "$localRunId:user"), + ReplayHistoryMessage("assistant", "local done", 2_000), + ), + inFlightRun = "run-other" to "other working", + ), + ) + controller.onGatewayConnected() + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("other working", controller.streamingAssistantText.value) + assertEquals(listOf("local work", "local done"), controller.messages.value.map { it.content.single().text }) + controller.handleGatewayEvent( + "chat", + chatDeltaPayload("main", localRunId, 1, "stale", "stale local"), + ) + assertEquals("other working", controller.streamingAssistantText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun reconnectReplacesPreviouslyAdoptedRunWithAuthoritativeSnapshotRun() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith( + "chat.history", + historyResponse("session-1", emptyList(), inFlightRun = "run-a" to "old work"), + ) + val controller = newController(gateway) + controller.load("main") + runCurrent() + assertEquals(1, controller.pendingRunCount.value) + assertEquals("old work", controller.streamingAssistantText.value) + + controller.onDisconnected("Reconnecting…") + gateway.respondWith( + "chat.history", + historyResponse("session-1", emptyList(), inFlightRun = "run-b" to "current work"), + ) + controller.onGatewayConnected() + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("current work", controller.streamingAssistantText.value) + controller.handleGatewayEvent("chat", chatDeltaPayload("main", "run-a", 1, " stale", "old work stale")) + assertEquals("current work", controller.streamingAssistantText.value) + controller.handleGatewayEvent("chat", chatDeltaPayload("main", "run-b", 1, " now", "current work now")) + assertEquals("current work now", controller.streamingAssistantText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun seqGapKeepsOptimisticUserWhileHistoryPersistenceLags() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("survive gap", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + gateway.respondWith( + "chat.history", + historyResponse("session-1", emptyList(), inFlightRun = runId to "working"), + ) + controller.handleGatewayEvent("seqGap", null) + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("working", controller.streamingAssistantText.value) + assertEquals(listOf("survive gap"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun sameSessionRefreshKeepsOptimisticRunOwnership() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("survive refresh", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + gateway.respondWith( + "chat.history", + historyResponse("session-1", emptyList(), inFlightRun = runId to "working"), + ) + controller.refresh() + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("working", controller.streamingAssistantText.value) + assertEquals(listOf("survive refresh"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun sameSessionRefreshClearsTransientUiForResolvedRun() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith( + "chat.history", + historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "partial"), + ) + val controller = newController(gateway) + controller.load("main") + runCurrent() + controller.handleGatewayEvent( + "agent", + """{"sessionKey":"main","runId":"run-active","seq":2,"ts":10,"stream":"tool","data":{"phase":"start","name":"exec","toolCallId":"tool-1"}}""", + ) + assertEquals("partial", controller.streamingAssistantText.value) + assertEquals(1, controller.pendingToolCalls.value.size) + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf(userTurn, ReplayHistoryMessage("assistant", "complete", 2_000)), + ), + ) + controller.refresh() + runCurrent() + + assertEquals(0, controller.pendingRunCount.value) + assertNull(controller.streamingAssistantText.value) + assertTrue(controller.pendingToolCalls.value.isEmpty()) + assertEquals(listOf("keep working", "complete"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun seqGapMissingRunClearsPendingButKeepsOptimisticUser() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("finished during gap", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + controller.handleGatewayEvent("seqGap", null) + runCurrent() + + assertEquals(0, controller.pendingRunCount.value) + assertNull(controller.streamingAssistantText.value) + assertEquals(listOf("finished during gap"), controller.messages.value.map { it.content.single().text }) + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "finished during gap", 1_000, idempotencyKey = "$runId:user"), + ReplayHistoryMessage("assistant", "done", 2_000), + ), + ), + ) + advanceTimeBy(750) + runCurrent() + + assertEquals(listOf("finished during gap", "done"), controller.messages.value.map { it.content.single().text }) + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun recoveryRetriesWhenUserPersistsBeforeAssistantReply() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("await reply", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + val persistedUser = ReplayHistoryMessage("user", "await reply", 1_000, idempotencyKey = "$runId:user") + gateway.respondWith("chat.history", historyResponse("session-1", listOf(persistedUser))) + controller.handleGatewayEvent("seqGap", null) + runCurrent() + assertEquals(listOf("await reply"), controller.messages.value.map { it.content.single().text }) + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf(persistedUser, ReplayHistoryMessage("assistant", "done", 2_000)), + ), + ) + advanceTimeBy(750) + runCurrent() + + assertEquals(listOf("await reply", "done"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun recoveryPerformsFinalRefreshWhenAssistantPersistsAfterFirstRetry() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("late reply", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + val persistedUser = ReplayHistoryMessage("user", "late reply", 1_000, idempotencyKey = "$runId:user") + var historyCall = 0 + gateway.respond("chat.history") { + historyCall += 1 + historyResponse( + "session-1", + if (historyCall < 3) { + listOf(persistedUser) + } else { + listOf(persistedUser, ReplayHistoryMessage("assistant", "eventually done", 2_000)) + }, + ) + } + + controller.handleGatewayEvent("seqGap", null) + runCurrent() + advanceTimeBy(750) + runCurrent() + assertEquals(listOf("late reply"), controller.messages.value.map { it.content.single().text }) + + advanceTimeBy(119_250) + runCurrent() + assertEquals(listOf("late reply", "eventually done"), controller.messages.value.map { it.content.single().text }) + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun newerRunReconciliationKeepsOlderUnresolvedReply() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("first", "off", emptyList())) + val firstRunId = requireNotNull(gateway.lastRunId) + val firstUser = ReplayHistoryMessage("user", "first", 1_000, idempotencyKey = "$firstRunId:user") + gateway.respondWith("chat.history", historyResponse("session-1", listOf(firstUser))) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", firstRunId, seq = 2, assistantText = "first done"), + ) + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("second", "off", emptyList())) + val secondRunId = requireNotNull(gateway.lastRunId) + controller.handleGatewayEvent( + "chat", + chatDeltaPayload("main", secondRunId, 1, "new", "second working"), + ) + val secondUser = ReplayHistoryMessage("user", "second", 2_000, idempotencyKey = "$secondRunId:user") + val secondReply = ReplayHistoryMessage("assistant", "second done", 3_000) + gateway.respondWith( + "chat.history", + historyResponse("session-1", listOf(firstUser, secondUser, secondReply)), + ) + controller.handleGatewayEvent("chat", chatTerminalPayload("main", firstRunId, seq = 3, state = "error")) + runCurrent() + assertEquals("second working", controller.streamingAssistantText.value) + assertNull(controller.errorText.value) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", secondRunId, seq = 2, assistantText = "second done"), + ) + runCurrent() + assertEquals(listOf("first", "second", "second done"), controller.messages.value.map { it.content.single().text }) + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf( + firstUser, + ReplayHistoryMessage("assistant", "first done", 1_500), + secondUser, + secondReply, + ), + ), + ) + advanceTimeBy(750) + runCurrent() + + assertEquals( + listOf("first", "first done", "second", "second done"), + controller.messages.value.map { it.content.single().text }, + ) + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun newerRefreshCarriesUnresolvedReplyReconciliation() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("carry reply", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + val persistedUser = ReplayHistoryMessage("user", "carry reply", 1_000, idempotencyKey = "$runId:user") + var historyCall = 0 + gateway.respond("chat.history") { + historyCall += 1 + historyResponse( + "session-1", + if (historyCall < 4) { + listOf(persistedUser) + } else { + listOf(persistedUser, ReplayHistoryMessage("assistant", "carried done", 2_000)) + }, + ) + } + + controller.handleGatewayEvent("seqGap", null) + runCurrent() + advanceTimeBy(750) + runCurrent() + controller.refresh() + runCurrent() + advanceTimeBy(750) + runCurrent() + + assertEquals(listOf("carry reply", "carried done"), controller.messages.value.map { it.content.single().text }) + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun successfulRecoveryRetryClearsHistoryError() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("recover error", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + var historyCall = 0 + gateway.respond("chat.history") { + historyCall += 1 + if (historyCall == 1) { + error("history unavailable") + } + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "recover error", 1_000, idempotencyKey = "$runId:user"), + ReplayHistoryMessage("assistant", "recovered", 2_000), + ), + ) + } + + controller.handleGatewayEvent("seqGap", null) + runCurrent() + assertEquals("history unavailable", controller.errorText.value) + advanceTimeBy(750) + runCurrent() + + assertNull(controller.errorText.value) + assertEquals(listOf("recover error", "recovered"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun reconnectFailureStillExpiresUnconfirmedUser() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("never persisted", "off", emptyList())) + controller.onDisconnected("Reconnecting…") + gateway.respond("chat.history") { error("history unavailable") } + controller.onGatewayConnected() + runCurrent() + + assertEquals(listOf("never persisted"), controller.messages.value.map { it.content.single().text }) + + advanceTimeBy(120_000) + runCurrent() + + assertTrue(controller.messages.value.isEmpty()) + assertEquals("Timed out waiting for a reply; try again or refresh.", controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun lateTerminalAfterTimeoutRefreshesHistoryWithoutClearingNewerRun() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("slow first", "off", emptyList())) + val firstRunId = requireNotNull(gateway.lastRunId) + advanceTimeBy(120_000) + runCurrent() + assertEquals("Timed out waiting for a reply; try again or refresh.", controller.errorText.value) + + assertTrue(controller.sendMessageAwaitAcceptance("newer work", "off", emptyList())) + val secondRunId = requireNotNull(gateway.lastRunId) + controller.handleGatewayEvent( + "chat", + chatDeltaPayload("main", secondRunId, 1, "new", "new reply"), + ) + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "slow first", 1_000, idempotencyKey = "$firstRunId:user"), + ReplayHistoryMessage("assistant", "slow done", 2_000), + ), + ), + ) + + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", firstRunId, seq = 2, assistantText = "slow done"), + ) + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("new reply", controller.streamingAssistantText.value) + assertNull(controller.errorText.value) + assertEquals( + listOf("slow first", "slow done", "newer work"), + controller.messages.value.map { it.content.single().text }, + ) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun staleRecoveryCompletionCannotCancelNewerReconciliation() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + assertTrue(controller.sendMessageAwaitAcceptance("ordered recovery", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + + val firstRecoveryStarted = CompletableDeferred() + val releaseFirstRecovery = CompletableDeferred() + var recoveryCall = 0 + gateway.respond("chat.history") { + recoveryCall += 1 + when (recoveryCall) { + 1 -> { + firstRecoveryStarted.complete(Unit) + releaseFirstRecovery.await() + } + 2 -> historyResponse("session-1", emptyList()) + else -> + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "ordered recovery", 1_000, idempotencyKey = "$runId:user"), + ReplayHistoryMessage("assistant", "done", 2_000), + ), + ) + } + } + + controller.handleGatewayEvent("seqGap", null) + runCurrent() + firstRecoveryStarted.await() + controller.handleGatewayEvent("seqGap", null) + runCurrent() + releaseFirstRecovery.complete(historyResponse("session-1", emptyList())) + runCurrent() + advanceTimeBy(750) + runCurrent() + + assertEquals(listOf("ordered recovery", "done"), controller.messages.value.map { it.content.single().text }) + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun olderSameGenerationRetryCannotOverwriteTerminalHistory() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + assertTrue(controller.sendMessageAwaitAcceptance("ordered result", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + controller.handleGatewayEvent("seqGap", null) + runCurrent() + + val retryStarted = CompletableDeferred() + val releaseRetry = CompletableDeferred() + var historyCall = 0 + gateway.respond("chat.history") { + historyCall += 1 + if (historyCall == 1) { + retryStarted.complete(Unit) + releaseRetry.await() + } else { + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "ordered result", 1_000, idempotencyKey = "$runId:user"), + ReplayHistoryMessage("assistant", "done", 2_000), + ), + ) + } + } + advanceTimeBy(750) + runCurrent() + retryStarted.await() + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", runId, seq = 2, assistantText = "done"), + ) + runCurrent() + releaseRetry.complete(historyResponse("session-1", emptyList())) + runCurrent() + + assertEquals(listOf("ordered result", "done"), controller.messages.value.map { it.content.single().text }) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun newerSameGenerationHistoryCompletionSuppressesOlderFailureAndClearsLoading() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + assertTrue(controller.sendMessageAwaitAcceptance("ordered loading", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + + val recoveryStarted = CompletableDeferred() + val releaseRecovery = CompletableDeferred() + var historyCall = 0 + gateway.respond("chat.history") { + historyCall += 1 + if (historyCall == 1) { + recoveryStarted.complete(Unit) + releaseRecovery.await() + } else { + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "ordered loading", 1_000, idempotencyKey = "$runId:user"), + ReplayHistoryMessage("assistant", "done", 2_000), + ), + ) + } + } + + controller.handleGatewayEvent("seqGap", null) + runCurrent() + recoveryStarted.await() + assertTrue(controller.historyLoading.value) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", runId, seq = 2, assistantText = "done"), + ) + runCurrent() + + assertFalse(controller.historyLoading.value) + assertEquals(listOf("ordered loading", "done"), controller.messages.value.map { it.content.single().text }) + + releaseRecovery.completeExceptionally(IllegalStateException("older history failed")) + runCurrent() + assertFalse(controller.historyLoading.value) + assertEquals(listOf("ordered loading", "done"), controller.messages.value.map { it.content.single().text }) + assertNull(controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun seqGapStaleSnapshotCannotReplaceLocallyOwnedRun() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.load("main") + runCurrent() + + assertTrue(controller.sendMessageAwaitAcceptance("new work", "off", emptyList())) + val localRunId = requireNotNull(gateway.lastRunId) + gateway.respondWith( + "chat.history", + historyResponse("session-1", emptyList(), inFlightRun = "run-stale" to "old text"), + ) + controller.handleGatewayEvent("seqGap", null) + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertNull(controller.streamingAssistantText.value) + controller.handleGatewayEvent( + "chat", + chatDeltaPayload("main", localRunId, 1, "ours", "ours"), + ) + assertEquals("ours", controller.streamingAssistantText.value) + assertEquals(listOf("new work"), controller.messages.value.map { it.content.single().text }) } @Test diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerStreamReplayTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerStreamReplayTest.kt index e5e3c1f25e6e..52ccb4c3c719 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerStreamReplayTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerStreamReplayTest.kt @@ -22,12 +22,15 @@ import org.junit.Test class ChatControllerStreamReplayTest { private val json = Json { ignoreUnknownKeys = true } - private fun TestScope.newController(gateway: ScriptedGateway): ChatController = - ChatController(scope = this, json = json, requestGateway = gateway::request) + private fun TestScope.newController(gateway: ScriptedGateway): ChatController = ChatController(scope = this, json = json, requestGateway = gateway::request) private fun transcript(controller: ChatController): List> = controller.messages.value.map { message -> - message.role to message.content.firstOrNull { it.type == "text" }?.text + val text = + message.content + .firstOrNull { it.type == "text" } + ?.text + message.role to text } @Test @@ -42,7 +45,10 @@ class ChatControllerStreamReplayTest { assertTrue(controller.sendMessageAwaitAcceptance("Hello there", "off", emptyList())) val runId = requireNotNull(gateway.lastRunId) assertEquals(1, controller.pendingRunCount.value) - val optimisticUserId = controller.messages.value.single { it.role == "user" }.id + val optimisticUserId = + controller.messages.value + .single { it.role == "user" } + .id controller.handleGatewayEvent("chat", chatDeltaPayload("main", runId, 1, "Str", "Str")) assertEquals("Str", controller.streamingAssistantText.value) @@ -63,7 +69,10 @@ class ChatControllerStreamReplayTest { ), ), ) - controller.handleGatewayEvent("chat", chatTerminalPayload("main", runId, seq = 3)) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", runId, seq = 3, assistantText = "Streamed reply."), + ) advanceUntilIdle() assertEquals( @@ -71,7 +80,12 @@ class ChatControllerStreamReplayTest { transcript(controller), ) // Gateway copy replaces the optimistic echo in place: same row identity, no duplicate. - assertEquals(optimisticUserId, controller.messages.value.single { it.role == "user" }.id) + assertEquals( + optimisticUserId, + controller.messages.value + .single { it.role == "user" } + .id, + ) assertEquals(0, controller.pendingRunCount.value) assertNull(controller.streamingAssistantText.value) assertNull(controller.errorText.value) @@ -106,16 +120,16 @@ class ChatControllerStreamReplayTest { ), ), ) - val terminal = chatTerminalPayload("main", runId, seq = 2) + val terminal = chatTerminalPayload("main", runId, seq = 2, assistantText = "Only once.") controller.handleGatewayEvent("chat", terminal) advanceUntilIdle() val idsAfterFirstTerminal = controller.messages.value.map { it.id } - // Redelivered terminal event triggers a second history refresh with the same payload. + // Once ownership resolves, redelivered terminal events are ignored. controller.handleGatewayEvent("chat", terminal) advanceUntilIdle() - assertEquals(2, gateway.callCount("chat.history")) + assertEquals(1, gateway.callCount("chat.history")) assertEquals( listOf("user" to "dedupe me", "assistant" to "Only once."), transcript(controller), @@ -151,6 +165,69 @@ class ChatControllerStreamReplayTest { assertEquals("Timed out waiting for a reply; try again or refresh.", controller.errorText.value) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun failedTerminalKeepsAcceptedUserUntilHistoryConfirmsIt() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondChatSend(status = "started") + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.handleGatewayEvent("health", null) + + assertTrue(controller.sendMessageAwaitAcceptance("failed send", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + controller.handleGatewayEvent("chat", chatTerminalPayload("main", runId, seq = 1, state = "error")) + runCurrent() + + assertEquals(0, controller.pendingRunCount.value) + assertTrue(transcript(controller).contains("user" to "failed send")) + assertEquals("Chat failed", controller.errorText.value) + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf(ReplayHistoryMessage("user", "failed send", 1_000, idempotencyKey = "$runId:user")), + ), + ) + advanceTimeBy(750) + runCurrent() + assertEquals(listOf("user" to "failed send"), transcript(controller)) + assertEquals("Chat failed", controller.errorText.value) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun messageLessSuccessfulTerminalResolvesAfterUserPersists() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondChatSend(status = "started") + val controller = newController(gateway) + controller.handleGatewayEvent("health", null) + + assertTrue(controller.sendMessageAwaitAcceptance("no output", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf(ReplayHistoryMessage("user", "no output", 1_000, idempotencyKey = "$runId:user")), + ), + ) + controller.handleGatewayEvent("chat", chatTerminalPayload("main", runId, seq = 1)) + runCurrent() + + assertEquals(listOf("user" to "no output"), transcript(controller)) + assertEquals(0, controller.pendingRunCount.value) + assertNull(controller.errorText.value) + + advanceTimeBy(120_000) + runCurrent() + assertEquals(listOf("user" to "no output"), transcript(controller)) + assertNull(controller.errorText.value) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun reconnectMidRunClearsTransientStateAndHistoryConverges() = @@ -162,7 +239,10 @@ class ChatControllerStreamReplayTest { assertTrue(controller.sendMessageAwaitAcceptance("survive reconnect", "off", emptyList())) val runId = requireNotNull(gateway.lastRunId) - val optimisticUserId = controller.messages.value.single { it.role == "user" }.id + val optimisticUserId = + controller.messages.value + .single { it.role == "user" } + .id controller.handleGatewayEvent( "chat", @@ -203,7 +283,12 @@ class ChatControllerStreamReplayTest { listOf("user" to "survive reconnect", "assistant" to "Recovered reply."), transcript(controller), ) - assertEquals(optimisticUserId, controller.messages.value.single { it.role == "user" }.id) + assertEquals( + optimisticUserId, + controller.messages.value + .single { it.role == "user" } + .id, + ) assertEquals("session-1", controller.sessionId.value) // Disconnect cancelled the 120s ack timer: the converged transcript must not decay. @@ -255,6 +340,7 @@ class ChatControllerStreamReplayTest { chatTerminalPayload("main", runId = "external-run", seq = 1), ) runCurrent() // history refetch for "main" is now suspended on the gate + assertEquals(2, gateway.callCount("chat.history")) controller.switchSession("other") advanceUntilIdle() @@ -269,6 +355,33 @@ class ChatControllerStreamReplayTest { assertEquals("session-other", controller.sessionId.value) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun unknownTerminalRefreshesIdleTranscript() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + listOf(ReplayHistoryMessage("assistant", "from another client", 2_000)), + ), + ) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", "external-run", seq = 1, assistantText = "from another client"), + ) + runCurrent() + + assertEquals(listOf("assistant" to "from another client"), transcript(controller)) + assertNull(controller.errorText.value) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun markdownFixtureStreamsByteIdenticalAndConvergesLosslessly() = @@ -316,7 +429,7 @@ class ChatControllerStreamReplayTest { ) controller.handleGatewayEvent( "chat", - chatTerminalPayload("main", runId, seq = chunks.size + 1), + chatTerminalPayload("main", runId, seq = chunks.size + 1, assistantText = fixture), ) advanceUntilIdle() diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerTerminalAckTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerTerminalAckTest.kt index 808bda40fb68..14b128b39f9b 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerTerminalAckTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerTerminalAckTest.kt @@ -1,9 +1,14 @@ package ai.openclaw.app.chat +import ai.openclaw.app.gateway.GatewayRequestNotEnqueued +import ai.openclaw.app.gateway.GatewayRequestRejected +import ai.openclaw.app.gateway.GatewaySession import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.Json +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNull @@ -68,6 +73,58 @@ class ChatControllerTerminalAckTest { assertTrue(controller.messages.value.hasUserText("message that started")) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun canonicalAckRunIdPreservesClientHistoryIdentity() = + runTest { + var clientRunId: String? = null + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, paramsJson -> + when (method) { + "chat.send" -> { + clientRunId = + requireNotNull(paramsJson) + .let(json::parseToJsonElement) + .jsonObject["idempotencyKey"] + ?.jsonPrimitive + ?.content + """{"runId":"canonical-run","status":"started"}""" + } + "chat.history" -> + historyResponse( + "session-1", + listOf( + ReplayHistoryMessage("user", "canonical", 1_000, idempotencyKey = "$clientRunId:user"), + ReplayHistoryMessage("assistant", "done", 2_000), + ), + ) + else -> "{}" + } + }, + ) + controller.handleGatewayEvent("health", null) + + assertTrue(controller.sendMessageAwaitAcceptance("canonical", "off", emptyList())) + controller.handleGatewayEvent( + "chat", + chatTerminalPayload("main", "canonical-run", seq = 2, assistantText = "done"), + ) + advanceUntilIdle() + + assertEquals(0, controller.pendingRunCount.value) + assertEquals(1, controller.messages.value.count { it.role == "user" }) + assertEquals( + "$clientRunId:user", + controller.messages.value + .single { it.role == "user" } + .idempotencyKey, + ) + assertNull(controller.errorText.value) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun terminalOkAckClearsOptimisticUserEchoAndRefreshesHistory() = @@ -140,6 +197,48 @@ class ChatControllerTerminalAckTest { assertFalse(controller.messages.value.hasUserText("message that errors before start")) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun definitiveRpcRejectionRestoresComposerOwnership() = + runTest { + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { _, _ -> + throw GatewayRequestRejected(GatewaySession.ErrorShape("INVALID_REQUEST", "message rejected")) + }, + ) + controller.handleGatewayEvent("health", null) + + val accepted = controller.sendMessageAwaitAcceptance("rejected", "off", emptyList()) + + assertFalse(accepted) + assertEquals(0, controller.pendingRunCount.value) + assertEquals("INVALID_REQUEST: message rejected", controller.errorText.value) + assertFalse(controller.messages.value.hasUserText("rejected")) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun requestNotEnqueuedRestoresComposerOwnership() = + runTest { + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { _, _ -> throw GatewayRequestNotEnqueued("not connected") }, + ) + controller.handleGatewayEvent("health", null) + + val accepted = controller.sendMessageAwaitAcceptance("never sent", "off", emptyList()) + + assertFalse(accepted) + assertEquals(0, controller.pendingRunCount.value) + assertEquals("not connected", controller.errorText.value) + assertFalse(controller.messages.value.hasUserText("never sent")) + } + private fun List.hasUserText(text: String): Boolean = any { message -> message.role == "user" && message.content.any { part -> part.text == text } diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt index 2b59c1835037..a25000935a84 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt @@ -15,7 +15,9 @@ import kotlinx.serialization.json.jsonPrimitive * chat/agent events through ChatController.handleGatewayEvent under * kotlinx-coroutines-test virtual time. */ -internal class ScriptedGateway(private val json: Json) { +internal class ScriptedGateway( + private val json: Json, +) { data class Call( val method: String, val paramsJson: String?, @@ -54,7 +56,13 @@ internal class ScriptedGateway(private val json: Json) { respond("chat.send") { paramsJson -> val runId = paramsJson - ?.let { json.parseToJsonElement(it).jsonObject["idempotencyKey"]?.jsonPrimitive?.content } + ?.let { value -> + json + .parseToJsonElement(value) + .jsonObject["idempotencyKey"] + ?.jsonPrimitive + ?.content + } lastRunId = runId buildJsonObject { if (runId != null) put("runId", JsonPrimitive(runId)) @@ -73,7 +81,13 @@ internal class ScriptedGateway(private val json: Json) { } fun sessionKeyOf(paramsJson: String?): String? = - paramsJson?.let { json.parseToJsonElement(it).jsonObject["sessionKey"]?.jsonPrimitive?.content } + paramsJson?.let { value -> + json + .parseToJsonElement(value) + .jsonObject["sessionKey"] + ?.jsonPrimitive + ?.content + } fun callCount(method: String): Int = calls.count { it.method == method } } @@ -158,12 +172,32 @@ internal fun chatTerminalPayload( runId: String, seq: Int, state: String = "final", + assistantText: String? = null, ): String = buildJsonObject { put("sessionKey", JsonPrimitive(sessionKey)) put("runId", JsonPrimitive(runId)) put("seq", JsonPrimitive(seq)) put("state", JsonPrimitive(state)) + if (assistantText != null) { + put( + "message", + buildJsonObject { + put("role", JsonPrimitive("assistant")) + put( + "content", + JsonArray( + listOf( + buildJsonObject { + put("type", JsonPrimitive("text")) + put("text", JsonPrimitive(assistantText)) + }, + ), + ), + ) + }, + ) + } }.toString() /** diff --git a/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt index 52302d1cc769..d1333164961e 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt @@ -131,7 +131,7 @@ class GatewaySessionInvokeTest { } @Test - fun disconnectCancelsPendingRpcWithoutWaitingForRequestTimeout() { + fun disconnectFailsPendingRpcWithUnknownOutcomeWithoutWaitingForTimeout() { runBlocking { val json = testJson() val connected = CompletableDeferred() @@ -173,7 +173,7 @@ class GatewaySessionInvokeTest { harness.session.disconnect() val result = withTimeout(2_000) { requestResult.await() } - assertEquals(true, result.exceptionOrNull() is CancellationException) + assertEquals(true, result.exceptionOrNull() is GatewayRequestOutcomeUnknown) serverWebSocket.get()?.close(1000, "done") withTimeoutOrNull(2_000) { while (lastDisconnect.get().isEmpty()) delay(10) @@ -189,6 +189,56 @@ class GatewaySessionInvokeTest { } } + @Test + fun disconnectReportsUnknownOutcomeForFireAndForgetRpc() { + runBlocking { + val json = testJson() + val connected = CompletableDeferred() + val requestSeen = CompletableDeferred() + val requestError = CompletableDeferred() + val lastDisconnect = AtomicReference("") + val serverWebSocket = AtomicReference(null) + val server = + startGatewayServer(json) { webSocket, id, method, _ -> + serverWebSocket.set(webSocket) + when (method) { + "connect" -> webSocket.send(connectResponseFrame(id)) + "fire.and.forget" -> requestSeen.complete(Unit) + } + } + val harness = + createNodeHarness( + connected = connected, + lastDisconnect = lastDisconnect, + ) { GatewaySession.InvokeResult.ok("""{"handled":true}""") } + + try { + connectNodeSession(harness.session, server.port) + awaitConnectedOrThrow(connected, lastDisconnect, server) + harness.session.sendRequestFrame( + method = "fire.and.forget", + paramsJson = null, + timeoutMs = 30_000, + onError = { requestError.complete(it) }, + ) + withTimeout(TEST_TIMEOUT_MS) { requestSeen.await() } + + harness.session.disconnect() + + val error = withTimeout(2_000) { requestError.await() } + assertEquals("UNAVAILABLE", error.code) + assertEquals("Gateway disconnected before response", error.message) + serverWebSocket.get()?.close(1000, "done") + } finally { + runCatching { serverWebSocket.get()?.close(1000, "done") } + delay(100) + harness.session.disconnect() + harness.sessionJob.cancelAndJoin() + server.shutdown() + } + } + } + @Test fun eventsAreDispatchedInWebSocketFrameOrder() = runBlocking { diff --git a/apps/android/app/src/test/java/ai/openclaw/app/ui/chat/ChatSheetContentTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/ui/chat/ChatSheetContentTest.kt index 832aa19aa9c7..c77a1c10635b 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/ui/chat/ChatSheetContentTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/ui/chat/ChatSheetContentTest.kt @@ -1,10 +1,7 @@ package ai.openclaw.app.ui.chat -import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals -import org.junit.Assert.assertFalse import org.junit.Assert.assertNull -import org.junit.Assert.assertTrue import org.junit.Test class ChatSheetContentTest { @@ -34,44 +31,6 @@ class ChatSheetContentTest { ) } - @Test - fun keepsPendingAssistantAutoSendWhenDispatchRejected() = - runBlocking { - var dispatchedPrompt: String? = null - - val consumed = - dispatchPendingAssistantAutoSend( - pendingPrompt = "summarize mail", - healthOk = true, - pendingRunCount = 0, - ) { prompt -> - dispatchedPrompt = prompt - false - } - - assertFalse(consumed) - assertEquals("summarize mail", dispatchedPrompt) - } - - @Test - fun clearsPendingAssistantAutoSendOnlyAfterAcceptedDispatch() = - runBlocking { - var dispatchedPrompt: String? = null - - val consumed = - dispatchPendingAssistantAutoSend( - pendingPrompt = "summarize mail", - healthOk = true, - pendingRunCount = 0, - ) { prompt -> - dispatchedPrompt = prompt - true - } - - assertTrue(consumed) - assertEquals("summarize mail", dispatchedPrompt) - } - @Test fun initialChatLoadUsesMainWhenNoSessionIsSelected() { assertEquals(