From 935ee7fedfb0ef8840972fc336db566450abc47e Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 25 Jul 2026 12:42:42 -0700 Subject: [PATCH] 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 --- .../ai/openclaw/app/wear/WearProxyBridge.kt | 18 +- .../openclaw/app/wear/WearProxyBridgeTest.kt | 108 ++++++++ .../java/ai/openclaw/wear/WearViewModel.kt | 75 +++++- .../ai/openclaw/wear/WearSessionScopeTest.kt | 245 ++++++++++++++++++ 4 files changed, 427 insertions(+), 19 deletions(-) diff --git a/apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyBridge.kt b/apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyBridge.kt index 0004920fde8e..b8fbd9271427 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyBridge.kt +++ b/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) } diff --git a/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt index 1d9bb5688e9f..ee02be725b51 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt +++ b/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) } diff --git a/apps/android/wear/src/main/java/ai/openclaw/wear/WearViewModel.kt b/apps/android/wear/src/main/java/ai/openclaw/wear/WearViewModel.kt index a14d67d7fbc7..e2c8ed3b490d 100644 --- a/apps/android/wear/src/main/java/ai/openclaw/wear/WearViewModel.kt +++ b/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 -> diff --git a/apps/android/wear/src/test/java/ai/openclaw/wear/WearSessionScopeTest.kt b/apps/android/wear/src/test/java/ai/openclaw/wear/WearSessionScopeTest.kt index fcad7b69262a..4a65e0310dee 100644 --- a/apps/android/wear/src/test/java/ai/openclaw/wear/WearSessionScopeTest.kt +++ b/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(), 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 = 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, + ) }