From 5c1857fc55dc14e4fc623ea53d4138aaaea83f0e Mon Sep 17 00:00:00 2001 From: Leandre Chamberland-Dozois Date: Thu, 16 Jul 2026 05:12:45 -0400 Subject: [PATCH] fix(android): stop killing long chat runs with hardcoded timeouts (#106863) (#106864) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(android): stop killing long chat runs with hardcoded timeouts Two independent timeouts made any agent turn longer than ~2 minutes fail when started from the Android app, while the same turn worked over Telegram (#106863): - chat.send hardcoded timeoutMs=30000, which the gateway turns into the server-side run expiry; the maintenance sweep then aborts longer runs ("CLI run aborted" / "Embedded agent failed before reply"). Omit the override so app runs get the configured default like other channels. - armPendingRunTimeout declared a timeout after 120s even when the refreshed history snapshot confirmed the run was still in flight, showing a spurious error and dropping the optimistic bubble mid-run. Re-arm the timer in that case; terminal events and the server-side expiry remain the liveness backstop. Fixes #106863 Co-Authored-By: Claude Fable 5 * chore(i18n): sync native inventory after ChatController line shifts Co-Authored-By: Claude Fable 5 * test(android): run ChatController on backgroundScope in reconnect tests The pending-run watchdog now re-arms while a run is confirmed in flight, so reconnect scenarios that end with a restored in-flight run keep a timer coroutine alive. Launch the controller on runTest's backgroundScope so that timer is cancelled at test end instead of failing the suite with UncompletedCoroutinesError. Co-Authored-By: Claude Fable 5 * fix(android): bound pending-run watchdog recovery Distinguish applied, superseded, and failed history refreshes so recovered runs neither expire during a newer authoritative load nor rearm forever after refresh failure. Co-authored-by: Léandre Chamberland-Dozois --------- Co-authored-by: Claude Fable 5 Co-authored-by: Peter Steinberger --- apps/.i18n/native-source.json | 28 +++--- .../ai/openclaw/app/chat/ChatController.kt | 82 +++++++++++------ .../ChatControllerReconnectRestoreTest.kt | 89 ++++++++++++++++++- 3 files changed, 156 insertions(+), 43 deletions(-) diff --git a/apps/.i18n/native-source.json b/apps/.i18n/native-source.json index c0d010280bb2..aac7f6e773ba 100644 --- a/apps/.i18n/native-source.json +++ b/apps/.i18n/native-source.json @@ -1723,7 +1723,7 @@ }, { "kind": "ui-call", - "line": 763, + "line": 769, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Wait for the current response to finish before starting a new chat.", "surface": "android", @@ -1731,7 +1731,7 @@ }, { "kind": "ui-call", - "line": 894, + "line": 900, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not update model.", "surface": "android", @@ -1739,7 +1739,7 @@ }, { "kind": "ui-call", - "line": 958, + "line": 964, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not update thinking level.", "surface": "android", @@ -1747,7 +1747,7 @@ }, { "kind": "ui-call", - "line": 1337, + "line": 1343, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Chat failed before the run started; try again.", "surface": "android", @@ -1755,7 +1755,7 @@ }, { "kind": "ui-call", - "line": 2140, + "line": 2148, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not stage an attachment for sending.", "surface": "android", @@ -1763,7 +1763,7 @@ }, { "kind": "ui-call", - "line": 2160, + "line": 2168, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not queue message for later delivery.", "surface": "android", @@ -1771,7 +1771,7 @@ }, { "kind": "ui-call", - "line": 2170, + "line": 2178, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Offline queue is full ($OUTBOX_MAX_QUEUED messages); delete queued items first.", "surface": "android", @@ -1779,7 +1779,7 @@ }, { "kind": "ui-call", - "line": 2174, + "line": 2182, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Attachments are too large to queue for one message; remove some and try again.", "surface": "android", @@ -1787,7 +1787,7 @@ }, { "kind": "ui-call", - "line": 2178, + "line": 2186, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Offline attachment storage is full; delete queued items first.", "surface": "android", @@ -1795,7 +1795,7 @@ }, { "kind": "ui-call", - "line": 2182, + "line": 2190, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Gateway health not OK; cannot send", "surface": "android", @@ -1803,7 +1803,7 @@ }, { "kind": "ui-call", - "line": 2813, + "line": 2821, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Chat failed", "surface": "android", @@ -1811,7 +1811,7 @@ }, { "kind": "ui-call", - "line": 2916, + "line": 2924, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Event stream interrupted; try refreshing.", "surface": "android", @@ -1819,7 +1819,7 @@ }, { "kind": "ui-call", - "line": 2996, + "line": 3024, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Timed out waiting for a reply; try again or refresh.", "surface": "android", @@ -1827,7 +1827,7 @@ }, { "kind": "ui-call", - "line": 3188, + "line": 3216, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Timed out confirming the sent message; refresh to check delivery.", "surface": "android", 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 9e694cc9f16b..2d81c69f7cba 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 @@ -164,6 +164,12 @@ class ChatController internal constructor( val generation: Long, ) + private enum class HistoryRefreshResult { + Applied, + Superseded, + Failed, + } + @Volatile private var liveHistoryMarker: LiveHistoryMarker? = null @@ -1442,7 +1448,9 @@ class ChatController internal constructor( put("sessionKey", JsonPrimitive(sessionKey)) put("message", JsonPrimitive(text)) put("thinking", JsonPrimitive(thinking)) - put("timeoutMs", JsonPrimitive(30_000)) + // No timeoutMs override: it becomes the server-side run expiry, and agent + // turns can legitimately run for many minutes. Omitting it applies the + // gateway's configured default, same as other channels. put("idempotencyKey", JsonPrimitive(idempotencyKey)) if (attachments.isNotEmpty()) { put( @@ -1673,14 +1681,14 @@ class ChatController internal constructor( // live chat.history response always replaces cached rows wholesale. primeFromCache(sessionKey, generation) try { - val historyApplied = + val historyResult = fetchAndApplyHistory( sessionKey, generation, updateSessionInfo = true, runIdsToReconcile = runIdsToReconcile, ) - if (!historyApplied) return + if (historyResult != HistoryRefreshResult.Applied) return if (!ownsReconnectRecovery) { pollHealthIfNeeded(force = forceHealth) @@ -1705,14 +1713,14 @@ class ChatController internal constructor( /** * Requests live history and applies it to controller state, replacing any cached transcript. - * Returns false when a newer load superseded this request (stale responses are dropped). + * Reports when a newer load superseded this request (stale responses are dropped). */ private suspend fun fetchAndApplyHistory( sessionKey: String, generation: Long, updateSessionInfo: Boolean, runIdsToReconcile: Set = emptySet(), - ): Boolean { + ): HistoryRefreshResult { val requestSequence = historyRequestSequence.incrementAndGet() val requestModelSelectionGeneration = modelSelectionGeneration.get() val requestCacheScope = currentCacheScope() @@ -1721,7 +1729,7 @@ class ChatController internal constructor( !isCurrentHistoryLoad(sessionKey, _sessionKey.value, generation, historyLoadGeneration.get()) || requestCacheScope != currentCacheScope() ) { - return false + return HistoryRefreshResult.Superseded } val history = try { @@ -1741,7 +1749,7 @@ class ChatController internal constructor( requestCacheScope != currentCacheScope() || requestSequence < latestAppliedHistoryRequest } - if (superseded) return false + if (superseded) return HistoryRefreshResult.Superseded throw err } val applied = @@ -1812,11 +1820,11 @@ class ChatController internal constructor( ?.let { _thinkingLevel.value = it } true } - if (!applied) return false + if (!applied) return HistoryRefreshResult.Superseded completeReconnectRecoveryIfOwned(sessionKey, generation) persistTranscript(requestCacheScope, sessionKey, history.messages) confirmDurableSendsFromHistory(requestCacheScope, history) - return true + return HistoryRefreshResult.Applied } private suspend fun awaitMainSessionReadiness( @@ -2970,25 +2978,45 @@ 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) + val watchdogSessionKey = _sessionKey.value + val latestAppliedBeforeRefresh = + synchronized(gatewayScopeApplyLock) { + latestAppliedHistoryRequest } - if (!stillPending && !replyStillUnresolved) return@launch + val historyResult = + refreshHistorySnapshotBestEffort( + sessionKey = watchdogSessionKey, + generation = historyLoadGeneration.get(), + runIdsToReconcile = emptySet(), + ) + val refreshState = + synchronized(gatewayScopeApplyLock) { + // A concurrent recovery load can supersede this request. Its newer + // current-session snapshot is equally authoritative confirmation. + val currentSession = watchdogSessionKey == _sessionKey.value + val freshSnapshotApplied = + historyResult == HistoryRefreshResult.Applied || latestAppliedHistoryRequest > latestAppliedBeforeRefresh + Triple(currentSession, freshSnapshotApplied, latestAppliedInFlightRunId == runId) + } + val (currentSession, freshSnapshotApplied, latestRunMatches) = refreshState + if (currentSession && freshSnapshotApplied && latestRunMatches) { + // The refreshed snapshot confirms the run is still executing; long agent + // turns can outlast one timeout window, so keep waiting instead of + // surfacing a false timeout and dropping the optimistic bubble. Terminal + // events and the server-side expiry remain the liveness backstop. + armPendingRunTimeout(runId) + return@launch + } + if (currentSession && !freshSnapshotApplied && historyResult == HistoryRefreshResult.Superseded) { + // The newer current-session load owns reconciliation but has not applied + // yet. Defer expiry; its snapshot or the next watchdog decides the run. + armPendingRunTimeout(runId) + return@launch + } + val replyStillUnresolved = unresolvedRepliesByRunId.containsKey(runId) clearPendingRun(runId) clearTransientRunUiIfIdle() + if (!replyStillUnresolved) return@launch removeOptimisticMessage(runId) unresolvedRepliesByRunId.remove(runId) terminalWithoutReplyRunIds.remove(runId) @@ -3197,7 +3225,7 @@ class ChatController internal constructor( sessionKey: String, generation: Long, runIdsToReconcile: Set, - ) { + ): HistoryRefreshResult = try { fetchAndApplyHistory( sessionKey, @@ -3209,8 +3237,8 @@ class ChatController internal constructor( throw err } catch (_: Throwable) { // The bounded expiry below remains the final reconciliation path. + HistoryRefreshResult.Failed } - } private fun refreshCurrentHistoryBestEffort( runIdsToReconcile: Set = emptySet(), 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 9705069c2a15..b1215eedbff0 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 @@ -25,11 +25,14 @@ 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) + // The controller runs on backgroundScope: while a restored run stays in flight the + // pending-run watchdog keeps re-arming, so its timer must be cancelled by runTest + // instead of counting as an uncompleted test coroutine. + private fun TestScope.newController(gateway: ScriptedGateway): ChatController = ChatController(scope = backgroundScope, json = json, requestGateway = gateway::request) private fun TestScope.newScopedController(gateway: ScriptedGateway): ChatController = ChatController( - scope = this, + scope = backgroundScope, json = json, requestGateway = gateway::request, requestGatewayForGateway = { _, method, paramsJson -> gateway.request(method, paramsJson) }, @@ -549,6 +552,88 @@ class ChatControllerReconnectRestoreTest { assertTrue(controller.pendingToolCalls.value.isEmpty()) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun recoveredPendingRunStopsWatchdogWhenRefreshFails() = + 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) + + gateway.respond("chat.history") { error("history unavailable") } + advanceTimeBy(120_000) + runCurrent() + + assertEquals(2, gateway.callCount("chat.history")) + assertEquals(0, controller.pendingRunCount.value) + assertNull(controller.streamingAssistantText.value) + + advanceTimeBy(120_000) + runCurrent() + assertEquals(2, gateway.callCount("chat.history")) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun newerRecoverySnapshotCanSupersedePendingRunWatchdogRefresh() = + 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 watchdogRefreshStarted = CompletableDeferred() + val releaseWatchdogRefresh = CompletableDeferred() + val newerRefreshStarted = CompletableDeferred() + val releaseNewerRefresh = CompletableDeferred() + var refreshCalls = 0 + gateway.respond("chat.history") { + refreshCalls += 1 + if (refreshCalls == 1) { + watchdogRefreshStarted.complete(Unit) + releaseWatchdogRefresh.await() + } else { + newerRefreshStarted.complete(Unit) + releaseNewerRefresh.await() + } + } + + advanceTimeBy(120_000) + runCurrent() + watchdogRefreshStarted.await() + controller.refresh() + runCurrent() + newerRefreshStarted.await() + releaseWatchdogRefresh.complete( + historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "stale working"), + ) + runCurrent() + + assertEquals(1, controller.pendingRunCount.value) + assertEquals("working", controller.streamingAssistantText.value) + assertNull(controller.errorText.value) + + releaseNewerRefresh.complete( + historyResponse("session-1", listOf(userTurn), inFlightRun = "run-active" to "still working"), + ) + runCurrent() + + assertEquals(3, gateway.callCount("chat.history")) + assertEquals(1, controller.pendingRunCount.value) + assertEquals("still working", controller.streamingAssistantText.value) + assertNull(controller.errorText.value) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun explicitRefreshClearsPriorHistoryError() =