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
2026-07-09 04:54:01 +00:00 · 2026-07-06 03:50:09 +01:00
parent 75c8753a2c
commit 2b48e98148
14 changed files with 1941 additions and 224 deletions
--- 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.
--- 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",
--- 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<String?> = _chatDraft
  private val _pendingAssistantAutoSend = MutableStateFlow<String?>(null)
  val pendingAssistantAutoSend: StateFlow<String?> = _pendingAssistantAutoSend
+  private val _assistantAutoSendInFlight = MutableStateFlow(false)
+  val assistantAutoSendInFlight: StateFlow<Boolean> = _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) {
--- 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
--- 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<List<ChatCommandEntry>> = _commands.asStateFlow()

  private val pendingRuns = mutableSetOf<String>()
+  private val disconnectedPendingRunIds = mutableSetOf<String>()
+  private val timedOutRunIds = ConcurrentHashMap.newKeySet<String>()
+  private val terminalWithoutReplyRunIds = ConcurrentHashMap.newKeySet<String>()
+  private val unknownOutcomeRunIds = ConcurrentHashMap.newKeySet<String>()
  private val pendingRunTimeoutJobs = ConcurrentHashMap<String, Job>()

  // Preserve sent messages locally until chat.history includes the gateway-confirmed copy.
-  private val optimisticMessagesByRunId = LinkedHashMap<String, ChatMessage>()
+  private val optimisticMessagesByRunId = ConcurrentHashMap<String, ChatMessage>()
+  // Keep reply ownership after the user row persists; the assistant row can land later.
+  private val unresolvedRepliesByRunId = ConcurrentHashMap<String, ChatMessage>()
  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<List<ChatOutboxItem>>(emptyList())
  val outboxItems: StateFlow<List<ChatOutboxItem>> = _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<String> = 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<String> = 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<ChatMessage>) {
    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<ChatMessage>) {
+    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<String>,
+  ) {
+    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<String>,
+  ) {
+    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<String> = 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)
+        }
      }
    }
  }
--- 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"))
      }
    }
  }
--- 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("") }
--- 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
--- 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
--- 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<Pair<String, String?>> =
    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()

--- 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<ChatMessage>.hasUserText(text: String): Boolean =
    any { message ->
      message.role == "user" && message.content.any { part -> part.text == text }
--- 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()

 /**
--- 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<Unit>()
@@ -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<Unit>()
+      val requestSeen = CompletableDeferred<Unit>()
+      val requestError = CompletableDeferred<GatewaySession.ErrorShape>()
+      val lastDisconnect = AtomicReference("")
+      val serverWebSocket = AtomicReference<WebSocket?>(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 {
--- 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(