fix(wear): keep active watch replies when older turns finish (#113804)

* fix(wear): preserve active replies when older turns end

* fix(wear): isolate terminal events at the watch

---------

Co-authored-by: Peter Steinberger <steipete@golden-gate.local>

apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyBridge.kt

@@ -532,9 +532,7 @@ internal class WearChatStreamProjector {
     val streamKey = streamKey(projected)
     if (state != "delta") {
       if (state == "final" || state == "aborted" || state == "error") {
-        streamKey?.let { terminalKey ->
-          streams.keys.removeAll { key -> key.sessionKey == terminalKey.sessionKey }
-        }
+        streamKey?.let(::clearTerminalStream)
       }
       return projected
     }
@@ -583,14 +581,24 @@ internal class WearChatStreamProjector {
     }
   }
 
+  private fun clearTerminalStream(terminalKey: StreamKey) {
+    if (terminalKey.runId != null) {
+      // A delayed identified terminal cannot prove that an anonymous
+      // accumulator belongs to the same run or interrupt another live run.
+      streams.remove(terminalKey)
+      return
+    }
+    streams.keys.removeAll { key -> key.sessionKey == terminalKey.sessionKey }
+  }
+
   private fun streamKey(projected: JsonObject): StreamKey? {
     val sessionKey =
       (projected["sessionKey"] as? JsonPrimitive)
         ?.contentOrNull
         ?.takeIf { it.isNotBlank() } ?: return null
     val runId = (projected["runId"] as? JsonPrimitive)?.contentOrNull
-    // Some gateway deltas omit runId. Sessions serialize active runs, and every
-    // terminal event clears all keys for that session before another run starts.
+    // Anonymous deltas adopt the session's latest identified accumulator;
+    // only a runless terminal can safely retire every run in that session.
     return StreamKey(sessionKey = sessionKey, runId = runId)
   }
 

apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt

@@ -297,6 +297,65 @@ class WearProxyBridgeTest {
     assertEquals("false", unknownPrefix.getValue("streamTextComplete").jsonPrimitive.content)
   }
 
+  @Test
+  fun foreignFinalPreservesTheActiveWatchStream() {
+    assertForeignTerminalPreservesActiveStream("final")
+  }
+
+  @Test
+  fun foreignAbortPreservesTheActiveWatchStream() {
+    assertForeignTerminalPreservesActiveStream("aborted")
+  }
+
+  @Test
+  fun foreignErrorPreservesTheActiveWatchStream() {
+    assertForeignTerminalPreservesActiveStream("error")
+  }
+
+  @Test
+  fun foreignFinalPreservesAnAnonymousWatchStream() {
+    assertForeignTerminalPreservesAnonymousStream("final")
+  }
+
+  @Test
+  fun foreignAbortPreservesAnAnonymousWatchStream() {
+    assertForeignTerminalPreservesAnonymousStream("aborted")
+  }
+
+  @Test
+  fun foreignErrorPreservesAnAnonymousWatchStream() {
+    assertForeignTerminalPreservesAnonymousStream("error")
+  }
+
+  @Test
+  fun identifiedTerminalClearsItsOwnStreamWithoutErasingAnotherRun() {
+    val projector = WearChatStreamProjector()
+    projectStreamEvent(projector, state = "delta", runId = "older-run", text = "Old", message = "Old")
+    projectStreamEvent(projector, state = "delta", runId = "active-run", text = "Hel", message = "Hel")
+
+    projectStreamEvent(projector, state = "final", runId = "older-run")
+
+    val active = projectStreamEvent(projector, state = "delta", runId = "active-run", text = "lo")
+    val retired = projectStreamEvent(projector, state = "delta", runId = "older-run", text = "new")
+    assertEquals("Hello", active.getValue("streamText").jsonPrimitive.content)
+    assertEquals("true", active.getValue("streamTextComplete").jsonPrimitive.content)
+    assertEquals("new", retired.getValue("streamText").jsonPrimitive.content)
+    assertEquals("false", retired.getValue("streamTextComplete").jsonPrimitive.content)
+  }
+
+  @Test
+  fun unidentifiedTerminalClearsEveryStreamInItsSession() {
+    val projector = WearChatStreamProjector()
+    projectStreamEvent(projector, state = "delta", runId = "older-run", text = "old", message = "old")
+    projectStreamEvent(projector, state = "delta", runId = "active-run", text = "stale", message = "stale")
+
+    projectStreamEvent(projector, state = "final")
+
+    val next = projectStreamEvent(projector, state = "delta", runId = "active-run", text = "fresh")
+    assertEquals("fresh", next.getValue("streamText").jsonPrimitive.content)
+    assertEquals("false", next.getValue("streamTextComplete").jsonPrimitive.content)
+  }
+
   @Test
   fun runIdLessDeltasUseSessionSnapshotAndKeepExactAppendSemantics() {
     val projector = WearChatStreamProjector()
@@ -842,6 +901,55 @@ class WearProxyBridgeTest {
       assertEquals(2, sent.count { it.path == WearProtocol.EVENT_PATH })
     }
 
+  private fun assertForeignTerminalPreservesActiveStream(state: String) {
+    val projector = WearChatStreamProjector()
+    projectStreamEvent(projector, state = "delta", runId = "active-run", text = "Hel", message = "Hel")
+
+    projectStreamEvent(projector, state = state, runId = "older-run")
+
+    val continued = projectStreamEvent(projector, state = "delta", runId = "active-run", text = "lo")
+    assertEquals("Hello", continued.getValue("streamText").jsonPrimitive.content)
+    assertEquals("true", continued.getValue("streamTextComplete").jsonPrimitive.content)
+  }
+
+  private fun assertForeignTerminalPreservesAnonymousStream(state: String) {
+    val projector = WearChatStreamProjector()
+    projectStreamEvent(projector, state = "delta", text = "Hel", message = "Hel")
+
+    projectStreamEvent(projector, state = state, runId = "older-run")
+
+    val continued = projectStreamEvent(projector, state = "delta", text = "lo")
+    assertEquals("Hello", continued.getValue("streamText").jsonPrimitive.content)
+    assertEquals("true", continued.getValue("streamTextComplete").jsonPrimitive.content)
+  }
+
+  private fun projectStreamEvent(
+    projector: WearChatStreamProjector,
+    state: String,
+    runId: String? = null,
+    text: String? = null,
+    message: String? = null,
+  ): JsonObject =
+    checkNotNull(
+      projector.project(
+        buildJsonObject {
+          put("sessionKey", "main")
+          runId?.let { put("runId", it) }
+          put("state", state)
+          text?.let { put("deltaText", it) }
+          message?.let { fullText ->
+            put(
+              "message",
+              buildJsonObject {
+                put("role", "assistant")
+                put("content", fullText)
+              },
+            )
+          }
+        },
+      ),
+    )
+
   private fun request(requestId: String): WearMessage.Request = WearMessage.Request(requestId = requestId, method = WearRpcMethod.ProxyStatus)
 }
 

apps/android/wear/src/main/java/ai/openclaw/wear/WearViewModel.kt

@@ -116,6 +116,61 @@ internal fun shouldAcceptWearTalkSnapshot(
   attemptId: String?,
 ): Boolean = snapshot.attemptId != null && snapshot.attemptId == attemptId
 
+internal data class WearTerminalChatTransition(
+  val state: WearUiState,
+  val reloadHistory: Boolean,
+  val observedMessage: WearChatMessage? = null,
+)
+
+internal fun reduceWearTerminalChatEvent(
+  current: WearUiState,
+  event: WearChatEvent,
+): WearTerminalChatTransition {
+  if (event.sessionKey != current.selectedSession?.key) {
+    return WearTerminalChatTransition(state = current, reloadHistory = false)
+  }
+  val finalMessage = event.message?.takeIf { event.state == "final" }
+  val preservedState =
+    finalMessage?.let { message ->
+      current.copy(messages = mergeEventMessage(current.messages, message))
+    } ?: current
+  if (current.activeRunId != null && event.runId != null && current.activeRunId != event.runId) {
+    // Preserve older finals and notifications without canceling another
+    // identified run or replacing it with a stale history snapshot.
+    return WearTerminalChatTransition(state = preservedState, reloadHistory = false)
+  }
+  val hasLiveReply = current.activeRunId != null || !current.streamText.isNullOrBlank()
+  if (hasLiveReply && (current.activeRunId == null || event.runId == null)) {
+    // Missing run identity cannot distinguish a delayed terminal from the
+    // live run's first identified terminal. Preserve the live reply until
+    // authoritative history resolves which run actually remains active.
+    return WearTerminalChatTransition(
+      state = preservedState,
+      reloadHistory = true,
+      observedMessage = finalMessage,
+    )
+  }
+  return when (event.state) {
+    "final" ->
+      WearTerminalChatTransition(
+        state =
+          current.copy(
+            messages = event.message?.let { mergeEventMessage(current.messages, it) } ?: current.messages,
+            streamText = if (event.message == null) current.streamText else null,
+            activeRunId = null,
+          ),
+        reloadHistory = true,
+        observedMessage = event.message,
+      )
+    "aborted", "error" ->
+      WearTerminalChatTransition(
+        state = current.copy(streamText = null, activeRunId = null),
+        reloadHistory = true,
+      )
+    else -> WearTerminalChatTransition(state = current, reloadHistory = false)
+  }
+}
+
 internal class WearViewModel(
   application: Application,
 ) : AndroidViewModel(application) {
@@ -908,21 +963,13 @@ internal class WearViewModel(
           )
         }
       }
-      "final" -> {
-        cancelLoad()
-        mutableState.update { current ->
-          current.copy(
-            messages = event.message?.let { mergeEventMessage(current.messages, it) } ?: current.messages,
-            streamText = if (event.message == null) current.streamText else null,
-            activeRunId = null,
-          )
+      "final", "aborted", "error" -> {
+        val transition = reduceWearTerminalChatEvent(mutableState.value, event)
+        if (transition.reloadHistory) cancelLoad()
+        mutableState.value = transition.state
+        if (transition.reloadHistory) {
+          loadHistory(selected, observedMessage = transition.observedMessage)
         }
-        loadHistory(selected, observedMessage = event.message)
-      }
-      "aborted", "error" -> {
-        cancelLoad()
-        mutableState.update { it.copy(streamText = null, activeRunId = null) }
-        loadHistory(selected)
       }
       else ->
         event.message?.let { message ->

apps/android/wear/src/test/java/ai/openclaw/wear/WearSessionScopeTest.kt

@@ -1,7 +1,9 @@
 package ai.openclaw.wear
 
 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 WearSessionScopeTest {
@@ -204,4 +206,247 @@ class WearSessionScopeTest {
     assertEquals("openai/model-0", switched.sessions.single().modelRef)
     assertEquals(emptyList<WearModel>(), switched.models)
   }
+
+  @Test
+  fun olderRunFinalMergesItsMessageWithoutEndingTheActiveReply() {
+    val previous = WearChatMessage("previous", "assistant", "Earlier", 1)
+    val completed = WearChatMessage("older", "assistant", "Finished older reply", 2)
+    val current = activeTerminalState(messages = listOf(previous))
+
+    val transition =
+      reduceWearTerminalChatEvent(
+        current,
+        terminalEvent(state = "final", runId = "older-run", message = completed),
+      )
+
+    assertEquals(listOf(previous, completed), transition.state.messages)
+    assertEquals("active-run", transition.state.activeRunId)
+    assertEquals("Hello", transition.state.streamText)
+    assertFalse(transition.reloadHistory)
+    assertNull(transition.observedMessage)
+  }
+
+  @Test
+  fun olderRunAbortCannotEndTheActiveReply() {
+    assertForeignTerminalPreservesLiveReply("aborted")
+  }
+
+  @Test
+  fun olderRunErrorCannotEndTheActiveReply() {
+    assertForeignTerminalPreservesLiveReply("error")
+  }
+
+  @Test
+  fun identifiedOlderFinalCannotEndAnAnonymousReply() {
+    assertForeignTerminalPreservesLiveReply("final", activeRunId = null)
+  }
+
+  @Test
+  fun identifiedOlderAbortCannotEndAnAnonymousReply() {
+    assertForeignTerminalPreservesLiveReply("aborted", activeRunId = null)
+  }
+
+  @Test
+  fun identifiedOlderErrorCannotEndAnAnonymousReply() {
+    assertForeignTerminalPreservesLiveReply("error", activeRunId = null)
+  }
+
+  @Test
+  fun anonymousReplyFinalReconcilesWhenTheRunIsFirstIdentified() {
+    val completed = WearChatMessage("completed", "assistant", "Finished reply", 2)
+    val current = activeTerminalState(activeRunId = null)
+    val transition =
+      reduceWearTerminalChatEvent(
+        current,
+        terminalEvent(state = "final", runId = "revealed-run", message = completed),
+      )
+
+    assertEquals(listOf(completed), transition.state.messages)
+    assertEquals("Hello", transition.state.streamText)
+    assertNull(transition.state.activeRunId)
+    assertTrue(transition.reloadHistory)
+    assertEquals(completed, transition.observedMessage)
+  }
+
+  @Test
+  fun anonymousReplyAbortReconcilesWhenTheRunIsFirstIdentified() {
+    assertUncertainTerminalPreservesReplyAndReloadsHistory(
+      state = "aborted",
+      activeRunId = null,
+      eventRunId = "revealed-run",
+    )
+  }
+
+  @Test
+  fun anonymousReplyErrorReconcilesWhenTheRunIsFirstIdentified() {
+    assertUncertainTerminalPreservesReplyAndReloadsHistory(
+      state = "error",
+      activeRunId = null,
+      eventRunId = "revealed-run",
+    )
+  }
+
+  @Test
+  fun matchingRunFinalEndsTheReplyAndReloadsItsFinalMessage() {
+    val completed = WearChatMessage("completed", "assistant", "Finished reply", 2)
+    val transition =
+      reduceWearTerminalChatEvent(
+        activeTerminalState(),
+        terminalEvent(state = "final", runId = "active-run", message = completed),
+      )
+
+    assertEquals(listOf(completed), transition.state.messages)
+    assertNull(transition.state.activeRunId)
+    assertNull(transition.state.streamText)
+    assertTrue(transition.reloadHistory)
+    assertEquals(completed, transition.observedMessage)
+  }
+
+  @Test
+  fun matchingRunAbortEndsTheReplyAndReloadsHistory() {
+    assertOwnTerminalEndsLiveReply("aborted", runId = "active-run")
+  }
+
+  @Test
+  fun matchingRunErrorEndsTheReplyAndReloadsHistory() {
+    assertOwnTerminalEndsLiveReply("error", runId = "active-run")
+  }
+
+  @Test
+  fun unidentifiedAbortReconcilesWithoutClearingAnIdentifiedReply() {
+    assertUncertainTerminalPreservesReplyAndReloadsHistory(
+      state = "aborted",
+      activeRunId = "active-run",
+      eventRunId = null,
+    )
+  }
+
+  @Test
+  fun unidentifiedErrorReconcilesWithoutClearingAnIdentifiedReply() {
+    assertUncertainTerminalPreservesReplyAndReloadsHistory(
+      state = "error",
+      activeRunId = "active-run",
+      eventRunId = null,
+    )
+  }
+
+  @Test
+  fun unidentifiedFinalMergesWithoutClearingAnIdentifiedReply() {
+    val completed = WearChatMessage("completed", "assistant", "Finished reply", 2)
+    val transition =
+      reduceWearTerminalChatEvent(
+        activeTerminalState(),
+        terminalEvent(state = "final", runId = null, message = completed),
+      )
+
+    assertEquals(listOf(completed), transition.state.messages)
+    assertEquals("active-run", transition.state.activeRunId)
+    assertEquals("Hello", transition.state.streamText)
+    assertTrue(transition.reloadHistory)
+    assertEquals(completed, transition.observedMessage)
+  }
+
+  @Test
+  fun otherSessionTerminalNeverChangesTheSelectedReply() {
+    val current = activeTerminalState()
+    val transition =
+      reduceWearTerminalChatEvent(
+        current,
+        terminalEvent(state = "final", runId = "active-run", sessionKey = "agent:other"),
+      )
+
+    assertEquals(current, transition.state)
+    assertFalse(transition.reloadHistory)
+    assertNull(transition.observedMessage)
+  }
+
+  private fun assertUncertainTerminalPreservesReplyAndReloadsHistory(
+    state: String,
+    activeRunId: String?,
+    eventRunId: String?,
+  ) {
+    val current = activeTerminalState(activeRunId = activeRunId)
+    val transition =
+      reduceWearTerminalChatEvent(
+        current,
+        terminalEvent(state = state, runId = eventRunId),
+      )
+
+    assertEquals(current, transition.state)
+    assertTrue(transition.reloadHistory)
+    assertNull(transition.observedMessage)
+  }
+
+  private fun assertForeignTerminalPreservesLiveReply(
+    state: String,
+    activeRunId: String? = "active-run",
+  ) {
+    val current = activeTerminalState(activeRunId = activeRunId)
+    val transition =
+      reduceWearTerminalChatEvent(
+        current,
+        terminalEvent(state = state, runId = "older-run"),
+      )
+
+    assertEquals(current, transition.state)
+    if (activeRunId == null) {
+      assertTrue(transition.reloadHistory)
+    } else {
+      assertFalse(transition.reloadHistory)
+    }
+    assertNull(transition.observedMessage)
+  }
+
+  private fun assertOwnTerminalEndsLiveReply(
+    state: String,
+    runId: String?,
+  ) {
+    val transition =
+      reduceWearTerminalChatEvent(
+        activeTerminalState(),
+        terminalEvent(state = state, runId = runId),
+      )
+
+    assertNull(transition.state.activeRunId)
+    assertNull(transition.state.streamText)
+    assertTrue(transition.reloadHistory)
+    assertNull(transition.observedMessage)
+  }
+
+  private fun activeTerminalState(
+    activeRunId: String? = "active-run",
+    messages: List<WearChatMessage> = emptyList(),
+  ): WearUiState {
+    val selected =
+      WearSession(
+        key = "agent:main",
+        title = "Main",
+        updatedAt = null,
+        hasActiveRun = true,
+        phoneNodeId = "phone-a",
+      )
+    return WearUiState(
+      selectedSession = selected,
+      messages = messages,
+      streamText = "Hello",
+      activeRunId = activeRunId,
+    )
+  }
+
+  private fun terminalEvent(
+    state: String,
+    runId: String?,
+    sessionKey: String = "agent:main",
+    message: WearChatMessage? = null,
+  ): WearChatEvent =
+    WearChatEvent(
+      sessionKey = sessionKey,
+      runId = runId,
+      state = state,
+      deltaText = null,
+      replace = false,
+      streamText = null,
+      streamTextComplete = false,
+      message = message,
+    )
 }