From e37e63ceb6375485cbc36d9eb52b300f149086ee Mon Sep 17 00:00:00 2001 From: WhatsSkiLL Date: Fri, 14 Aug 2026 18:22:26 +0200 Subject: [PATCH] feat(android): add Wear Agent Pulse (#122123) * feat(android): add Wear agent pulse * fix-android-wear-pulse-selected-session * fix(android): count direct Wear Pulse subagents * fix(android): gate Wear Pulse page by capability * fix:bound-wear-pulse-gateway-reads * fix: keep native locale artifacts generated --------- Co-authored-by: IWhatsskill <284122573+IWhatsskill@users.noreply.github.com> Co-authored-by: Colin Johnson Co-authored-by: Solvely-Colin <159171061+Solvely-Colin@users.noreply.github.com> Co-authored-by: Colin Johnson <211764741+Solvely-Colin@users.noreply.github.com> --- apps/.i18n/native-source.json | 191 +++++++++ .../main/java/ai/openclaw/app/NodeRuntime.kt | 115 +++++- .../ai/openclaw/app/chat/ChatController.kt | 114 ++++++ .../ai/openclaw/app/chat/ChatSwarmProgress.kt | 8 + .../app/wear/WearAgentPulseProjection.kt | 129 ++++++ .../openclaw/app/wear/WearProxyController.kt | 10 + .../chat/ChatControllerSwarmProgressTest.kt | 324 +++++++++++++++ .../app/chat/ChatSwarmProgressTest.kt | 21 + .../app/wear/WearAgentPulseProjectionTest.kt | 197 ++++++++++ .../openclaw/app/wear/WearProxyBridgeTest.kt | 57 ++- .../app/wear/WearProxyControllerTest.kt | 47 +++ .../ai/openclaw/wear/shared/WearProtocol.kt | 4 + .../openclaw/wear/shared/WearProtocolTest.kt | 2 + apps/android/wear/BEHAVIOR.md | 1 + .../java/ai/openclaw/wear/MainActivity.kt | 2 + .../ai/openclaw/wear/WearCompanionUiModels.kt | 11 + .../ai/openclaw/wear/WearGatewayRepository.kt | 168 ++++++++ .../java/ai/openclaw/wear/WearProxyClient.kt | 33 +- .../main/java/ai/openclaw/wear/WearScreens.kt | 369 ++++++++++++++++-- .../java/ai/openclaw/wear/WearViewModel.kt | 255 ++++++++++++ .../wear/src/main/res/values/strings.xml | 26 ++ .../java/ai/openclaw/wear/MainActivityTest.kt | 87 +++++ .../wear/WearGatewayRepositoryTest.kt | 125 +++++- .../ai/openclaw/wear/WearLaunchIntentTest.kt | 20 + .../ai/openclaw/wear/WearProxyClientTest.kt | 20 + .../wear/WearViewModelLifecycleTest.kt | 126 ++++++ 26 files changed, 2422 insertions(+), 40 deletions(-) create mode 100644 apps/android/app/src/main/java/ai/openclaw/app/wear/WearAgentPulseProjection.kt create mode 100644 apps/android/app/src/test/java/ai/openclaw/app/wear/WearAgentPulseProjectionTest.kt diff --git a/apps/.i18n/native-source.json b/apps/.i18n/native-source.json index bc859f5bcde3..b9b50c36c726 100644 --- a/apps/.i18n/native-source.json +++ b/apps/.i18n/native-source.json @@ -1415,6 +1415,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/SettingsScreens.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -1440,6 +1444,17 @@ } ] }, + { + "id": "native.android.ccd226a2c0e85242", + "source": "Active task list is at its limit", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.28e1dcd489f67784", "source": "Adaptive", @@ -2325,6 +2340,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/VoiceScreen.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -2707,6 +2726,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatScreen.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -2787,6 +2810,17 @@ } ] }, + { + "id": "native.android.7e716e5025c96755", + "source": "Bounded active and recent snapshot", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.b62482f58e031dc3", "source": "Bridge", @@ -3771,6 +3805,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/BackgroundTasksSheet.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -5991,6 +6029,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatMessageActions.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -6467,6 +6509,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatMessageViews.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -7495,6 +7541,17 @@ } ] }, + { + "id": "native.android.cee2b043b9ff723d", + "source": "Groups", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.4f316f3050dd4439", "source": "Have a terminal open on the device running OpenClaw.", @@ -7718,6 +7775,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/ShellScreen.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -8462,6 +8523,17 @@ } ] }, + { + "id": "native.android.982d8645f94585d2", + "source": "Loading Pulse", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.0ba4fafa3c405a73", "source": "Loading automation…", @@ -9083,6 +9155,17 @@ } ] }, + { + "id": "native.android.e238a849f3745e16", + "source": "More phases not shown", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.a5beb359c71ff68c", "source": "Motion", @@ -12039,6 +12122,17 @@ } ] }, + { + "id": "native.android.4c0349bb7fb1300c", + "source": "Pending requests", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.95992be09ebae8eb", "source": "Permission needed", @@ -12083,6 +12177,17 @@ } ] }, + { + "id": "native.android.b90b0acb54a1e930", + "source": "Phase %1$d", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.008ac966ec9191f9", "source": "Phone Capabilities", @@ -12658,6 +12763,17 @@ } ] }, + { + "id": "native.android.139b5e6557177a23", + "source": "Pulse", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.c0fef45b3b102aac", "source": "QR code did not contain a valid setup code.", @@ -12732,6 +12848,21 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/BackgroundTasksSheet.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, + { + "id": "native.android.139768629b9f2483", + "source": "Queued %1$d, Running %2$d, Done %3$d, Failed %4$d, Hidden %5$d", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -12922,6 +13053,17 @@ } ] }, + { + "id": "native.android.e8bfe61eb7a178e9", + "source": "Reading bounded aggregates from the paired phone.", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.dc842ee2019a7fda", "source": "Ready", @@ -13143,6 +13285,17 @@ } ] }, + { + "id": "native.android.f26643818268e147", + "source": "Recent task list is at its limit", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.cfd90a828c5f6bda", "source": "Recents", @@ -13425,6 +13578,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/SkillsSettingsScreen.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -14070,6 +14227,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/BackgroundTasksSheet.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -15032,6 +15193,17 @@ } ] }, + { + "id": "native.android.076e5ee0a1e35da3", + "source": "Session swarm", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.84424d26d15b7f68", "source": "Session target", @@ -16084,6 +16256,17 @@ } ] }, + { + "id": "native.android.245f1127ee17dae9", + "source": "Status: %1$s", + "surface": "android", + "sites": [ + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" + } + ] + }, { "id": "native.android.c5aca22f498a5dda", "source": "Step 1", @@ -17658,6 +17841,10 @@ { "kind": "ui-call", "path": "apps/android/app/src/main/java/ai/openclaw/app/ui/chat/ChatQuestionCard.kt" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, @@ -17715,6 +17902,10 @@ { "kind": "resource-string", "path": "apps/android/app/src/main/res/values/strings.xml" + }, + { + "kind": "resource-string", + "path": "apps/android/wear/src/main/res/values/strings.xml" } ] }, diff --git a/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt b/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt index 42d0f0659283..46b1344a038e 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt @@ -122,6 +122,7 @@ import ai.openclaw.app.wear.WearProxyGatewayException import ai.openclaw.app.wear.WearProxyModel import ai.openclaw.app.wear.WearRealtimeAttemptOwner import ai.openclaw.app.wear.WearRealtimeTalkController +import ai.openclaw.app.wear.projectWearAgentPulse import ai.openclaw.app.wear.wearConnectionFailure import ai.openclaw.wear.shared.WearMessage import ai.openclaw.wear.shared.WearRealtimeTalkCodec @@ -141,6 +142,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay @@ -159,6 +161,7 @@ import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.withTimeoutOrNull import kotlinx.serialization.Serializable import kotlinx.serialization.json.Json import kotlinx.serialization.json.JsonArray @@ -184,6 +187,39 @@ private const val CRON_JOBS_SNAPSHOT_MAX_ATTEMPTS = 3 private const val OperatorAdminScope = "operator.admin" private const val OperatorPairingScope = "operator.pairing" +internal const val WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS = 8_000L + +internal data class WearAgentPulseReads( + val tasks: Tasks?, + val swarm: Swarm?, +) + +internal suspend fun readWearAgentPulseConcurrently( + readTasks: suspend () -> Tasks, + readSwarm: suspend () -> Swarm, + budgetMillis: Long = WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS, +): WearAgentPulseReads = + coroutineScope { + val tasks = async { readWearAgentPulseComponent(budgetMillis, readTasks) } + val swarm = async { readWearAgentPulseComponent(budgetMillis, readSwarm) } + WearAgentPulseReads( + tasks = tasks.await(), + swarm = swarm.await(), + ) + } + +private suspend fun readWearAgentPulseComponent( + budgetMillis: Long, + read: suspend () -> T, +): T? = + try { + withTimeoutOrNull(budgetMillis) { read() } + } catch (err: CancellationException) { + throw err + } catch (_: Throwable) { + null + } + private fun execApprovalOutcomeUnknownMessage(): String = nativeText("Resolution outcome unknown. Actions stay disabled until the Gateway record is verified.").source private fun execApprovalStillPendingMessage(): String = nativeText("The Gateway still shows this approval as pending. Review it before trying again.").source @@ -1310,6 +1346,7 @@ class NodeRuntime private constructor( val execApprovalsNotice: StateFlow = _execApprovalsNotice.asStateFlow() private val execApprovalsRefreshSeq = AtomicLong(0) private val execApprovalsStateLock = Any() + private var execApprovalsSnapshotReady = false private val resolvedExecApprovalIds = Collections.newSetFromMap(ConcurrentHashMap()) private val pendingExecApprovalWrites = mutableMapOf() @@ -1518,9 +1555,7 @@ class NodeRuntime private constructor( isGatewayConnected = operatorSession::isReady, gatewayStatusText = { synchronized(gatewayStatusLock) { operatorStatusText } }, hasOperatorAdminScope = { OperatorAdminScope in _operatorScopes.value }, - activeAgentId = { - resolveAgentIdFromMainSessionKey(mainSessionKey.value) ?: gatewayDefaultAgentId.value - }, + activeAgentId = ::currentWearAgentId, activeSessionKey = { chatSessionKey.value }, selectedModelRef = { chatSelectedModelRef.value }, agents = { @@ -1560,6 +1595,7 @@ class NodeRuntime private constructor( }, connectGateway = { refreshGatewayConnection() }, disconnectGateway = { disconnect() }, + loadAgentPulse = ::loadWearAgentPulse, startRealtimeTalk = { nodeId, sessionKey, attemptId, language, attemptScopedAudio -> if (startWearRealtimeTalk(nodeId, sessionKey, attemptId, language, attemptScopedAudio)) wearRealtimeTalkSnapshot.value else null }, @@ -1569,6 +1605,67 @@ class NodeRuntime private constructor( ) } + private fun currentWearAgentId(): String? = resolveAgentIdFromMainSessionKey(mainSessionKey.value) ?: gatewayDefaultAgentId.value + + private suspend fun loadWearAgentPulse(requestedSessionKey: String?): JsonObject { + val gatewayScope = captureGatewayDataScope() + val agentId = currentWearAgentId() + val connected = gatewayScope != null && operatorSession.isReady() + val reads = + if (connected && agentId != null) { + readWearAgentPulseConcurrently( + readTasks = { listBackgroundTasks(agentId) }, + readSwarm = { + requestedSessionKey?.let { sessionKey -> + chat.readSwarmSnapshotFor(sessionKey, agentId) + } + }, + ) + } else { + null + } + val tasks = reads?.tasks + val swarmSnapshot = reads?.swarm + // Capture every projection input before the final route check so a route + // change cannot mix a current task result with later-route aggregates. + val approvals = currentWearAgentPulseApprovals() + val routeStillCurrent = + gatewayScope?.let { capturedScope -> + connected && + isGatewayDataScopeCurrent(capturedScope) && + operatorSession.isReady() && + currentWearAgentId() == agentId + } == true + val swarmAvailable = + routeStillCurrent && + requestedSessionKey != null && + swarmSnapshot?.isAvailableFor(requestedSessionKey) == true + return projectWearAgentPulse( + gatewayConnected = routeStillCurrent, + tasks = tasks.takeIf { routeStillCurrent }, + swarmAvailable = swarmAvailable, + swarmGroups = if (swarmAvailable) swarmSnapshot.groups else emptyList(), + pendingApprovalCount = approvals.pendingCount, + approvalsAvailable = routeStillCurrent && approvals.available, + approvalsRefreshing = approvals.refreshing, + ) + } + + private fun currentWearAgentPulseApprovals(): WearAgentPulseApprovalSnapshot = + synchronized(execApprovalsStateLock) { + WearAgentPulseApprovalSnapshot( + pendingCount = _execApprovals.value.size, + available = execApprovalsSnapshotReady && _execApprovalsErrorText.value == null, + refreshing = _execApprovalsRefreshing.value, + ) + } + + private data class WearAgentPulseApprovalSnapshot( + val pendingCount: Int, + val available: Boolean, + val refreshing: Boolean, + ) + internal suspend fun handleWearProxyRequest( sourceNodeId: String, request: WearMessage.Request, @@ -1668,8 +1765,11 @@ class NodeRuntime private constructor( } invalidateExecApprovalRefreshes() resolvedExecApprovalIds.clear() - if (retirePendingCronRuns) { - synchronized(execApprovalsStateLock) { pendingExecApprovalWrites.clear() } + synchronized(execApprovalsStateLock) { + execApprovalsSnapshotReady = false + if (retirePendingCronRuns) { + pendingExecApprovalWrites.clear() + } } _execApprovals.value = emptyList() _execApprovalsRefreshing.value = false @@ -7056,7 +7156,9 @@ class NodeRuntime private constructor( val gatewayScope = captureGatewayDataScope() ?: return val refreshGeneration = synchronized(execApprovalsStateLock) { - execApprovalsRefreshSeq.incrementAndGet() + val nextGeneration = execApprovalsRefreshSeq.incrementAndGet() + execApprovalsSnapshotReady = false + nextGeneration } publishGatewayData(gatewayScope) { _execApprovalsRefreshing.value = true @@ -7623,6 +7725,7 @@ class NodeRuntime private constructor( resolvedExecApprovalIds.addAll(terminalIds) terminalIds.forEach(pendingExecApprovalWrites::remove) val nextRows = rows.filterNot { it.id in resolvedExecApprovalIds }.filterActiveExecApprovals() + execApprovalsSnapshotReady = true _execApprovals.value = nextRows scheduleExecApprovalExpiryPrune(nextRows) } diff --git a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt index fdc3fe207019..94dc60d005c2 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt @@ -58,6 +58,9 @@ import java.util.concurrent.atomic.AtomicLong internal const val SESSION_LIST_FETCH_LIMIT = 200 private val QUESTION_REFRESH_RETRY_DELAYS_MS = longArrayOf(1_000L, 2_000L, 4_000L) private val SWARM_REFRESH_RETRY_DELAYS_MS = longArrayOf(1_000L, 2_000L, 4_000L) +private const val WEAR_AGENT_PULSE_SWARM_MAX_ROWS = 1_000 +private const val WEAR_AGENT_PULSE_SWARM_FETCH_LIMIT = WEAR_AGENT_PULSE_SWARM_MAX_ROWS + 1 +private const val WEAR_AGENT_PULSE_DIRECT_CHILDREN_GROUP = "__wear_agent_pulse_direct_children__" private const val SUBAGENT_ACTIVITY_RETENTION_MS = 60_000L private const val SESSION_EDITOR_MAX_BASE64_CHARS = ((OUTBOX_MAX_COMMAND_ATTACHMENT_BYTES + 2) / 3) * 4 private val MANAGED_MEDIA_PATH_REGEX = @@ -385,6 +388,116 @@ class ChatController internal constructor( private var swarmSessionKey: String? = null private var swarmEnabled = false + internal fun currentSwarmSnapshot(): ChatSwarmSnapshot = + synchronized(swarmLock) { + ChatSwarmSnapshot( + sessionKey = swarmSessionKey, + enabled = swarmEnabled, + groups = _swarmGroups.value, + ) + } + + /** Reads Swarm state for a Wear-selected session without changing the Phone chat selection. */ + internal suspend fun readSwarmSnapshotFor( + sessionKey: String, + agentId: String, + ): ChatSwarmSnapshot? { + val requestedSessionKey = sessionKey.trim().takeIf(String::isNotEmpty) ?: return null + val requestedAgentId = agentId.trim().takeIf(String::isNotEmpty) ?: return null + val scopedSessionKey = normalizeRequestedSessionKey(requestedSessionKey) + val scopedAgentId = resolveAgentIdFromMainSessionKey(scopedSessionKey) + val restrictToParentAgent = + when { + scopedAgentId == requestedAgentId -> false + requestedSessionKey == "main" && scopedSessionKey == "main" && scopedAgentId == null -> true + else -> return null + } + val requestCacheScope = currentCacheScope() ?: return null + val enabled = + try { + val params = + buildJsonObject { + put("agentId", JsonPrimitive(requestedAgentId)) + } + val root = + json + .parseToJsonElement( + requestGatewayBound( + requestCacheScope.gatewayId, + "chat.metadata", + params.toString(), + ), + ).asObjectOrNull() + root?.get("swarmEnabled").asBooleanOrNull() == true + } catch (err: CancellationException) { + throw err + } catch (_: Throwable) { + return null + } + if (!enabled) { + return ChatSwarmSnapshot( + sessionKey = requestedSessionKey, + enabled = false, + groups = emptyList(), + ) + } + val rows = + try { + val params = + buildJsonObject { + put("includeGlobal", JsonPrimitive(false)) + put("includeUnknown", JsonPrimitive(false)) + put("configuredAgentsOnly", JsonPrimitive(true)) + if (restrictToParentAgent) { + put("agentId", JsonPrimitive(requestedAgentId)) + } + put("spawnedBy", JsonPrimitive(scopedSessionKey)) + put("limit", JsonPrimitive(WEAR_AGENT_PULSE_SWARM_FETCH_LIMIT)) + put("offset", JsonPrimitive(0)) + } + val root = + json + .parseToJsonElement( + requestGatewayBound( + requestCacheScope.gatewayId, + "sessions.list", + params.toString(), + ), + ).asObjectOrNull() + ?: return null + val sessionElements = root["sessions"] as? JsonArray ?: return null + val parsedRows = sessionElements.mapNotNull { parseSessionEntry(it.asObjectOrNull()) } + val truncated = + root["hasMore"].asBooleanOrNull() == true || + root["totalCount"].asLongOrNull()?.let { total -> total > sessionElements.size.toLong() } == true || + sessionElements.size > WEAR_AGENT_PULSE_SWARM_MAX_ROWS + if (truncated || parsedRows.size != sessionElements.size) { + return null + } + parsedRows + } catch (err: CancellationException) { + throw err + } catch (_: Throwable) { + return null + } + val projectedRows = + rows.map { row -> + val hasExplicitGroup = !row.swarmGroupId.isNullOrBlank() + val hasSubagentProvenance = row.subagentRunState != null || row.hasActiveSubagentRun != null + if (!hasExplicitGroup && hasSubagentProvenance) { + row.copy(swarmGroupId = WEAR_AGENT_PULSE_DIRECT_CHILDREN_GROUP) + } else { + row + } + } + if (requestCacheScope != currentCacheScope()) return null + return ChatSwarmSnapshot( + sessionKey = requestedSessionKey, + enabled = true, + groups = buildChatSwarmGroups(projectedRows) { candidate -> sameOutboxSession(candidate, scopedSessionKey) }, + ) + } + private val _sessionBranches = MutableStateFlow>(emptyList()) val sessionBranches: StateFlow> = _sessionBranches.asStateFlow() @@ -4238,6 +4351,7 @@ class ChatController internal constructor( ) { return } + swarmSessionKey = snapshot.parentKey _swarmGroups.value = groups } } diff --git a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatSwarmProgress.kt b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatSwarmProgress.kt index 832df53d0a8e..5f03b823076c 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatSwarmProgress.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatSwarmProgress.kt @@ -43,6 +43,14 @@ data class ChatSwarmGroup( val phases: List, ) +internal data class ChatSwarmSnapshot( + val sessionKey: String?, + val enabled: Boolean, + val groups: List, +) { + fun isAvailableFor(sessionKey: String?): Boolean = enabled && sessionKey != null && this.sessionKey == sessionKey +} + internal data class ChatSwarmSessionPage( val sessions: List, val totalCount: Int?, diff --git a/apps/android/app/src/main/java/ai/openclaw/app/wear/WearAgentPulseProjection.kt b/apps/android/app/src/main/java/ai/openclaw/app/wear/WearAgentPulseProjection.kt new file mode 100644 index 000000000000..1675c7b898df --- /dev/null +++ b/apps/android/app/src/main/java/ai/openclaw/app/wear/WearAgentPulseProjection.kt @@ -0,0 +1,129 @@ +package ai.openclaw.app.wear + +import ai.openclaw.app.chat.BackgroundTask +import ai.openclaw.app.chat.ChatSwarmDotStatus +import ai.openclaw.app.chat.ChatSwarmGroup +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.buildJsonArray +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put + +internal fun projectWearAgentPulse( + gatewayConnected: Boolean, + tasks: List?, + swarmAvailable: Boolean, + swarmGroups: List, + pendingApprovalCount: Int, + approvalsAvailable: Boolean, + approvalsRefreshing: Boolean, +): JsonObject = + buildJsonObject { + put( + "tasks", + buildJsonObject { + if (!gatewayConnected || tasks == null) { + put("state", "unavailable") + } else { + val queued = tasks.count { task -> task.status == "queued" } + val running = tasks.count { task -> task.status == "running" } + val completed = tasks.count { task -> task.status == "completed" } + val failed = + tasks.count { task -> + task.status == "failed" || task.status == "cancelled" || task.status == "timed_out" + } + put("state", "ready") + put("scope", "bounded") + put("queued", queued) + put("running", running) + put("completed", completed) + put("failed", failed) + put("activeAtLimit", queued + running >= ACTIVE_TASK_LIMIT) + put("recentAtLimit", completed + failed >= RECENT_TASK_LIMIT) + } + }, + ) + put( + "swarm", + buildJsonObject { + when { + !gatewayConnected || !swarmAvailable -> put("state", "unavailable") + swarmGroups.isEmpty() -> { + put("state", "idle") + put("scope", "selected-session") + } + else -> { + val phaseBuckets = MutableList(MAX_PHASE_BUCKETS) { MutablePhaseCounts() } + var morePhases = false + swarmGroups.forEach { group -> + if (group.phases.size > MAX_PHASE_BUCKETS) morePhases = true + group.phases.take(MAX_PHASE_BUCKETS).forEachIndexed { index, phase -> + val bucket = phaseBuckets[index] + phase.dots.forEach { dot -> + when (dot.status) { + ChatSwarmDotStatus.Queued -> bucket.queued += 1 + ChatSwarmDotStatus.Running -> bucket.running += 1 + ChatSwarmDotStatus.Done -> bucket.done += 1 + ChatSwarmDotStatus.Failed -> bucket.failed += 1 + } + } + bucket.hidden += phase.hidden + } + } + put("state", "active") + put("scope", "selected-session") + put("groups", swarmGroups.size) + put("running", swarmGroups.sumOf(ChatSwarmGroup::running)) + put("done", swarmGroups.sumOf(ChatSwarmGroup::done)) + put("failed", swarmGroups.sumOf(ChatSwarmGroup::failed)) + put( + "phases", + buildJsonArray { + phaseBuckets + .dropLastWhile { phase -> !phase.hasData() } + .forEach { phase -> + add( + buildJsonObject { + put("queued", phase.queued) + put("running", phase.running) + put("done", phase.done) + put("failed", phase.failed) + put("hidden", phase.hidden) + }, + ) + } + }, + ) + put("morePhases", morePhases) + } + } + }, + ) + put( + "approvals", + buildJsonObject { + when { + !gatewayConnected -> put("state", "unavailable") + approvalsRefreshing -> put("state", "refreshing") + !approvalsAvailable -> put("state", "unavailable") + else -> { + put("state", "ready") + put("pending", pendingApprovalCount.coerceAtLeast(0)) + } + } + }, + ) + } + +private data class MutablePhaseCounts( + var queued: Int = 0, + var running: Int = 0, + var done: Int = 0, + var failed: Int = 0, + var hidden: Int = 0, +) { + fun hasData(): Boolean = queued != 0 || running != 0 || done != 0 || failed != 0 || hidden != 0 +} + +private const val ACTIVE_TASK_LIMIT = 100 +private const val RECENT_TASK_LIMIT = 50 +private const val MAX_PHASE_BUCKETS = 8 diff --git a/apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyController.kt b/apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyController.kt index dae3532313fc..dfd44ef55648 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyController.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/wear/WearProxyController.kt @@ -51,6 +51,9 @@ internal class WearProxyController( private val selectSessionModel: suspend (sessionKey: String, modelRef: String) -> Boolean = { _, _ -> false }, private val connectGateway: suspend () -> Unit = {}, private val disconnectGateway: suspend () -> Unit = {}, + private val loadAgentPulse: suspend (sessionKey: String?) -> JsonObject = { + throw WearProxyGatewayException("unavailable", "Agent Pulse is unavailable") + }, private val startRealtimeTalk: suspend (nodeId: String, sessionKey: String, attemptId: String, language: String?, attemptScopedAudio: Boolean) -> WearRealtimeTalkSnapshot? = { _, _, _, _, _ -> null }, private val stopRealtimeTalk: suspend (nodeId: String, attemptId: String) -> WearRealtimeTalkSnapshot? = { _, _ -> null }, @@ -63,6 +66,7 @@ internal class WearProxyController( val result = when (request.method) { WearRpcMethod.ProxyStatus -> proxyStatus(request.params) + WearRpcMethod.AgentPulse -> agentPulse(request.params) WearRpcMethod.SessionsList -> listSessions(request.params) WearRpcMethod.AgentsList -> listAgents(request.params) WearRpcMethod.AgentsSelect -> selectAgent(request.params) @@ -87,6 +91,12 @@ internal class WearProxyController( failure(request.requestId, code = "unavailable", message = "Phone gateway request failed") } + private suspend fun agentPulse(params: JsonObject): JsonObject { + params.requireOnly("sessionKey") + val sessionKey = params.optionalStringParam("sessionKey", MAX_SESSION_KEY_CHARS) + return loadAgentPulse(sessionKey) + } + private suspend fun talkStart( sourceNodeId: String, params: JsonObject, diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerSwarmProgressTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerSwarmProgressTest.kt index 895cf1009412..01a51025ea41 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerSwarmProgressTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerSwarmProgressTest.kt @@ -40,6 +40,330 @@ class ChatControllerSwarmProgressTest { assertTrue(controller.swarmGroups.value.isEmpty()) } + @Test + fun readsCrossAgentDirectChildrenForWearWithoutMutatingPhoneSwarmState() = + runTest { + val target = "agent:main:wear-b" + val requests = mutableListOf>() + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, params -> + requests += method to params + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":true}""" + "sessions.list" -> + """ + {"sessions":[ + { + "key":"agent:codex:subagent:gateway-watch", + "parentSessionKey":"$target", + "spawnedBy":"$target", + "status":"running", + "subagentRunState":"active", + "hasActiveSubagentRun":true + }, + { + "key":"agent:clock:subagent:clock-watch", + "parentSessionKey":"$target", + "spawnedBy":"$target", + "status":"running", + "subagentRunState":"active", + "hasActiveSubagentRun":true + }, + { + "key":"agent:load:subagent:load-watch", + "parentSessionKey":"$target", + "spawnedBy":"$target", + "status":"running", + "subagentRunState":"active", + "hasActiveSubagentRun":true + } + ], + "totalCount":3, + "hasMore":false + } + """.trimIndent() + else -> error("unexpected method $method") + } + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + val phoneSessionBefore = controller.sessionKey.value + val phoneSwarmBefore = controller.currentSwarmSnapshot() + + val snapshot = controller.readSwarmSnapshotFor(target, "main") + + assertTrue(snapshot?.isAvailableFor(target) == true) + assertEquals(3, snapshot?.groups?.single()?.running) + assertTrue( + requests.any { (method, params) -> + method == "chat.metadata" && params.orEmpty().contains("\"agentId\":\"main\"") + }, + ) + assertTrue( + requests.any { (method, params) -> + method == "sessions.list" && + params.orEmpty().contains("\"spawnedBy\":\"$target\"") && + !params.orEmpty().contains("\"agentId\"") + }, + ) + assertEquals(phoneSessionBefore, controller.sessionKey.value) + assertEquals(phoneSwarmBefore, controller.currentSwarmSnapshot()) + } + + @Test + fun keepsCollectGroupsSeparateFromDirectWearChildren() = + runTest { + val target = "agent:main:wear-b" + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, _ -> + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":true}""" + "sessions.list" -> + """ + {"sessions":[ + { + "key":"agent:writer:subagent:collect", + "spawnedBy":"$target", + "swarmGroupId":"swarm:$target:turn-1", + "status":"running" + }, + { + "key":"agent:clock:subagent:direct", + "spawnedBy":"$target", + "status":"running", + "subagentRunState":"active" + } + ],"totalCount":2,"hasMore":false} + """.trimIndent() + else -> error("unexpected method $method") + } + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + + val snapshot = controller.readSwarmSnapshotFor(target, "main") + + assertEquals(2, snapshot?.groups?.size) + assertEquals(2, snapshot?.groups?.sumOf { it.running }) + } + + @Test + fun acceptsGatewayFilteredWearRowsWhenNavigationParentOwnsTheChild() = + runTest { + val target = "agent:main:wear-b" + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, _ -> + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":true}""" + "sessions.list" -> + """ + {"sessions":[{ + "key":"agent:codex:subagent:foreign", + "spawnedBy":"agent:main:other", + "parentSessionKey":"$target", + "swarmGroupId":"swarm:$target:turn-1", + "status":"running" + }],"totalCount":1,"hasMore":false} + """.trimIndent() + else -> error("unexpected method $method") + } + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + + val snapshot = controller.readSwarmSnapshotFor(target, "main") + + assertTrue(snapshot?.isAvailableFor(target) == true) + assertEquals(1, snapshot?.groups?.single()?.running) + } + + @Test + fun ignoresUngroupedRowsWithoutSubagentProvenance() = + runTest { + val target = "agent:main:wear-b" + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, _ -> + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":true}""" + "sessions.list" -> + """ + {"sessions":[{ + "key":"agent:main:ordinary-session", + "spawnedBy":"$target", + "status":"running" + }],"totalCount":1,"hasMore":false} + """.trimIndent() + else -> error("unexpected method $method") + } + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + + val snapshot = controller.readSwarmSnapshotFor(target, "main") + + assertTrue(snapshot?.isAvailableFor(target) == true) + assertTrue(snapshot?.groups?.isEmpty() == true) + } + + @Test + fun readsEmptyWearSelectedSessionAsAvailableIdleWithoutMutatingPhoneState() = + runTest { + val target = "main" + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, params -> + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":true}""" + "sessions.list" -> { + assertTrue(params.orEmpty().contains("\"agentId\":\"main\"")) + assertTrue(params.orEmpty().contains("\"spawnedBy\":\"$target\"")) + """{"sessions":[],"totalCount":0,"hasMore":false}""" + } + else -> error("unexpected method $method") + } + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + val phoneSessionBefore = controller.sessionKey.value + val phoneSwarmBefore = controller.currentSwarmSnapshot() + + val snapshot = controller.readSwarmSnapshotFor(target, "main") + + assertTrue(snapshot?.isAvailableFor(target) == true) + assertTrue(snapshot?.groups?.isEmpty() == true) + assertEquals(phoneSessionBefore, controller.sessionKey.value) + assertEquals(phoneSwarmBefore, controller.currentSwarmSnapshot()) + } + + @Test + fun rejectsForeignAgentWearSessionBeforeAnyGatewayRead() = + runTest { + val methods = mutableListOf() + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, _ -> + methods += method + error("foreign session must fail before Gateway read") + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + + val snapshot = controller.readSwarmSnapshotFor("agent:other:foreign", "main") + + assertEquals(null, snapshot) + assertTrue(methods.isEmpty()) + } + + @Test + fun rejectsTruncatedWearSessionSwarmInsteadOfPublishingPartialCounts() = + runTest { + val target = "agent:main:wear-large" + var sessionsListCalls = 0 + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, params -> + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":true}""" + "sessions.list" -> { + sessionsListCalls += 1 + assertTrue(params.orEmpty().contains("\"limit\":1001")) + assertTrue(params.orEmpty().contains("\"offset\":0")) + """ + { + "sessions":[], + "totalCount":1001, + "hasMore":true, + "nextOffset":1001 + } + """.trimIndent() + } + else -> error("unexpected method $method") + } + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + + val snapshot = controller.readSwarmSnapshotFor(target, "main") + + assertEquals(null, snapshot) + assertEquals(1, sessionsListCalls) + } + + @Test + fun disabledWearSessionSwarmStaysUnavailableWithoutListingChildren() = + runTest { + var sessionsListCalls = 0 + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, _ -> + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":false}""" + "sessions.list" -> { + sessionsListCalls += 1 + error("disabled Swarm must not list children") + } + else -> error("unexpected method $method") + } + }, + cacheScope = { ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) }, + ) + + val snapshot = controller.readSwarmSnapshotFor("agent:main:wear-b", "main") + + assertEquals(false, snapshot?.enabled) + assertEquals(0, sessionsListCalls) + } + + @Test + fun discardsWearSessionSwarmWhenGatewayScopeChangesDuringRead() = + runTest { + var currentScope = ChatCacheScope(gatewayId = "gateway-a", connectionGeneration = 1) + val controller = + ChatController( + scope = this, + json = json, + requestGateway = { method, _ -> + when (method) { + "chat.metadata" -> """{"commands":[],"models":[],"swarmEnabled":true}""" + "sessions.list" -> { + currentScope = currentScope.copy(connectionGeneration = 2) + """{"sessions":[],"totalCount":0,"hasMore":false}""" + } + else -> error("unexpected method $method") + } + }, + cacheScope = { currentScope }, + ) + val phoneSessionBefore = controller.sessionKey.value + val phoneSwarmBefore = controller.currentSwarmSnapshot() + + val snapshot = controller.readSwarmSnapshotFor("agent:main:wear-b", "main") + + assertEquals(null, snapshot) + assertEquals(phoneSessionBefore, controller.sessionKey.value) + assertEquals(phoneSwarmBefore, controller.currentSwarmSnapshot()) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun swarmChildLifecycleStillUpdatesCanonicalSessionProjection() = diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatSwarmProgressTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatSwarmProgressTest.kt index 814181ae6f6f..79379b174925 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatSwarmProgressTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatSwarmProgressTest.kt @@ -3,6 +3,7 @@ package ai.openclaw.app.chat import kotlinx.serialization.json.JsonPrimitive import kotlinx.serialization.json.buildJsonObject import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse import org.junit.Assert.assertTrue import org.junit.Test @@ -226,6 +227,26 @@ class ChatSwarmProgressTest { assertTrue(phase.dots.any { it.status == ChatSwarmDotStatus.Running }) } + @Test + fun swarmSnapshotRequiresEnabledMatchingSessionOwner() { + val group = + ChatSwarmGroup( + groupId = "private-group", + label = "private-label", + running = 1, + done = 0, + failed = 0, + narrator = null, + phases = emptyList(), + ) + val snapshot = ChatSwarmSnapshot(sessionKey = "agent:main:a", enabled = true, groups = listOf(group)) + + assertTrue(snapshot.isAvailableFor("agent:main:a")) + assertFalse(snapshot.isAvailableFor("agent:main:b")) + assertFalse(snapshot.copy(enabled = false).isAvailableFor("agent:main:a")) + assertFalse(snapshot.copy(sessionKey = null).isAvailableFor("agent:main:a")) + } + private fun session( key: String, status: String?, diff --git a/apps/android/app/src/test/java/ai/openclaw/app/wear/WearAgentPulseProjectionTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/wear/WearAgentPulseProjectionTest.kt new file mode 100644 index 000000000000..8232eeeec672 --- /dev/null +++ b/apps/android/app/src/test/java/ai/openclaw/app/wear/WearAgentPulseProjectionTest.kt @@ -0,0 +1,197 @@ +package ai.openclaw.app.wear + +import ai.openclaw.app.WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS +import ai.openclaw.app.chat.BackgroundTask +import ai.openclaw.app.chat.ChatSwarmDot +import ai.openclaw.app.chat.ChatSwarmDotStatus +import ai.openclaw.app.chat.ChatSwarmGroup +import ai.openclaw.app.chat.ChatSwarmPhase +import ai.openclaw.app.readWearAgentPulseConcurrently +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.delay +import kotlinx.coroutines.test.currentTime +import kotlinx.coroutines.test.runTest +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.jsonObject +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test + +class WearAgentPulseProjectionTest { + @Test + fun projectsOnlyBoundedAggregateFields() { + val result = + projectWearAgentPulse( + gatewayConnected = true, + tasks = + listOf( + task("private-task-queued", "queued"), + task("private-task-running", "running"), + task("private-task-completed", "completed"), + task("private-task-cancelled", "cancelled"), + ), + swarmAvailable = true, + swarmGroups = + listOf( + ChatSwarmGroup( + groupId = "private-group", + label = "private-label", + running = 1, + done = 2, + failed = 3, + narrator = "private-narrator", + phases = + listOf( + ChatSwarmPhase( + key = "private-phase", + title = "private-title", + dots = + listOf( + dot("queued", ChatSwarmDotStatus.Queued), + dot("running", ChatSwarmDotStatus.Running), + dot("done", ChatSwarmDotStatus.Done), + dot("failed", ChatSwarmDotStatus.Failed), + ), + hidden = 7, + ), + ), + ), + ), + pendingApprovalCount = 2, + approvalsAvailable = true, + approvalsRefreshing = false, + ) + + assertEquals( + Json + .parseToJsonElement( + """{"tasks":{"state":"ready","scope":"bounded","queued":1,"running":1,"completed":1,"failed":1,"activeAtLimit":false,"recentAtLimit":false},"swarm":{"state":"active","scope":"selected-session","groups":1,"running":1,"done":2,"failed":3,"phases":[{"queued":1,"running":1,"done":1,"failed":1,"hidden":7}],"morePhases":false},"approvals":{"state":"ready","pending":2}}""", + ).jsonObject, + result, + ) + assertFalse(result.toString().contains("private-")) + } + + @Test + fun marksBoundedLimitsAndKeepsUnknownApprovalCountUnavailable() { + val bounded = + projectWearAgentPulse( + gatewayConnected = true, + tasks = + List(100) { index -> task("active-$index", "running") } + + List(50) { index -> task("recent-$index", "failed") }, + swarmAvailable = true, + swarmGroups = emptyList(), + pendingApprovalCount = 9, + approvalsAvailable = false, + approvalsRefreshing = true, + ) + + assertEquals( + Json + .parseToJsonElement( + """{"tasks":{"state":"ready","scope":"bounded","queued":0,"running":100,"completed":0,"failed":50,"activeAtLimit":true,"recentAtLimit":true},"swarm":{"state":"idle","scope":"selected-session"},"approvals":{"state":"refreshing"}}""", + ).jsonObject, + bounded, + ) + } + + @Test + fun makesEveryComponentUnavailableWhenTheGatewayRouteIsStale() { + val result = + projectWearAgentPulse( + gatewayConnected = false, + tasks = listOf(task("private-task", "running")), + swarmAvailable = true, + swarmGroups = emptyList(), + pendingApprovalCount = 1, + approvalsAvailable = true, + approvalsRefreshing = true, + ) + + assertEquals( + Json + .parseToJsonElement( + """{"tasks":{"state":"unavailable"},"swarm":{"state":"unavailable"},"approvals":{"state":"unavailable"}}""", + ).jsonObject, + result, + ) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun concurrentPhoneBudgetKeepsTheCompletedComponentAndBoundsTheSlowOne() = + runTest { + var tasksStartedAt = -1L + var swarmStartedAt = -1L + + val reads = + readWearAgentPulseConcurrently( + readTasks = { + tasksStartedAt = currentTime + "tasks" + }, + readSwarm = { + swarmStartedAt = currentTime + delay(WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS * 2) + "swarm" + }, + ) + + assertEquals("tasks", reads.tasks) + assertEquals(null, reads.swarm) + assertEquals(0L, tasksStartedAt) + assertEquals(0L, swarmStartedAt) + assertEquals(WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS, currentTime) + } + + @Test + fun concurrentPhoneBudgetPreservesCallerCancellation() = + runTest { + val failure = + runCatching { + readWearAgentPulseConcurrently( + readTasks = { throw CancellationException("request retired") }, + readSwarm = { + delay(WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS * 2) + "swarm" + }, + ) + }.exceptionOrNull() + + assertTrue(failure is CancellationException) + } + + private fun task( + id: String, + status: String, + ): BackgroundTask = + BackgroundTask( + id = id, + status = status, + runtime = "private-runtime", + title = "private-title", + agentId = "private-agent", + childSessionKey = "private-session", + createdAtMs = 1, + updatedAtMs = 2, + startedAtMs = 3, + endedAtMs = 4, + progress = "private-progress", + terminal = "private-terminal", + error = "private-error", + prompt = "private-prompt", + ) + + private fun dot( + suffix: String, + status: ChatSwarmDotStatus, + ): ChatSwarmDot = + ChatSwarmDot( + key = "private-dot-$suffix", + label = "private-child-$suffix", + status = status, + ) +} 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 4f83a4cccc9e..95be5d2d960a 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 @@ -1,5 +1,7 @@ package ai.openclaw.app.wear +import ai.openclaw.app.WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS +import ai.openclaw.app.readWearAgentPulseConcurrently import ai.openclaw.wear.shared.WearConnectionFailure import ai.openclaw.wear.shared.WearDecodeResult import ai.openclaw.wear.shared.WearEventType @@ -18,7 +20,10 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job import kotlinx.coroutines.async import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.delay import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.currentTime import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest import kotlinx.coroutines.withTimeout @@ -215,6 +220,53 @@ class WearProxyBridgeTest { assertEquals(1L, event.sequence) } + @Test + fun pulsePhoneBudgetReleasesQueuedEventsBeforeTheWatchDeadline() = + runTest { + val requestStarted = CompletableDeferred() + val (bridge, sent) = + backgroundScope.recordingBridge( + handleRequest = { _, request -> + requestStarted.complete(Unit) + readWearAgentPulseConcurrently( + readTasks = { + delay(WearProtocol.RPC_REQUEST_TIMEOUT_MILLIS * 2) + Unit + }, + readSwarm = { + delay(WearProtocol.RPC_REQUEST_TIMEOUT_MILLIS * 2) + Unit + }, + ) + WearMessage.Response(requestId = request.requestId, ok = true) + }, + ) + + val requestJob = + async { + bridge.handleMessage( + "watch-1", + WearProtocolCodec.encode(request("pulse-1", WearRpcMethod.AgentPulse)), + ) + } + runCurrent() + requestStarted.await() + bridge.publishConnection(connected = true, status = "Connected") + runCurrent() + assertTrue(sent.isEmpty()) + + advanceTimeBy(WEAR_AGENT_PULSE_PHONE_BUDGET_MILLIS - 1) + runCurrent() + assertTrue(sent.isEmpty()) + advanceTimeBy(1) + runCurrent() + + assertTrue(requestJob.await()) + bridge.awaitIdleForTests() + assertEquals(listOf(WearProtocol.RESPONSE_PATH, WearProtocol.EVENT_PATH), sent.map { it.path }) + assertTrue(currentTime < WearProtocol.RPC_REQUEST_TIMEOUT_MILLIS) + } + @Test fun chatStreamProjectionCarriesCanonicalTextAndFallbackCompleteness() { val projector = WearChatStreamProjector() @@ -744,7 +796,10 @@ class WearProxyBridgeTest { private fun SentWearMessage.event(): WearMessage.Event = (WearProtocolCodec.decode(data) as WearDecodeResult.Success).message as WearMessage.Event - private fun request(requestId: String): WearMessage.Request = WearMessage.Request(requestId = requestId, method = WearRpcMethod.ProxyStatus) + private fun request( + requestId: String, + method: WearRpcMethod = WearRpcMethod.ProxyStatus, + ): WearMessage.Request = WearMessage.Request(requestId = requestId, method = method) } private data class SentWearMessage( diff --git a/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyControllerTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyControllerTest.kt index ac9cce4342d5..7d995ac604da 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyControllerTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyControllerTest.kt @@ -98,6 +98,53 @@ class WearProxyControllerTest { ) } + @Test + fun agentPulseAcceptsOnlyTheSelectedSessionAndCallsItsLoaderOnce() = + runTest { + var loaderCalls = 0 + var requestedSessionKey: String? = null + val projected = + buildJsonObject { + put("tasks", buildJsonObject { put("state", "unavailable") }) + put("swarm", buildJsonObject { put("state", "idle") }) + put("approvals", buildJsonObject { put("state", "unavailable") }) + } + val controller = + WearProxyController( + requestGateway = { _, _ -> error("Agent Pulse must stay on the Phone runtime boundary") }, + isGatewayConnected = { true }, + gatewayStatusText = { "Connected" }, + loadAgentPulse = { sessionKey -> + loaderCalls += 1 + requestedSessionKey = sessionKey + projected + }, + ) + + val response = + controller.handle( + request( + WearRpcMethod.AgentPulse, + buildJsonObject { put("sessionKey", "agent:main:thread-7") }, + ), + ) + val rejected = + controller.handle( + request( + WearRpcMethod.AgentPulse, + buildJsonObject { put("detail", true) }, + ), + ) + + assertTrue(response.ok) + assertEquals(projected, response.result) + assertEquals("agent:main:thread-7", requestedSessionKey) + assertEquals(1, loaderCalls) + assertFalse(rejected.ok) + assertEquals("invalid_request", rejected.error?.code) + assertEquals(1, loaderCalls) + } + @Test fun agentsAndGatewayControlsStayOnThePhoneRuntimeBoundary() = runTest { diff --git a/apps/android/wear-shared/src/main/java/ai/openclaw/wear/shared/WearProtocol.kt b/apps/android/wear-shared/src/main/java/ai/openclaw/wear/shared/WearProtocol.kt index 3a3bd714cd25..24a2f2ac7f42 100644 --- a/apps/android/wear-shared/src/main/java/ai/openclaw/wear/shared/WearProtocol.kt +++ b/apps/android/wear-shared/src/main/java/ai/openclaw/wear/shared/WearProtocol.kt @@ -75,6 +75,7 @@ enum class WearProxyCapability( GatewayControls(wireValue = "gateway-controls"), ModelControls(wireValue = "model-controls"), SessionSelectionLookup(wireValue = "session-selection-lookup"), + AgentPulse(wireValue = "agent-pulse"), AttemptScopedRealtimeAudio(wireValue = "attempt-scoped-realtime-audio"), ; @@ -100,6 +101,9 @@ enum class WearRpcMethod { @SerialName("proxy.status") ProxyStatus, + @SerialName("agent.pulse") + AgentPulse, + @SerialName("sessions.list") SessionsList, diff --git a/apps/android/wear-shared/src/test/java/ai/openclaw/wear/shared/WearProtocolTest.kt b/apps/android/wear-shared/src/test/java/ai/openclaw/wear/shared/WearProtocolTest.kt index 50bb5e47720c..86242ca5d0c5 100644 --- a/apps/android/wear-shared/src/test/java/ai/openclaw/wear/shared/WearProtocolTest.kt +++ b/apps/android/wear-shared/src/test/java/ai/openclaw/wear/shared/WearProtocolTest.kt @@ -68,6 +68,7 @@ class WearProtocolTest { val methodNames = mapOf( WearRpcMethod.ProxyStatus to "proxy.status", + WearRpcMethod.AgentPulse to "agent.pulse", WearRpcMethod.SessionsList to "sessions.list", WearRpcMethod.AgentsList to "agents.list", WearRpcMethod.AgentsSelect to "agents.select", @@ -143,6 +144,7 @@ class WearProtocolTest { assertEquals("gateway-controls", WearProxyCapability.GatewayControls.wireValue) assertEquals("model-controls", WearProxyCapability.ModelControls.wireValue) assertEquals("session-selection-lookup", WearProxyCapability.SessionSelectionLookup.wireValue) + assertEquals("agent-pulse", WearProxyCapability.AgentPulse.wireValue) assertEquals( "attempt-scoped-realtime-audio", WearProxyCapability.AttemptScopedRealtimeAudio.wireValue, diff --git a/apps/android/wear/BEHAVIOR.md b/apps/android/wear/BEHAVIOR.md index 0ad5efaaa779..e66e234a94ba 100644 --- a/apps/android/wear/BEHAVIOR.md +++ b/apps/android/wear/BEHAVIOR.md @@ -12,4 +12,5 @@ The watch is a paired-phone companion. It never asks for, receives, or stores Ga - Given a final assistant message while the app is not visible and notifications are allowed, the watch shows one local-only notification with direct reply. Phone-process recreation rediscovers the reachable watch before delivery. If the preferred phone changes before a notification reply, recovery opens the app to reload the session instead of retrying the stale phone. - Given Android 13 or newer without notification permission, Controls offers both an explicit request action and direct access to the watch's app-notification settings. Granting or revoking permission outside the app is reflected when the app resumes, and denying it leaves the rest of the companion usable. - Given a theme or automatic-speech selection, the watch persists that local UI preference without sending it to the phone or Gateway. +- Given a connected paired phone that advertises Agent Pulse, the Pulse page pulls bounded aggregate background-task, selected-session swarm, and pending-approval state only while that page is visible and the app is resumed. Leaving the page, pausing, changing phone, agent, or session, or losing connectivity stops polling and suppresses stale Pulse state. The watch exposes no task text, identifiers, approval details or actions, credentials, or direct Gateway access. - Given the OpenClaw Tile, tapping its mascot Talk action opens the watch app directly on Voice, while the secondary Open edge action starts on Chat. The tile uses the official Android app launcher artwork from the phone manifest's `@mipmap/ic_launcher_foreground` resource with the Core app's dark visual tokens. Neither launch starts microphone capture. Tile rendering performs no phone or network work and persists no cache. diff --git a/apps/android/wear/src/main/java/ai/openclaw/wear/MainActivity.kt b/apps/android/wear/src/main/java/ai/openclaw/wear/MainActivity.kt index 811fc1968e77..a456a8f4ffde 100644 --- a/apps/android/wear/src/main/java/ai/openclaw/wear/MainActivity.kt +++ b/apps/android/wear/src/main/java/ai/openclaw/wear/MainActivity.kt @@ -475,6 +475,8 @@ internal fun OpenClawWearApp( leaveConversationContext() viewModel.selectModel(modelRef) }, + onAgentPulseVisibilityChanged = viewModel::setAgentPulseVisible, + onAgentPulseRefresh = viewModel::refreshAgentPulse, onRefresh = viewModel::refresh, onGatewayEnabledChange = { enabled -> speaker.stop() diff --git a/apps/android/wear/src/main/java/ai/openclaw/wear/WearCompanionUiModels.kt b/apps/android/wear/src/main/java/ai/openclaw/wear/WearCompanionUiModels.kt index a864f3332cfa..0b3b745363f5 100644 --- a/apps/android/wear/src/main/java/ai/openclaw/wear/WearCompanionUiModels.kt +++ b/apps/android/wear/src/main/java/ai/openclaw/wear/WearCompanionUiModels.kt @@ -58,6 +58,10 @@ internal data class WearConversationSnapshot( val selectedModelRef: String? = null, val failure: WearConversationFailure? = null, val realtimeTalk: WearRealtimeTalkSnapshot = WearRealtimeTalkSnapshot(), + val agentPulseSupported: Boolean = false, + val agentPulse: WearAgentPulseSnapshot? = null, + val agentPulseLoading: Boolean = false, + val agentPulseFailure: WearConversationFailure? = null, ) internal enum class WearConversationFailure { @@ -81,6 +85,9 @@ internal enum class WearInteractionState { internal fun WearUiState.toConversationSnapshot(): WearConversationSnapshot? { if (phoneNodeId == null) return null + val pulseSupported = + connected && + WearProxyCapability.AgentPulse in proxyCapabilities return WearConversationSnapshot( gatewayState = if (connected) WearGatewayState.CONNECTED else WearGatewayState.DISCONNECTED, activeAgentId = activeAgentId, @@ -120,5 +127,9 @@ internal fun WearUiState.toConversationSnapshot(): WearConversationSnapshot? { selectedModelRef = selectedModelRef, failure = failure, realtimeTalk = realtimeTalk, + agentPulseSupported = pulseSupported, + agentPulse = agentPulse.takeIf { pulseSupported }, + agentPulseLoading = pulseSupported && agentPulseLoading, + agentPulseFailure = agentPulseFailure.takeIf { pulseSupported }, ) } diff --git a/apps/android/wear/src/main/java/ai/openclaw/wear/WearGatewayRepository.kt b/apps/android/wear/src/main/java/ai/openclaw/wear/WearGatewayRepository.kt index 11b2bcce1e6e..d128a908c4e5 100644 --- a/apps/android/wear/src/main/java/ai/openclaw/wear/WearGatewayRepository.kt +++ b/apps/android/wear/src/main/java/ai/openclaw/wear/WearGatewayRepository.kt @@ -11,6 +11,7 @@ import kotlinx.serialization.json.JsonPrimitive import kotlinx.serialization.json.booleanOrNull import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.contentOrNull +import kotlinx.serialization.json.intOrNull import kotlinx.serialization.json.longOrNull import kotlinx.serialization.json.put import java.util.UUID @@ -26,6 +27,65 @@ internal data class WearProxyStatus( val eventStreamId: String? = null, ) +internal enum class WearAgentPulseTaskState { + Ready, + Unavailable, +} + +internal data class WearAgentPulseTasks( + val state: WearAgentPulseTaskState, + val queued: Int? = null, + val running: Int? = null, + val completed: Int? = null, + val failed: Int? = null, + val activeAtLimit: Boolean? = null, + val recentAtLimit: Boolean? = null, +) + +internal enum class WearAgentPulseSwarmState { + Active, + Idle, + Unavailable, +} + +internal data class WearAgentPulsePhase( + val queued: Int, + val running: Int, + val done: Int, + val failed: Int, + val hidden: Int, +) + +internal data class WearAgentPulseSwarm( + val state: WearAgentPulseSwarmState, + val groups: Int? = null, + val running: Int? = null, + val done: Int? = null, + val failed: Int? = null, + val phases: List = emptyList(), + val morePhases: Boolean? = null, +) + +internal enum class WearAgentPulseApprovalsState { + Ready, + Refreshing, + Unavailable, +} + +internal data class WearAgentPulseApprovals( + val state: WearAgentPulseApprovalsState, + val pending: Int? = null, +) + +internal data class WearAgentPulseSnapshot( + val tasks: WearAgentPulseTasks, + val swarm: WearAgentPulseSwarm, + val approvals: WearAgentPulseApprovals, + val eventSequence: Long?, + val phoneNodeId: String, + val eventStreamId: String? = null, +) + internal data class WearAgent( val id: String, val name: String, @@ -158,6 +218,32 @@ internal class WearGatewayRepository( ) } + suspend fun agentPulse( + expectedNodeId: String, + capabilities: Set, + selectedSessionKey: String? = null, + ): WearAgentPulseSnapshot { + capabilities.require(WearProxyCapability.AgentPulse) + val response = + requester.request( + WearRpcMethod.AgentPulse, + buildJsonObject { + selectedSessionKey?.takeIf(String::isNotBlank)?.let { put("sessionKey", it) } + }, + expectedNodeId, + requirePreferredNode = true, + ) + val result = response.payload.asObject("agent.pulse") + return WearAgentPulseSnapshot( + tasks = parseAgentPulseTasks(result["tasks"]), + swarm = parseAgentPulseSwarm(result["swarm"]), + approvals = parseAgentPulseApprovals(result["approvals"]), + eventStreamId = response.eventStreamId, + eventSequence = response.eventSequence, + phoneNodeId = response.sourceNodeId, + ) + } + suspend fun agents( expectedNodeId: String, capabilities: Set, @@ -423,6 +509,88 @@ internal fun parseWearChatEvent(payload: JsonElement?): WearChatEvent? { ) } +private fun parseAgentPulseTasks(element: JsonElement?): WearAgentPulseTasks { + val source = element as? JsonObject ?: invalidAgentPulse() + return when (source.string("state")) { + "ready" -> { + if (source.string("scope") != "bounded") invalidAgentPulse() + WearAgentPulseTasks( + state = WearAgentPulseTaskState.Ready, + queued = source.nonNegativeInt("queued"), + running = source.nonNegativeInt("running"), + completed = source.nonNegativeInt("completed"), + failed = source.nonNegativeInt("failed"), + activeAtLimit = source.requiredBoolean("activeAtLimit"), + recentAtLimit = source.requiredBoolean("recentAtLimit"), + ) + } + "unavailable" -> WearAgentPulseTasks(state = WearAgentPulseTaskState.Unavailable) + else -> invalidAgentPulse() + } +} + +private fun parseAgentPulseSwarm(element: JsonElement?): WearAgentPulseSwarm { + val source = element as? JsonObject ?: invalidAgentPulse() + return when (source.string("state")) { + "active" -> { + if (source.string("scope") != "selected-session") invalidAgentPulse() + val phases = source["phases"] as? JsonArray ?: invalidAgentPulse() + if (phases.size > MAX_AGENT_PULSE_PHASES) invalidAgentPulse() + WearAgentPulseSwarm( + state = WearAgentPulseSwarmState.Active, + groups = source.nonNegativeInt("groups"), + running = source.nonNegativeInt("running"), + done = source.nonNegativeInt("done"), + failed = source.nonNegativeInt("failed"), + phases = phases.map(::parseAgentPulsePhase), + morePhases = source.requiredBoolean("morePhases"), + ) + } + "idle" -> { + if (source.string("scope") != "selected-session") invalidAgentPulse() + WearAgentPulseSwarm(state = WearAgentPulseSwarmState.Idle) + } + "unavailable" -> WearAgentPulseSwarm(state = WearAgentPulseSwarmState.Unavailable) + else -> invalidAgentPulse() + } +} + +private fun parseAgentPulsePhase(element: JsonElement): WearAgentPulsePhase { + val source = element as? JsonObject ?: invalidAgentPulse() + return WearAgentPulsePhase( + queued = source.nonNegativeInt("queued"), + running = source.nonNegativeInt("running"), + done = source.nonNegativeInt("done"), + failed = source.nonNegativeInt("failed"), + hidden = source.nonNegativeInt("hidden"), + ) +} + +private fun parseAgentPulseApprovals(element: JsonElement?): WearAgentPulseApprovals { + val source = element as? JsonObject ?: invalidAgentPulse() + return when (source.string("state")) { + "ready" -> + WearAgentPulseApprovals( + state = WearAgentPulseApprovalsState.Ready, + pending = source.nonNegativeInt("pending"), + ) + "refreshing" -> WearAgentPulseApprovals(state = WearAgentPulseApprovalsState.Refreshing) + "unavailable" -> WearAgentPulseApprovals(state = WearAgentPulseApprovalsState.Unavailable) + else -> invalidAgentPulse() + } +} + +private fun JsonObject.nonNegativeInt(name: String): Int { + val value = (this[name] as? JsonPrimitive)?.takeUnless(JsonPrimitive::isString)?.intOrNull + return value?.takeIf { it >= 0 } ?: invalidAgentPulse() +} + +private fun JsonObject.requiredBoolean(name: String): Boolean = boolean(name) ?: invalidAgentPulse() + +private fun invalidAgentPulse(): Nothing = throw WearProxyException("invalid_response", "agent.pulse returned invalid data") + +private const val MAX_AGENT_PULSE_PHASES = 8 + private fun parseSession( element: JsonElement, phoneNodeId: String, diff --git a/apps/android/wear/src/main/java/ai/openclaw/wear/WearProxyClient.kt b/apps/android/wear/src/main/java/ai/openclaw/wear/WearProxyClient.kt index a199991ba67d..aea43842da1a 100644 --- a/apps/android/wear/src/main/java/ai/openclaw/wear/WearProxyClient.kt +++ b/apps/android/wear/src/main/java/ai/openclaw/wear/WearProxyClient.kt @@ -397,6 +397,10 @@ internal data class WearResponseRequest( val eventGeneration: Long, ) +internal data class WearReadOnlyResponseRequest( + val eventGeneration: Long, +) + internal class WearEventSequenceTracker { private var streamId: String? = null private var lastSequence: Long? = null @@ -461,6 +465,11 @@ internal class WearEventSequenceTracker { return WearResponseRequest(responseGeneration = responseGeneration, eventGeneration = eventGeneration) } + // Read-only projections may overlap a model request. Their owner supplies + // feature-local cancellation, while this token only binds the event cursor. + @Synchronized + fun beginReadOnlyResponseRequest(): WearReadOnlyResponseRequest = WearReadOnlyResponseRequest(eventGeneration = eventGeneration) + @Synchronized fun invalidateResponseRequests() { responseGeneration += 1 @@ -473,11 +482,31 @@ internal class WearEventSequenceTracker { sequence: Long?, ): Boolean { if (request.responseGeneration != responseGeneration) return false + return isEventCursorCurrent(request.eventGeneration, streamId, sequence) + } + + @Synchronized + fun isReadOnlyResponseCurrent( + request: WearReadOnlyResponseRequest, + streamId: String?, + sequence: Long?, + ): Boolean = isEventCursorCurrent(request.eventGeneration, streamId, sequence) + + private fun isEventCursorCurrent( + requestEventGeneration: Long, + responseStreamId: String?, + sequence: Long?, + ): Boolean { if (awaitingSnapshot) return false - if (this.streamId != streamId && (this.streamId != null || streamId != null)) return false + if ( + this.streamId != responseStreamId && + (this.streamId != null || responseStreamId != null) + ) { + return false + } val currentSequence = lastSequence return if (sequence == null) { - request.eventGeneration == eventGeneration + requestEventGeneration == eventGeneration } else { sequence == currentSequence } diff --git a/apps/android/wear/src/main/java/ai/openclaw/wear/WearScreens.kt b/apps/android/wear/src/main/java/ai/openclaw/wear/WearScreens.kt index 7f07c593e655..851c5f3786b8 100644 --- a/apps/android/wear/src/main/java/ai/openclaw/wear/WearScreens.kt +++ b/apps/android/wear/src/main/java/ai/openclaw/wear/WearScreens.kt @@ -38,6 +38,7 @@ import androidx.compose.foundation.selection.selectable import androidx.compose.foundation.shape.CircleShape import androidx.compose.foundation.shape.RoundedCornerShape import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableIntStateOf @@ -70,6 +71,9 @@ import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.unit.Dp import androidx.compose.ui.unit.dp import androidx.compose.ui.unit.sp +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleEventObserver +import androidx.lifecycle.compose.LocalLifecycleOwner import androidx.wear.compose.foundation.lazy.TransformingLazyColumn import androidx.wear.compose.foundation.lazy.rememberTransformingLazyColumnState import androidx.wear.compose.foundation.pager.HorizontalPager @@ -90,8 +94,16 @@ internal enum class WearHomePage { Chat, Voice, Controls, + Pulse, } +internal fun wearHomePages(agentPulseSupported: Boolean): List = + if (agentPulseSupported) { + WearHomePage.entries + } else { + WearHomePage.entries.filterNot { it == WearHomePage.Pulse } + } + private const val VOICE_MODE_COUNT = 2 private const val VOICE_HOME_MODE = 0 private const val VOICE_THREAD_MODE = 1 @@ -154,6 +166,8 @@ internal fun OpenClawWearScreens( onSelectAgent: (String) -> Unit, onSelectSession: (String) -> Unit, onSelectModel: (String) -> Unit, + onAgentPulseVisibilityChanged: (Boolean) -> Unit = {}, + onAgentPulseRefresh: () -> Unit = {}, onRefresh: () -> Unit, onGatewayEnabledChange: (Boolean) -> Unit, onThemeModeChange: (WearThemeMode) -> Unit, @@ -163,6 +177,47 @@ internal fun OpenClawWearScreens( onSpeakLatest: () -> Unit, onStopSpeaking: () -> Unit, ) { + val lifecycleOwner = LocalLifecycleOwner.current + val agentPulseSupported = snapshot?.agentPulseSupported == true + val homePages = remember(agentPulseSupported) { wearHomePages(agentPulseSupported) } + val initialPageIndex = homePages.indexOf(initialPage).takeIf { it >= 0 } ?: 0 + val pagerState = + rememberPagerState( + initialPage = initialPageIndex, + pageCount = { homePages.size }, + ) + var lifecycleResumed by remember(lifecycleOwner) { + mutableStateOf( + lifecycleOwner.lifecycle.currentState.isAtLeast(Lifecycle.State.RESUMED), + ) + } + DisposableEffect(lifecycleOwner) { + val observer = + LifecycleEventObserver { _, _ -> + val resumed = + lifecycleOwner.lifecycle.currentState.isAtLeast(Lifecycle.State.RESUMED) + lifecycleResumed = resumed + if (!resumed) { + onAgentPulseVisibilityChanged(false) + } + } + lifecycleOwner.lifecycle.addObserver(observer) + onDispose { + lifecycleOwner.lifecycle.removeObserver(observer) + onAgentPulseVisibilityChanged(false) + } + } + LaunchedEffect(homePages) { + if (homePages.getOrNull(pagerState.currentPage) == null) { + pagerState.scrollToPage(homePages.indexOf(WearHomePage.Chat)) + } + } + LaunchedEffect(pagerState, lifecycleResumed, snapshot != null, homePages) { + snapshotFlow { homePages.getOrNull(pagerState.currentPage) == WearHomePage.Pulse } + .collect { selected -> + onAgentPulseVisibilityChanged(snapshot != null && lifecycleResumed && selected) + } + } if (snapshot == null) { ConnectionStateScreen( loading = loading, @@ -173,11 +228,6 @@ internal fun OpenClawWearScreens( } val colors = OpenClawWearTheme.colors - val pagerState = - rememberPagerState( - initialPage = initialPage.ordinal, - pageCount = { WearHomePage.entries.size }, - ) val voicePagerState = rememberPagerState(pageCount = { VOICE_MODE_COUNT }) val pagerScope = rememberCoroutineScope() val realtimeActive = snapshot.realtimeTalk.active || realtimeCapturing @@ -187,11 +237,12 @@ internal fun OpenClawWearScreens( LaunchedEffect(navigationRequest?.id) { val request = navigationRequest ?: return@LaunchedEffect val destination = wearLaunchPage(request.target, realtimeActive) - pagerState.scrollToPage(destination.ordinal) + val destinationIndex = homePages.indexOf(destination).takeIf { it >= 0 } ?: 0 + pagerState.scrollToPage(destinationIndex) onNavigationRequestHandled(request.id) } LaunchedEffect(pagerState.currentPage, showVoiceSwipeHint) { - if (pagerState.currentPage == WearHomePage.Voice.ordinal && showVoiceSwipeHint) { + if (homePages.getOrNull(pagerState.currentPage) == WearHomePage.Voice && showVoiceSwipeHint) { delay(1_800L) showVoiceSwipeHint = false } @@ -212,9 +263,9 @@ internal fun OpenClawWearScreens( delay(250L) } } - BackHandler(enabled = pagerState.currentPage == WearHomePage.Voice.ordinal) { + BackHandler(enabled = homePages.getOrNull(pagerState.currentPage) == WearHomePage.Voice) { pagerScope.launch { - pagerState.animateScrollToPage(WearHomePage.Chat.ordinal) + pagerState.animateScrollToPage(homePages.indexOf(WearHomePage.Chat)) } } HorizontalPagerScaffold( @@ -229,12 +280,12 @@ internal fun OpenClawWearScreens( modifier = Modifier.fillMaxSize(), rotaryScrollableBehavior = null, userScrollEnabled = - pagerState.currentPage != WearHomePage.Voice.ordinal || + homePages.getOrNull(pagerState.currentPage) != WearHomePage.Voice || voicePagerState.currentPage == VOICE_HOME_MODE || voicePagerState.currentPage == VOICE_THREAD_MODE, ) { page -> - when (page) { - WearHomePage.Chat.ordinal -> + when (homePages.getOrNull(page)) { + WearHomePage.Chat -> ChatPage( snapshot = snapshot, interaction = interaction, @@ -251,25 +302,10 @@ internal fun OpenClawWearScreens( onSpeakLatest = onSpeakLatest, onStopSpeaking = onStopSpeaking, ) - WearHomePage.Controls.ordinal -> - ControlsPage( - snapshot = snapshot, - themeMode = themeMode, - autoSpeak = autoSpeak, - notificationsGranted = notificationsGranted, - gatewayControlSupported = snapshot.gatewayControlsSupported, - actionBusy = actionBusy, - onThemeModeChange = onThemeModeChange, - onAutoSpeakChange = onAutoSpeakChange, - onRequestNotifications = onRequestNotifications, - onOpenNotificationSettings = onOpenNotificationSettings, - onRefresh = onRefresh, - onGatewayEnabledChange = onGatewayEnabledChange, - ) - else -> + WearHomePage.Voice -> VoicePage( voicePagerState = voicePagerState, - showSwipeHint = showVoiceSwipeHint && pagerState.currentPage == WearHomePage.Voice.ordinal, + showSwipeHint = showVoiceSwipeHint && homePages.getOrNull(pagerState.currentPage) == WearHomePage.Voice, realtimeTalk = snapshot.realtimeTalk, speaking = speaking, realtimeCapturing = realtimeCapturing, @@ -285,6 +321,33 @@ internal fun OpenClawWearScreens( onRealtimeTalk = onRealtimeTalk, onStopSpeaking = onStopSpeaking, ) + WearHomePage.Controls -> + ControlsPage( + snapshot = snapshot, + themeMode = themeMode, + autoSpeak = autoSpeak, + notificationsGranted = notificationsGranted, + gatewayControlSupported = snapshot.gatewayControlsSupported, + actionBusy = actionBusy, + onThemeModeChange = onThemeModeChange, + onAutoSpeakChange = onAutoSpeakChange, + onRequestNotifications = onRequestNotifications, + onOpenNotificationSettings = onOpenNotificationSettings, + onRefresh = onRefresh, + onGatewayEnabledChange = onGatewayEnabledChange, + ) + WearHomePage.Pulse -> + AgentPulsePage( + snapshot = snapshot, + onRefresh = { + if (snapshot.agentPulseSupported) { + onAgentPulseRefresh() + } else { + onRefresh() + } + }, + ) + else -> Unit } } } @@ -1425,6 +1488,254 @@ private fun ControlsPage( } } +@Composable +private fun AgentPulsePage( + snapshot: WearConversationSnapshot, + onRefresh: () -> Unit, +) { + val pulse = snapshot.agentPulse + WearPage(pageLabel = stringResource(R.string.pulse)) { + when { + snapshot.gatewayState != WearGatewayState.CONNECTED -> + item { + EmptyPanel( + title = stringResource(R.string.pulse_unavailable), + detail = stringResource(R.string.gateway_offline_detail), + ) + } + !snapshot.agentPulseSupported -> + item { + EmptyPanel( + title = stringResource(R.string.pulse_unavailable), + detail = stringResource(R.string.update_required_detail), + ) + } + pulse == null -> + item { + EmptyPanel( + title = + if (snapshot.agentPulseLoading) { + stringResource(R.string.pulse_loading) + } else { + stringResource(R.string.pulse_unavailable) + }, + detail = + when { + snapshot.agentPulseLoading -> stringResource(R.string.pulse_loading_detail) + snapshot.agentPulseFailure != null -> failureDetail(snapshot.agentPulseFailure) + else -> stringResource(R.string.try_again) + }, + ) + } + else -> { + item { AgentPulseTasksPanel(pulse.tasks) } + item { AgentPulseSwarmPanel(pulse.swarm) } + item { AgentPulseApprovalsPanel(pulse.approvals) } + snapshot.agentPulseFailure?.let { pulseFailure -> + item { InlineError(text = failureDetail(pulseFailure)) } + } + } + } + item { + SecondaryButton( + label = stringResource(R.string.refresh), + enabled = !snapshot.agentPulseLoading, + onClick = onRefresh, + ) + } + } +} + +@Composable +private fun AgentPulseTasksPanel(tasks: WearAgentPulseTasks) { + val ready = tasks.state == WearAgentPulseTaskState.Ready + Panel { + AgentPulsePanelHeader( + title = stringResource(R.string.pulse_tasks), + status = + if (ready) { + stringResource(R.string.pulse_ready) + } else { + stringResource(R.string.pulse_unavailable) + }, + statusColor = + if (ready) { + OpenClawWearTheme.colors.success + } else { + OpenClawWearTheme.colors.danger + }, + ) + if (ready) { + AgentPulseMetricRow(stringResource(R.string.pulse_queued), tasks.queued) + AgentPulseMetricRow(stringResource(R.string.pulse_running), tasks.running) + AgentPulseMetricRow(stringResource(R.string.pulse_completed), tasks.completed) + AgentPulseMetricRow(stringResource(R.string.pulse_failed), tasks.failed) + AgentPulseDetail(text = stringResource(R.string.pulse_task_snapshot_bounded)) + if (tasks.activeAtLimit == true) { + AgentPulseDetail( + text = stringResource(R.string.pulse_active_at_limit), + color = OpenClawWearTheme.colors.warning, + ) + } + if (tasks.recentAtLimit == true) { + AgentPulseDetail( + text = stringResource(R.string.pulse_recent_at_limit), + color = OpenClawWearTheme.colors.warning, + ) + } + } + } +} + +@Composable +private fun AgentPulseSwarmPanel(swarm: WearAgentPulseSwarm) { + val colors = OpenClawWearTheme.colors + val status = + when (swarm.state) { + WearAgentPulseSwarmState.Active -> stringResource(R.string.pulse_swarm_active) + WearAgentPulseSwarmState.Idle -> stringResource(R.string.pulse_swarm_idle) + WearAgentPulseSwarmState.Unavailable -> stringResource(R.string.pulse_unavailable) + } + val statusColor = + when (swarm.state) { + WearAgentPulseSwarmState.Active -> colors.warning + WearAgentPulseSwarmState.Idle -> colors.success + WearAgentPulseSwarmState.Unavailable -> colors.danger + } + Panel { + AgentPulsePanelHeader( + title = stringResource(R.string.pulse_swarm), + status = status, + statusColor = statusColor, + ) + if (swarm.state == WearAgentPulseSwarmState.Active) { + AgentPulseMetricRow(stringResource(R.string.pulse_groups), swarm.groups) + AgentPulseMetricRow(stringResource(R.string.pulse_running), swarm.running) + AgentPulseMetricRow(stringResource(R.string.pulse_done), swarm.done) + AgentPulseMetricRow(stringResource(R.string.pulse_failed), swarm.failed) + swarm.phases.forEachIndexed { index, phase -> + Spacer(modifier = Modifier.height(6.dp)) + Text( + text = stringResource(R.string.pulse_phase, index + 1), + color = colors.text, + fontSize = 11.sp, + fontWeight = FontWeight.SemiBold, + ) + Text( + text = + stringResource( + R.string.pulse_phase_counts, + phase.queued, + phase.running, + phase.done, + phase.failed, + phase.hidden, + ), + color = colors.textMuted, + fontSize = 10.sp, + lineHeight = 14.sp, + ) + } + if (swarm.morePhases == true) { + AgentPulseDetail(text = stringResource(R.string.pulse_more_phases)) + } + } + } +} + +@Composable +private fun AgentPulseApprovalsPanel(approvals: WearAgentPulseApprovals) { + val colors = OpenClawWearTheme.colors + val status = + when (approvals.state) { + WearAgentPulseApprovalsState.Ready -> stringResource(R.string.pulse_ready) + WearAgentPulseApprovalsState.Refreshing -> stringResource(R.string.pulse_refreshing) + WearAgentPulseApprovalsState.Unavailable -> stringResource(R.string.pulse_unavailable) + } + val statusColor = + when (approvals.state) { + WearAgentPulseApprovalsState.Ready -> colors.success + WearAgentPulseApprovalsState.Refreshing -> colors.warning + WearAgentPulseApprovalsState.Unavailable -> colors.danger + } + Panel { + AgentPulsePanelHeader( + title = stringResource(R.string.pulse_attention), + status = status, + statusColor = statusColor, + ) + if (approvals.state == WearAgentPulseApprovalsState.Ready) { + AgentPulseMetricRow( + label = stringResource(R.string.pulse_pending_requests), + value = approvals.pending, + ) + } + } +} + +@Composable +private fun AgentPulsePanelHeader( + title: String, + status: String, + statusColor: Color, +) { + Text( + text = title, + color = OpenClawWearTheme.colors.text, + fontSize = 15.sp, + fontWeight = FontWeight.SemiBold, + modifier = Modifier.fillMaxWidth(), + ) + Text( + text = stringResource(R.string.pulse_status, status), + color = statusColor, + fontSize = 11.sp, + fontWeight = FontWeight.SemiBold, + modifier = Modifier.fillMaxWidth(), + ) + Spacer(modifier = Modifier.height(6.dp)) +} + +@Composable +private fun AgentPulseMetricRow( + label: String, + value: Int?, +) { + Row( + modifier = Modifier.fillMaxWidth(), + horizontalArrangement = Arrangement.SpaceBetween, + verticalAlignment = Alignment.CenterVertically, + ) { + Text( + text = label, + color = OpenClawWearTheme.colors.textMuted, + fontSize = 11.sp, + modifier = Modifier.weight(1f), + ) + Text( + text = value?.toString() ?: stringResource(R.string.pulse_unknown), + color = OpenClawWearTheme.colors.text, + fontSize = 11.sp, + fontWeight = FontWeight.SemiBold, + ) + } +} + +@Composable +private fun AgentPulseDetail( + text: String, + color: Color = OpenClawWearTheme.colors.textMuted, +) { + Spacer(modifier = Modifier.height(4.dp)) + Text( + text = text, + color = color, + fontSize = 10.sp, + lineHeight = 14.sp, + modifier = Modifier.fillMaxWidth(), + ) +} + @Composable private fun ConnectionStateScreen( loading: Boolean, 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 bdbb8e81ca90..8c997c524d88 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 @@ -11,6 +11,7 @@ import androidx.lifecycle.viewModelScope import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.Job +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow @@ -45,6 +46,9 @@ internal data class WearUiState( val talkBusy: Boolean = false, val controlBusy: Boolean = false, val failure: WearConversationFailure? = null, + val agentPulse: WearAgentPulseSnapshot? = null, + val agentPulseLoading: Boolean = false, + val agentPulseFailure: WearConversationFailure? = null, ) internal fun WearUiState.resetForPhoneChange(): WearUiState = @@ -71,6 +75,9 @@ internal fun WearUiState.resetForPhoneChange(): WearUiState = talkBusy = false, controlBusy = false, failure = null, + agentPulse = null, + agentPulseLoading = false, + agentPulseFailure = null, ) internal fun WearUiState.switchAgentContext(agentId: String): WearUiState = @@ -83,6 +90,9 @@ internal fun WearUiState.switchAgentContext(agentId: String): WearUiState = activeRunId = null, selectedModelRef = null, models = emptyList(), + agentPulse = null, + agentPulseLoading = false, + agentPulseFailure = null, ) internal fun WearUiState.switchSessionContext(session: WearSession): WearUiState = @@ -97,6 +107,9 @@ internal fun WearUiState.switchSessionContext(session: WearSession): WearUiState realtimeMouthLevel = 0f, talkBusy = false, failure = null, + agentPulse = null, + agentPulseLoading = false, + agentPulseFailure = null, ) internal fun WearUiState.switchModelContext(modelRef: String): WearUiState { @@ -187,6 +200,9 @@ internal class WearViewModel( private val controlBusyOwner = WearControlBusyOwner() private var loadJob: Job? = null private var phoneRouteGeneration = 0L + private var agentPulsePollJob: Job? = null + private var agentPulseVisible = false + private var agentPulseRequestGeneration = 0L // Session switches clear the prior bounded catalog; only the matching phone/session may refill it. private var modelLoadJob: Job? = null @@ -242,6 +258,26 @@ internal class WearViewModel( loadSessions() } + fun setAgentPulseVisible(visible: Boolean) { + if (agentPulseVisible == visible) { + if (visible && agentPulsePollJob?.isActive != true) { + restartAgentPulsePolling(forceLoading = true) + } + return + } + agentPulseVisible = visible + if (visible) { + restartAgentPulsePolling(forceLoading = true) + } else { + invalidateAgentPulse(clearSnapshot = true) + } + } + + fun refreshAgentPulse() { + if (!agentPulseVisible) return + restartAgentPulsePolling(forceLoading = true) + } + fun openSession(session: WearSession) { val current = mutableState.value if ( @@ -256,9 +292,11 @@ internal class WearViewModel( } endRealtimeTalkForNavigation() cancelModelLoad() + invalidateAgentPulse(clearSnapshot = true) mutableState.update { it.switchSessionContext(session) } loadModels(session) loadHistory(session) + restartAgentPulsePolling(forceLoading = true) } fun closeSession() { @@ -548,6 +586,7 @@ internal class WearViewModel( } private fun loadSessions(expectedNodeId: String? = null) { + invalidateAgentPulse(clearSnapshot = true) cancelLoad() cancelModelLoad() loadJob = @@ -677,6 +716,8 @@ internal class WearViewModel( pendingEvents.forEach(::handleEvent) if (status.connected && selectedSession != null) { loadHistory(selectedSession) + } else { + restartAgentPulsePolling(forceLoading = true) } } catch (err: CancellationException) { throw err @@ -771,6 +812,7 @@ internal class WearViewModel( ) } pendingEvents.forEach(::handleEvent) + restartAgentPulsePolling(forceLoading = true) if (catalogScopeChanged) loadModels(loadedSession) } catch (err: CancellationException) { throw err @@ -786,6 +828,7 @@ internal class WearViewModel( } if (currentLoad && mutableState.value.selectedSession?.key == session.key) { recordFailure(err, loading = false) + restartAgentPulsePolling(forceLoading = true) } } } @@ -890,6 +933,7 @@ internal class WearViewModel( sourceChanged: Boolean, ) { // A source switch or sequence gap invalidates the old phone's live state. + invalidateAgentPulse(clearSnapshot = true) // Buffer this boundary event until the selected phone supplies a watermark. eventSequenceTracker.requireSnapshot() resyncEventBuffer.start(event) @@ -932,6 +976,7 @@ internal class WearViewModel( } private fun resetForPhoneRouteChange() { + invalidateAgentPulse(clearSnapshot = true) phoneRouteGeneration += 1 controlBusyOwner.reset() mutableState.update(WearUiState::resetForPhoneChange) @@ -952,6 +997,7 @@ internal class WearViewModel( cancelLoad() val connected = payload.boolean("connected") ?: false if (!connected) { + invalidateAgentPulse(clearSnapshot = true) talkStartJob?.cancel() talkStartJob = null talkAttemptId = null @@ -1028,6 +1074,181 @@ internal class WearViewModel( eventSequenceTracker.invalidateResponseRequests() } + private fun restartAgentPulsePolling(forceLoading: Boolean = false) { + invalidateAgentPulse(clearSnapshot = false) + val initialState = mutableState.value + if (!shouldPollAgentPulse(initialState, agentPulseVisible)) { + mutableState.update { + it.copy( + agentPulse = null, + agentPulseLoading = false, + agentPulseFailure = null, + ) + } + return + } + val phoneNodeId = initialState.phoneNodeId ?: return + val activeAgentId = initialState.activeAgentId + val selectedSessionKey = initialState.selectedSession?.key + val routeGeneration = phoneRouteGeneration + val requestGeneration = agentPulseRequestGeneration + agentPulsePollJob = + viewModelScope.launch { + var showForcedLoading = forceLoading + try { + while ( + isCurrentAgentPulseRoute( + phoneNodeId = phoneNodeId, + activeAgentId = activeAgentId, + selectedSessionKey = selectedSessionKey, + routeGeneration = routeGeneration, + requestGeneration = requestGeneration, + ) + ) { + mutableState.update { state -> + if ( + isCurrentAgentPulseRoute( + phoneNodeId, + activeAgentId, + selectedSessionKey, + routeGeneration, + requestGeneration, + state, + ) + ) { + state.copy( + agentPulseLoading = showForcedLoading || state.agentPulse == null, + agentPulseFailure = null, + ) + } else { + state + } + } + try { + val responseRequest = eventSequenceTracker.beginReadOnlyResponseRequest() + val pulse = + repository.agentPulse( + expectedNodeId = phoneNodeId, + capabilities = initialState.proxyCapabilities, + selectedSessionKey = selectedSessionKey, + ) + if ( + pulse.phoneNodeId != phoneNodeId || + !isCurrentAgentPulseRoute( + phoneNodeId, + activeAgentId, + selectedSessionKey, + routeGeneration, + requestGeneration, + ) + ) { + return@launch + } + if ( + !eventSequenceTracker.isReadOnlyResponseCurrent( + responseRequest, + pulse.eventStreamId, + pulse.eventSequence, + ) + ) { + // Rebuild from a canonical snapshot instead of exposing an old Phone cursor. + loadSessions(pulse.phoneNodeId) + return@launch + } + mutableState.update { state -> + if ( + isCurrentAgentPulseRoute( + phoneNodeId, + activeAgentId, + selectedSessionKey, + routeGeneration, + requestGeneration, + state, + ) + ) { + state.copy( + agentPulse = pulse, + agentPulseLoading = false, + agentPulseFailure = null, + ) + } else { + state + } + } + } catch (err: CancellationException) { + throw err + } catch (err: Throwable) { + if ( + err is WearProxyException && + err.code == "phone_changed" && + requestGeneration == agentPulseRequestGeneration + ) { + reloadForPreferredPhone(nodeId = null) + return@launch + } + mutableState.update { state -> + if ( + isCurrentAgentPulseRoute( + phoneNodeId, + activeAgentId, + selectedSessionKey, + routeGeneration, + requestGeneration, + state, + ) + ) { + state.copy( + agentPulse = if (err.isConnectivityFailure()) null else state.agentPulse, + agentPulseLoading = false, + agentPulseFailure = err.toWearConversationFailure(), + ) + } else { + state + } + } + } + showForcedLoading = false + delay(AGENT_PULSE_POLL_INTERVAL_MS) + } + } finally { + if (agentPulsePollJob === coroutineContext[Job]) agentPulsePollJob = null + } + } + } + + private fun isCurrentAgentPulseRoute( + phoneNodeId: String, + activeAgentId: String?, + selectedSessionKey: String?, + routeGeneration: Long, + requestGeneration: Long, + state: WearUiState = mutableState.value, + ): Boolean = + wearAgentPulseRouteIsCurrent( + requestedPhoneNodeId = phoneNodeId, + requestedAgentId = activeAgentId, + requestedSessionKey = selectedSessionKey, + requestedRouteGeneration = routeGeneration, + currentRouteGeneration = phoneRouteGeneration, + requestedGeneration = requestGeneration, + currentGeneration = agentPulseRequestGeneration, + pulseVisible = agentPulseVisible, + state = state, + ) + + private fun invalidateAgentPulse(clearSnapshot: Boolean) { + agentPulseRequestGeneration += 1 + agentPulsePollJob?.cancel() + agentPulsePollJob = null + mutableState.update { + it.copy( + agentPulse = if (clearSnapshot) null else it.agentPulse, + agentPulseLoading = false, + agentPulseFailure = null, + ) + } + } + private fun reloadHistoryIfSelected( session: WearSession, routeGeneration: Long, @@ -1090,6 +1311,7 @@ internal class WearViewModel( ) { val disconnected = error.isConnectivityFailure() if (disconnected) { + invalidateAgentPulse(clearSnapshot = true) talkStartJob?.cancel() talkStartJob = null talkAttemptId = null @@ -1132,11 +1354,44 @@ internal class WearViewModel( override fun onCleared() { modelLoadJob?.cancel() + agentPulseVisible = false + agentPulseRequestGeneration += 1 + agentPulsePollJob?.cancel() + agentPulsePollJob = null talkStartJob?.cancel() realtimeTalkClient.shutdown() } } +private const val AGENT_PULSE_POLL_INTERVAL_MS = 15_000L + +internal fun shouldPollAgentPulse( + state: WearUiState, + pulseVisible: Boolean, +): Boolean = + pulseVisible && + state.connected && + !state.phoneNodeId.isNullOrBlank() && + WearProxyCapability.AgentPulse in state.proxyCapabilities + +internal fun wearAgentPulseRouteIsCurrent( + requestedPhoneNodeId: String, + requestedAgentId: String?, + requestedSessionKey: String?, + requestedRouteGeneration: Long, + currentRouteGeneration: Long, + requestedGeneration: Long, + currentGeneration: Long, + pulseVisible: Boolean, + state: WearUiState, +): Boolean = + requestedRouteGeneration == currentRouteGeneration && + requestedGeneration == currentGeneration && + shouldPollAgentPulse(state, pulseVisible) && + state.phoneNodeId == requestedPhoneNodeId && + state.activeAgentId == requestedAgentId && + state.selectedSession?.key == requestedSessionKey + internal fun coherentWearActiveSessionKey( statusAgentId: String?, statusSessionKey: String?, diff --git a/apps/android/wear/src/main/res/values/strings.xml b/apps/android/wear/src/main/res/values/strings.xml index 41453456614b..83091df3cfc4 100644 --- a/apps/android/wear/src/main/res/values/strings.xml +++ b/apps/android/wear/src/main/res/values/strings.xml @@ -88,4 +88,30 @@ Open sessions and reply through your paired phone OPEN PHONE PROXY + Pulse + Background tasks + Session swarm + Attention + Status: %1$s + Ready + Refreshing + Unavailable + Loading Pulse + Reading bounded aggregates from the paired phone. + Queued + Running + Completed + Done + Failed + Groups + Pending requests + Bounded active and recent snapshot + Active task list is at its limit + Recent task list is at its limit + Active + Idle + Phase %1$d + Queued %1$d, Running %2$d, Done %3$d, Failed %4$d, Hidden %5$d + More phases not shown + Unknown diff --git a/apps/android/wear/src/test/java/ai/openclaw/wear/MainActivityTest.kt b/apps/android/wear/src/test/java/ai/openclaw/wear/MainActivityTest.kt index fde10e771d54..d42a2094b111 100644 --- a/apps/android/wear/src/test/java/ai/openclaw/wear/MainActivityTest.kt +++ b/apps/android/wear/src/test/java/ai/openclaw/wear/MainActivityTest.kt @@ -1,5 +1,6 @@ package ai.openclaw.wear +import ai.openclaw.wear.shared.WearProxyCapability import ai.openclaw.wear.shared.WearRealtimeTalkEntry import ai.openclaw.wear.shared.WearRealtimeTalkRole import ai.openclaw.wear.shared.WearRealtimeTalkSnapshot @@ -266,6 +267,92 @@ class MainActivityTest { assertNull(snapshot?.sessions?.single()?.title) } + @Test + fun conversationSnapshotExposesPulseOnlyForConnectedCapablePhone() { + val pulse = + WearAgentPulseSnapshot( + tasks = + WearAgentPulseTasks( + state = WearAgentPulseTaskState.Ready, + queued = 2, + running = 3, + completed = 5, + failed = 1, + activeAtLimit = false, + recentAtLimit = false, + ), + swarm = + WearAgentPulseSwarm( + state = WearAgentPulseSwarmState.Active, + groups = 1, + running = 1, + done = 0, + failed = 0, + phases = + listOf( + WearAgentPulsePhase( + queued = 2, + running = 1, + done = 0, + failed = 0, + hidden = 0, + ), + ), + morePhases = false, + ), + approvals = + WearAgentPulseApprovals( + state = WearAgentPulseApprovalsState.Ready, + pending = 2, + ), + eventSequence = 7L, + phoneNodeId = "phone-1", + eventStreamId = "epoch-1", + ) + val capable = + WearUiState( + loading = false, + connected = true, + phoneNodeId = "phone-1", + proxyCapabilities = setOf(WearProxyCapability.AgentPulse), + agentPulse = pulse, + agentPulseLoading = true, + agentPulseFailure = WearConversationFailure.INTERNAL_ERROR, + ).toConversationSnapshot() + + assertTrue(capable?.agentPulseSupported == true) + assertEquals(pulse, capable?.agentPulse) + assertTrue(capable?.agentPulseLoading == true) + assertEquals(WearConversationFailure.INTERNAL_ERROR, capable?.agentPulseFailure) + + val offline = + WearUiState( + loading = false, + connected = false, + phoneNodeId = "phone-1", + proxyCapabilities = setOf(WearProxyCapability.AgentPulse), + agentPulse = pulse, + agentPulseLoading = true, + agentPulseFailure = WearConversationFailure.INTERNAL_ERROR, + ).toConversationSnapshot() + + assertFalse(offline?.agentPulseSupported == true) + assertNull(offline?.agentPulse) + assertFalse(offline?.agentPulseLoading == true) + assertNull(offline?.agentPulseFailure) + + val unsupported = + WearUiState( + loading = false, + connected = true, + phoneNodeId = "phone-1", + agentPulse = pulse, + ).toConversationSnapshot() + + assertFalse(unsupported?.agentPulseSupported == true) + assertNull(unsupported?.agentPulse) + } + @Test fun connectionEventsPreserveTypedAndLegacyIncompatibilityReasons() { assertEquals( diff --git a/apps/android/wear/src/test/java/ai/openclaw/wear/WearGatewayRepositoryTest.kt b/apps/android/wear/src/test/java/ai/openclaw/wear/WearGatewayRepositoryTest.kt index 7c203e060fe0..f7984b39f4b3 100644 --- a/apps/android/wear/src/test/java/ai/openclaw/wear/WearGatewayRepositoryTest.kt +++ b/apps/android/wear/src/test/java/ai/openclaw/wear/WearGatewayRepositoryTest.kt @@ -28,6 +28,127 @@ class WearGatewayRepositoryTest { assertFalse(shouldAcceptWearTalkSnapshot(WearRealtimeTalkSnapshot(), "attempt-current")) } + @Test + fun agentPulseRequiresNegotiatedCapabilityBeforeSendingItsRpc() = + runTest { + val requester = RecordingRequester { method, _ -> error("unexpected $method") } + val repository = WearGatewayRepository(requester) + + val failure = + runCatching { + repository.agentPulse( + expectedNodeId = "phone-a", + capabilities = emptySet(), + selectedSessionKey = "agent:main", + ) + }.exceptionOrNull() + + assertEquals("unsupported_peer", (failure as? WearProxyException)?.code) + assertTrue(requester.calls.isEmpty()) + } + + @Test + fun agentPulseParsesOnlyAggregateCountsAndRequiresThePreferredPhone() = + runTest { + val requester = + RecordingRequester { method, _ -> + assertEquals(WearRpcMethod.AgentPulse, method) + json.parseToJsonElement( + """{"tasks":{"state":"ready","scope":"bounded","queued":2,"running":3,"completed":5,"failed":1,"activeAtLimit":false,"recentAtLimit":true},"swarm":{"state":"active","scope":"selected-session","groups":2,"running":4,"done":6,"failed":1,"phases":[{"queued":1,"running":2,"done":3,"failed":0,"hidden":4}],"morePhases":false},"approvals":{"state":"ready","pending":2}}""", + ) + } + val repository = WearGatewayRepository(requester) + + val pulse = + repository.agentPulse( + expectedNodeId = "phone-a", + capabilities = setOf(WearProxyCapability.AgentPulse), + selectedSessionKey = "agent:main", + ) + + assertEquals(WearAgentPulseTaskState.Ready, pulse.tasks.state) + assertEquals(2, pulse.tasks.queued) + assertEquals(3, pulse.tasks.running) + assertEquals(5, pulse.tasks.completed) + assertEquals(1, pulse.tasks.failed) + assertEquals(false, pulse.tasks.activeAtLimit) + assertEquals(true, pulse.tasks.recentAtLimit) + assertEquals(WearAgentPulseSwarmState.Active, pulse.swarm.state) + assertEquals(2, pulse.swarm.groups) + assertEquals(4, pulse.swarm.running) + assertEquals(6, pulse.swarm.done) + assertEquals(1, pulse.swarm.failed) + assertEquals(WearAgentPulsePhase(1, 2, 3, 0, 4), pulse.swarm.phases.single()) + assertEquals(false, pulse.swarm.morePhases) + assertEquals(WearAgentPulseApprovalsState.Ready, pulse.approvals.state) + assertEquals(2, pulse.approvals.pending) + assertEquals(7L, pulse.eventSequence) + assertEquals("phone-a", pulse.phoneNodeId) + assertEquals( + json.parseToJsonElement("""{"sessionKey":"agent:main"}""").jsonObject, + requester.calls.single().second, + ) + assertEquals("phone-a", requester.expectedNodeIds.single()) + assertTrue(requester.requirePreferredNodes.single()) + } + + @Test + fun agentPulseKeepsUnavailableIdleAndRefreshingDistinctWithoutInventingZeroes() = + runTest { + val requester = + RecordingRequester { _, _ -> + json.parseToJsonElement( + """{"tasks":{"state":"unavailable"},"swarm":{"state":"idle","scope":"selected-session"},"approvals":{"state":"refreshing"}}""", + ) + } + + val pulse = + WearGatewayRepository(requester).agentPulse( + expectedNodeId = "phone-a", + capabilities = setOf(WearProxyCapability.AgentPulse), + selectedSessionKey = " ", + ) + + assertEquals(WearAgentPulseTaskState.Unavailable, pulse.tasks.state) + assertNull(pulse.tasks.running) + assertEquals(WearAgentPulseSwarmState.Idle, pulse.swarm.state) + assertNull(pulse.swarm.groups) + assertEquals(WearAgentPulseApprovalsState.Refreshing, pulse.approvals.state) + assertNull(pulse.approvals.pending) + assertTrue( + requester.calls + .single() + .second + .isEmpty(), + ) + } + + @Test + fun agentPulseRejectsUnknownStatesNegativeCountsAndOversizedPhaseLists() = + runTest { + val invalidPayloads = + listOf( + """{"tasks":{"state":"future"},"swarm":{"state":"unavailable"},"approvals":{"state":"unavailable"}}""", + """{"tasks":{"state":"ready","scope":"bounded","queued":-1,"running":0,"completed":0,"failed":0,"activeAtLimit":false,"recentAtLimit":false},"swarm":{"state":"unavailable"},"approvals":{"state":"unavailable"}}""", + """{"tasks":{"state":"unavailable"},"swarm":{"state":"active","scope":"selected-session","groups":1,"running":0,"done":0,"failed":0,"phases":[{"queued":0,"running":0,"done":0,"failed":0,"hidden":-1}],"morePhases":false},"approvals":{"state":"unavailable"}}""", + """{"tasks":{"state":"unavailable"},"swarm":{"state":"unavailable"},"approvals":{"state":"ready","pending":-1}}""", + """{"tasks":{"state":"unavailable"},"swarm":{"state":"active","scope":"selected-session","groups":1,"running":0,"done":0,"failed":0,"phases":[{},{},{},{},{},{},{},{},{}],"morePhases":true},"approvals":{"state":"unavailable"}}""", + ) + + invalidPayloads.forEach { payload -> + val requester = RecordingRequester { _, _ -> json.parseToJsonElement(payload) } + val failure = + runCatching { + WearGatewayRepository(requester).agentPulse( + expectedNodeId = "phone-a", + capabilities = setOf(WearProxyCapability.AgentPulse), + ) + }.exceptionOrNull() + + assertEquals("invalid_response", (failure as? WearProxyException)?.code) + } + } + @Test fun sessionsAndHistoryParseOnlyProjectedContract() = runTest { @@ -86,7 +207,7 @@ class WearGatewayRepositoryTest { WearRpcMethod.AgentsSelect -> JsonObject(emptyMap()) WearRpcMethod.GatewayDisconnect -> json.parseToJsonElement( - """{"connected":false,"status":"Offline","activeAgentId":"main","selectedModelRef":"openai/gpt-test","capabilities":["agent-controls","gateway-controls","model-controls","session-selection-lookup","attempt-scoped-realtime-audio"]}""", + """{"connected":false,"status":"Offline","activeAgentId":"main","selectedModelRef":"openai/gpt-test","capabilities":["agent-controls","gateway-controls","model-controls","session-selection-lookup","agent-pulse","attempt-scoped-realtime-audio"]}""", ) else -> error("unexpected $method") } @@ -154,7 +275,7 @@ class WearGatewayRepositoryTest { val requester = RecordingRequester { _, _ -> json.parseToJsonElement( - """{"connected":true,"status":"Connected","capabilities":["agent-controls","future-capability","gateway-controls","model-controls","session-selection-lookup","attempt-scoped-realtime-audio"]}""", + """{"connected":true,"status":"Connected","capabilities":["agent-controls","future-capability","gateway-controls","model-controls","session-selection-lookup","agent-pulse","attempt-scoped-realtime-audio"]}""", ) } diff --git a/apps/android/wear/src/test/java/ai/openclaw/wear/WearLaunchIntentTest.kt b/apps/android/wear/src/test/java/ai/openclaw/wear/WearLaunchIntentTest.kt index dd960e037303..35e2fa7d5e7f 100644 --- a/apps/android/wear/src/test/java/ai/openclaw/wear/WearLaunchIntentTest.kt +++ b/apps/android/wear/src/test/java/ai/openclaw/wear/WearLaunchIntentTest.kt @@ -34,6 +34,26 @@ class WearLaunchIntentTest { ) } + @Test + fun olderPhoneKeepsOriginalHomePages() { + assertEquals( + listOf(WearHomePage.Chat, WearHomePage.Voice, WearHomePage.Controls), + wearHomePages(agentPulseSupported = false), + ) + } + + @Test + fun capablePhoneAppendsPulseWithoutChangingExistingHomePageOrdinals() { + assertEquals(0, WearHomePage.Chat.ordinal) + assertEquals(1, WearHomePage.Voice.ordinal) + assertEquals(2, WearHomePage.Controls.ordinal) + assertEquals(3, WearHomePage.Pulse.ordinal) + assertEquals( + listOf(WearHomePage.Chat, WearHomePage.Voice, WearHomePage.Controls, WearHomePage.Pulse), + wearHomePages(agentPulseSupported = true), + ) + } + @Test fun tileTalkLaunchStartsOnVoice() { val target = diff --git a/apps/android/wear/src/test/java/ai/openclaw/wear/WearProxyClientTest.kt b/apps/android/wear/src/test/java/ai/openclaw/wear/WearProxyClientTest.kt index 327456f6e65c..f67070c2edb5 100644 --- a/apps/android/wear/src/test/java/ai/openclaw/wear/WearProxyClientTest.kt +++ b/apps/android/wear/src/test/java/ai/openclaw/wear/WearProxyClientTest.kt @@ -697,6 +697,26 @@ class WearProxyClientTest { assertTrue(tracker.isResponseCurrent(newerRequest, "stream", 10)) } + @Test + fun readOnlyRpcCursorDoesNotInvalidateAnOverlappingModelRequest() { + val tracker = WearEventSequenceTracker() + + tracker.adoptSnapshot("stream", 10) + val pulseBeforeModel = tracker.beginReadOnlyResponseRequest() + val modelRequest = tracker.beginResponseRequest() + val pulseRequest = tracker.beginReadOnlyResponseRequest() + tracker.beginReadOnlyResponseRequest() + + assertTrue(tracker.isReadOnlyResponseCurrent(pulseBeforeModel, "stream", 10)) + assertTrue(tracker.isResponseCurrent(modelRequest, "stream", 10)) + assertTrue(tracker.isReadOnlyResponseCurrent(pulseRequest, "stream", 10)) + + val staleLegacyPulse = tracker.beginReadOnlyResponseRequest() + assertEquals(WearSequenceDecision.Accepted, tracker.accept("stream", 11)) + assertFalse(tracker.isReadOnlyResponseCurrent(staleLegacyPulse, null, null)) + assertFalse(tracker.isReadOnlyResponseCurrent(pulseRequest, "stream", 10)) + } + @Test fun resyncBufferReplaysEventsNewerThanTheSnapshotWatermark() { val tracker = WearEventSequenceTracker() diff --git a/apps/android/wear/src/test/java/ai/openclaw/wear/WearViewModelLifecycleTest.kt b/apps/android/wear/src/test/java/ai/openclaw/wear/WearViewModelLifecycleTest.kt index c50ae71424dd..cd49f016d2e7 100644 --- a/apps/android/wear/src/test/java/ai/openclaw/wear/WearViewModelLifecycleTest.kt +++ b/apps/android/wear/src/test/java/ai/openclaw/wear/WearViewModelLifecycleTest.kt @@ -1,5 +1,6 @@ package ai.openclaw.wear +import ai.openclaw.wear.shared.WearProxyCapability import androidx.lifecycle.ViewModelProvider import androidx.lifecycle.ViewModelStore import androidx.lifecycle.ViewModelStoreOwner @@ -7,6 +8,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import org.junit.Assert.assertFalse import org.junit.Assert.assertNotSame +import org.junit.Assert.assertNull import org.junit.Assert.assertTrue import org.junit.Test import org.junit.runner.RunWith @@ -39,6 +41,110 @@ class WearViewModelLifecycleTest { } } + @Test + fun agentPulsePollingRequiresVisibleConnectedCapablePreferredRoute() { + val connected = + WearUiState( + loading = false, + connected = true, + phoneNodeId = "phone-a", + proxyCapabilities = setOf(WearProxyCapability.AgentPulse), + ) + + assertTrue(shouldPollAgentPulse(connected, pulseVisible = true)) + assertFalse(shouldPollAgentPulse(connected, pulseVisible = false)) + assertFalse(shouldPollAgentPulse(connected.copy(connected = false), pulseVisible = true)) + assertFalse(shouldPollAgentPulse(connected.copy(phoneNodeId = null), pulseVisible = true)) + assertFalse( + shouldPollAgentPulse( + connected.copy(proxyCapabilities = emptySet()), + pulseVisible = true, + ), + ) + } + + @Test + fun agentPulseRouteRejectsPhoneAgentSessionAndGenerationChanges() { + val session = + WearSession( + key = "agent:main:one", + title = "One", + updatedAt = null, + hasActiveRun = false, + phoneNodeId = "phone-a", + agentId = "main", + ) + val current = + WearUiState( + loading = false, + connected = true, + phoneNodeId = "phone-a", + activeAgentId = "main", + proxyCapabilities = setOf(WearProxyCapability.AgentPulse), + selectedSession = session, + ) + + fun accepts( + state: WearUiState = current, + routeGeneration: Long = 7L, + requestGeneration: Long = 11L, + ): Boolean = + wearAgentPulseRouteIsCurrent( + requestedPhoneNodeId = "phone-a", + requestedAgentId = "main", + requestedSessionKey = session.key, + requestedRouteGeneration = 7L, + currentRouteGeneration = routeGeneration, + requestedGeneration = 11L, + currentGeneration = requestGeneration, + pulseVisible = true, + state = state, + ) + + assertTrue(accepts()) + assertFalse(accepts(current.copy(phoneNodeId = "phone-b"))) + assertFalse(accepts(current.copy(activeAgentId = "secondary"))) + assertFalse(accepts(current.copy(selectedSession = session.copy(key = "agent:main:two")))) + assertFalse(accepts(routeGeneration = 8L)) + assertFalse(accepts(requestGeneration = 12L)) + } + + @Test + fun hidingAgentPulseCancelsTheSinglePoller() { + val app = RuntimeEnvironment.getApplication() as WearApplication + val factory = ViewModelProvider.AndroidViewModelFactory.getInstance(app) + val owner = TestViewModelStoreOwner() + val viewModel = ViewModelProvider(owner, factory)[WearViewModel::class.java] + val pollJob = Job() + try { + viewModel.setAgentPulseVisibleForTest(true) + viewModel.setAgentPulsePollJobForTest(pollJob) + + viewModel.setAgentPulseVisible(false) + + assertFalse(pollJob.isActive) + assertNull(viewModel.agentPulsePollJobForTest()) + assertFalse(viewModel.state.value.agentPulseLoading) + } finally { + owner.viewModelStore.clear() + } + } + + @Test + fun clearingViewModelCancelsTheAgentPulsePoller() { + val app = RuntimeEnvironment.getApplication() as WearApplication + val factory = ViewModelProvider.AndroidViewModelFactory.getInstance(app) + val owner = TestViewModelStoreOwner() + val viewModel = ViewModelProvider(owner, factory)[WearViewModel::class.java] + val pollJob = Job() + viewModel.setAgentPulsePollJobForTest(pollJob) + + owner.viewModelStore.clear() + + assertFalse(pollJob.isActive) + assertNull(viewModel.agentPulsePollJobForTest()) + } + private class TestViewModelStoreOwner : ViewModelStoreOwner { override val viewModelStore = ViewModelStore() } @@ -49,6 +155,26 @@ class WearViewModelLifecycleTest { get(this@realtimeTalkClientForTest) as WearRealtimeTalkClient } + private fun WearViewModel.setAgentPulseVisibleForTest(visible: Boolean) { + javaClass.getDeclaredField("agentPulseVisible").run { + isAccessible = true + setBoolean(this@setAgentPulseVisibleForTest, visible) + } + } + + private fun WearViewModel.setAgentPulsePollJobForTest(job: Job?) { + javaClass.getDeclaredField("agentPulsePollJob").run { + isAccessible = true + set(this@setAgentPulsePollJobForTest, job) + } + } + + private fun WearViewModel.agentPulsePollJobForTest(): Job? = + javaClass.getDeclaredField("agentPulsePollJob").run { + isAccessible = true + get(this@agentPulsePollJobForTest) as? Job + } + private fun WearRealtimeTalkClient.scopeForTest(): CoroutineScope = javaClass.getDeclaredField("scope").run { isAccessible = true