Skip to content

Commit 8229f62

Browse files
ashishpatel26steipete
ashishpatel26
and
steipete
authored
fix(android): queue node events until gateway connect (#92602)
* fix(android): queue notification events until gateway ready Co-authored-by: Ashish Patel <[email protected]> * style(android): format notification gateway changes * style(android): fix gateway resolver formatting * chore(android): sync native i18n inventory --------- Co-authored-by: Peter Steinberger <[email protected]>
1 parent 32762da commit 8229f62

7 files changed

Lines changed: 868 additions & 57 deletions

File tree

‎apps/.i18n/native-source.json‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -67,31 +67,31 @@
6767
},
6868
{
6969
"kind": "ui-state-text",
70-
"line": 360,
70+
"line": 496,
7171
"path": "apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt",
7272
"source": "Offline",
7373
"surface": "android",
7474
"id": "native.android.c65b61de70a063e7"
7575
},
7676
{
7777
"kind": "conditional-branch",
78-
"line": 1773,
78+
"line": 1981,
7979
"path": "apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt",
8080
"source": "Failed: this host requires wss:// or Tailscale Serve. No TLS endpoint detected.",
8181
"surface": "android",
8282
"id": "native.android.81009c40eed2216d"
8383
},
8484
{
8585
"kind": "conditional-branch",
86-
"line": 1775,
86+
"line": 1983,
8787
"path": "apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt",
8888
"source": "Failed: secure endpoint reached, but TLS fingerprint verification timed out. Check Tailscale Serve or gateway TLS and retry.",
8989
"surface": "android",
9090
"id": "native.android.467899bb510b8e34"
9191
},
9292
{
9393
"kind": "conditional-branch",
94-
"line": 1777,
94+
"line": 1985,
9595
"path": "apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt",
9696
"source": "Failed: couldn't reach the secure gateway endpoint for this host.",
9797
"surface": "android",
@@ -259,15 +259,15 @@
259259
},
260260
{
261261
"kind": "conditional-branch",
262-
"line": 1054,
262+
"line": 1093,
263263
"path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
264264
"source": "Connecting…",
265265
"surface": "android",
266266
"id": "native.android.1cbaba9cbffb2f97"
267267
},
268268
{
269269
"kind": "conditional-branch",
270-
"line": 1054,
270+
"line": 1093,
271271
"path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
272272
"source": "Reconnecting…",
273273
"surface": "android",

‎apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt‎

Lines changed: 218 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import ai.openclaw.app.gateway.GatewaySession
1313
import ai.openclaw.app.gateway.GatewayTlsProbeFailure
1414
import ai.openclaw.app.gateway.GatewayTlsProbeResult
1515
import ai.openclaw.app.gateway.GatewayUpdateAvailableSummary
16+
import ai.openclaw.app.gateway.NodeEventSendOutcome
1617
import ai.openclaw.app.gateway.normalizeGatewayTlsFingerprint
1718
import ai.openclaw.app.gateway.parseChatSendAck
1819
import ai.openclaw.app.gateway.probeGatewayTlsFingerprint
@@ -59,6 +60,7 @@ import androidx.core.content.ContextCompat
5960
import kotlinx.coroutines.CoroutineScope
6061
import kotlinx.coroutines.Dispatchers
6162
import kotlinx.coroutines.SupervisorJob
63+
import kotlinx.coroutines.channels.Channel
6264
import kotlinx.coroutines.delay
6365
import kotlinx.coroutines.flow.MutableStateFlow
6466
import kotlinx.coroutines.flow.StateFlow
@@ -79,6 +81,140 @@ import java.util.UUID
7981
import java.util.concurrent.ConcurrentHashMap
8082
import java.util.concurrent.atomic.AtomicLong
8183

84+
private const val MAX_PENDING_NOTIFICATION_EVENTS = 128
85+
86+
internal data class PendingNotificationNodeEvent(
87+
val event: String,
88+
val payloadJson: String?,
89+
)
90+
91+
private data class QueuedNotificationNodeEvent(
92+
val generation: Long,
93+
val event: PendingNotificationNodeEvent,
94+
)
95+
96+
internal class NotificationNodeEventOutbox(
97+
private val capacity: Int = MAX_PENDING_NOTIFICATION_EVENTS,
98+
private val isAuthorized: (PendingNotificationNodeEvent) -> Boolean = { true },
99+
private val isConnected: () -> Boolean = { true },
100+
private val deliveryIntervalMs: () -> Long = { 0L },
101+
private val nowEpochMs: () -> Long = System::currentTimeMillis,
102+
private val sleep: suspend (Long) -> Unit = { delay(it) },
103+
private val invalidateConnection: () -> Unit = {},
104+
private val send: suspend (PendingNotificationNodeEvent) -> NodeEventSendOutcome,
105+
) {
106+
private val stateLock = Any()
107+
private val generation = AtomicLong()
108+
private val lastDeliveryAtMs = AtomicLong(-1L)
109+
private val pending = ArrayDeque<QueuedNotificationNodeEvent>(capacity)
110+
private val wakeDelivery = Channel<Unit>(Channel.CONFLATED)
111+
private var inFlight: QueuedNotificationNodeEvent? = null
112+
113+
init {
114+
require(capacity > 0) { "capacity must be positive" }
115+
}
116+
117+
fun enqueue(event: PendingNotificationNodeEvent) {
118+
synchronized(stateLock) {
119+
if (pending.size == capacity) pending.removeFirst()
120+
pending.addLast(QueuedNotificationNodeEvent(generation = generation.get(), event = event))
121+
}
122+
wakeDelivery.trySend(Unit)
123+
}
124+
125+
fun clear() {
126+
synchronized(stateLock) {
127+
clearLocked()
128+
}
129+
wakeDelivery.trySend(Unit)
130+
}
131+
132+
fun <T> updatePolicy(update: () -> T): T {
133+
val result =
134+
synchronized(stateLock) {
135+
// Admission checks share this lock, so the new policy is visible before the next generation.
136+
update().also { clearLocked() }
137+
}
138+
wakeDelivery.trySend(Unit)
139+
return result
140+
}
141+
142+
fun onConnected() {
143+
wakeDelivery.trySend(Unit)
144+
}
145+
146+
suspend fun deliver() {
147+
while (true) {
148+
wakeDelivery.receive()
149+
while (true) {
150+
val queued = synchronized(stateLock) { pending.firstOrNull() } ?: break
151+
if (queued.generation != generation.get() || !isAuthorized(queued.event)) {
152+
synchronized(stateLock) {
153+
if (pending.firstOrNull() === queued) pending.removeFirst()
154+
}
155+
continue
156+
}
157+
if (!isConnected()) break
158+
if (!awaitDeliverySlot(queued)) continue
159+
val admitted =
160+
synchronized(stateLock) {
161+
if (
162+
pending.firstOrNull() !== queued ||
163+
queued.generation != generation.get() ||
164+
!isAuthorized(queued.event) ||
165+
!isConnected()
166+
) {
167+
false
168+
} else {
169+
pending.removeFirst()
170+
inFlight = queued
171+
true
172+
}
173+
}
174+
if (!admitted) continue
175+
176+
val outcome = send(queued.event)
177+
synchronized(stateLock) {
178+
if (inFlight === queued) inFlight = null
179+
if (queued.generation == generation.get() && isAuthorized(queued.event)) {
180+
when (outcome) {
181+
NodeEventSendOutcome.COMPLETED -> lastDeliveryAtMs.set(nowEpochMs())
182+
NodeEventSendOutcome.DISCONNECTED -> {
183+
// This outcome is rejected before send, so it is safe to retain for reconnect.
184+
if (pending.size == capacity) pending.removeLast()
185+
pending.addFirst(queued)
186+
}
187+
// Ambiguous failures may have reached the gateway: do not retry, but charge their rate slot.
188+
NodeEventSendOutcome.FAILED -> lastDeliveryAtMs.set(nowEpochMs())
189+
}
190+
}
191+
}
192+
if (outcome == NodeEventSendOutcome.DISCONNECTED) break
193+
}
194+
}
195+
}
196+
197+
private suspend fun awaitDeliverySlot(queued: QueuedNotificationNodeEvent): Boolean {
198+
while (queued.generation == generation.get() && isAuthorized(queued.event)) {
199+
val lastDelivery = lastDeliveryAtMs.get()
200+
if (lastDelivery < 0L) return true
201+
val waitMs = lastDelivery + deliveryIntervalMs().coerceAtLeast(0L) - nowEpochMs()
202+
if (waitMs <= 0L) return true
203+
// Short slices make policy/gateway invalidation responsive without charging stale quota.
204+
sleep(minOf(waitMs, 250L))
205+
}
206+
return false
207+
}
208+
209+
private fun clearLocked() {
210+
// Only an admitted RPC needs transport invalidation; queued payloads have no socket side effect.
211+
if (inFlight?.generation == generation.get()) invalidateConnection()
212+
generation.incrementAndGet()
213+
lastDeliveryAtMs.set(-1L)
214+
pending.clear()
215+
}
216+
}
217+
82218
/**
83219
* Process runtime that owns gateway sessions, node command handlers, capture managers, and UI-facing state.
84220
*/
@@ -591,6 +727,7 @@ class NodeRuntime(
591727
_nodeConnected.value = true
592728
nodeStatusText = "Connected"
593729
}
730+
notificationOutbox.onConnected()
594731
showLocalCanvasOnConnect()
595732
publishNodePresenceAliveBeacon(NodePresenceAliveBeacon.Trigger.Connect)
596733
val endpoint = connectedEndpoint
@@ -628,11 +765,49 @@ class NodeRuntime(
628765
},
629766
)
630767

768+
private val notificationOutbox: NotificationNodeEventOutbox by lazy {
769+
NotificationNodeEventOutbox(
770+
isAuthorized = ::isNotificationEventStillAuthorized,
771+
isConnected = nodeSession::isReady,
772+
deliveryIntervalMs = ::notificationDeliveryIntervalMs,
773+
invalidateConnection = nodeSession::reconnect,
774+
send = { pending ->
775+
nodeSession.sendNodeEventWithOutcome(event = pending.event, payloadJson = pending.payloadJson)
776+
},
777+
)
778+
}
779+
780+
private fun notificationDeliveryIntervalMs(): Long {
781+
val maxEvents =
782+
prefs.notificationForwardingMaxEventsPerMinute.value
783+
.coerceAtLeast(1)
784+
.toLong()
785+
return (60_000L + maxEvents - 1L) / maxEvents
786+
}
787+
788+
private fun isNotificationEventStillAuthorized(event: PendingNotificationNodeEvent): Boolean {
789+
if (event.event != "notifications.changed") return false
790+
if (!DeviceNotificationListenerService.isAccessEnabled(appContext)) return false
791+
val payload =
792+
runCatching { event.payloadJson?.let(json::parseToJsonElement).asObjectOrNull() }
793+
.getOrNull()
794+
?: return false
795+
val packageName = payload["packageName"].asStringOrNull()?.trim().orEmpty()
796+
if (packageName.isEmpty()) return false
797+
val policy = prefs.getNotificationForwardingPolicy(appPackageName = appContext.packageName)
798+
val eventSessionKey = payload["sessionKey"].asStringOrNull()?.trim()?.ifEmpty { null }
799+
return policy.enabled &&
800+
policy.sessionKey == eventSessionKey &&
801+
policy.allowsPackage(packageName) &&
802+
!policy.isWithinQuietHours(nowEpochMs = System.currentTimeMillis())
803+
}
804+
631805
init {
806+
scope.launch { notificationOutbox.deliver() }
632807
DeviceNotificationListenerService.setNodeEventSink { event, payloadJson ->
633-
scope.launch {
634-
nodeSession.sendNodeEvent(event = event, payloadJson = payloadJson)
635-
}
808+
notificationOutbox.enqueue(
809+
PendingNotificationNodeEvent(event = event, payloadJson = payloadJson),
810+
)
636811
}
637812
}
638813

@@ -1267,29 +1442,60 @@ class NodeRuntime(
12671442
}
12681443

12691444
fun setNotificationForwardingEnabled(value: Boolean) {
1270-
prefs.setNotificationForwardingEnabled(value)
1445+
if (prefs.notificationForwardingEnabled.value == value) return
1446+
notificationOutbox.updatePolicy { prefs.setNotificationForwardingEnabled(value) }
12711447
}
12721448

12731449
fun setNotificationForwardingMode(mode: NotificationPackageFilterMode) {
1274-
prefs.setNotificationForwardingMode(mode)
1450+
if (prefs.notificationForwardingMode.value == mode) return
1451+
notificationOutbox.updatePolicy { prefs.setNotificationForwardingMode(mode) }
12751452
}
12761453

12771454
fun setNotificationForwardingPackages(packages: List<String>) {
1278-
prefs.setNotificationForwardingPackages(packages)
1455+
val normalized = packages.map(String::trim).filter(String::isNotEmpty).toSet()
1456+
if (prefs.notificationForwardingPackages.value == normalized) return
1457+
notificationOutbox.updatePolicy { prefs.setNotificationForwardingPackages(normalized.toList()) }
12791458
}
12801459

12811460
fun setNotificationForwardingQuietHours(
12821461
enabled: Boolean,
12831462
start: String,
12841463
end: String,
1285-
): Boolean = prefs.setNotificationForwardingQuietHours(enabled = enabled, start = start, end = end)
1464+
): Boolean {
1465+
if (!enabled) {
1466+
if (!prefs.notificationForwardingQuietHoursEnabled.value) return true
1467+
return notificationOutbox.updatePolicy {
1468+
prefs.setNotificationForwardingQuietHours(enabled = false, start = start, end = end)
1469+
}
1470+
}
1471+
val normalizedStart = normalizeLocalHourMinute(start) ?: return false
1472+
val normalizedEnd = normalizeLocalHourMinute(end) ?: return false
1473+
val unchanged =
1474+
prefs.notificationForwardingQuietHoursEnabled.value &&
1475+
prefs.notificationForwardingQuietStart.value == normalizedStart &&
1476+
prefs.notificationForwardingQuietEnd.value == normalizedEnd
1477+
if (unchanged) return true
1478+
return notificationOutbox.updatePolicy {
1479+
prefs.setNotificationForwardingQuietHours(
1480+
enabled = true,
1481+
start = normalizedStart,
1482+
end = normalizedEnd,
1483+
)
1484+
}
1485+
}
12861486

12871487
fun setNotificationForwardingMaxEventsPerMinute(value: Int) {
1288-
prefs.setNotificationForwardingMaxEventsPerMinute(value)
1488+
val normalized = value.coerceAtLeast(1)
1489+
if (prefs.notificationForwardingMaxEventsPerMinute.value == normalized) return
1490+
notificationOutbox.updatePolicy {
1491+
prefs.setNotificationForwardingMaxEventsPerMinute(normalized)
1492+
}
12891493
}
12901494

12911495
fun setNotificationForwardingSessionKey(value: String?) {
1292-
prefs.setNotificationForwardingSessionKey(value)
1496+
val normalized = value?.trim()?.takeIf(String::isNotEmpty)
1497+
if (prefs.notificationForwardingSessionKey.value == normalized) return
1498+
notificationOutbox.updatePolicy { prefs.setNotificationForwardingSessionKey(normalized) }
12931499
}
12941500

12951501
fun setVoiceScreenActive(active: Boolean) {
@@ -1675,6 +1881,8 @@ class NodeRuntime(
16751881
endpoint: GatewayEndpoint,
16761882
auth: GatewayConnectAuth,
16771883
) {
1884+
// A user-selected connect target must never inherit notification content from another gateway.
1885+
notificationOutbox.clear()
16781886
val connectAttemptId = connectAttemptSeq.incrementAndGet()
16791887
_pendingGatewayTrust.value = null
16801888
val tls = connectionManager.resolveTlsParams(endpoint)
@@ -1826,6 +2034,7 @@ class NodeRuntime(
18262034
}
18272035

18282036
fun disconnect() {
2037+
notificationOutbox.clear()
18292038
connectAttemptSeq.incrementAndGet()
18302039
stopActiveVoiceSession()
18312040
connectedEndpoint = null

0 commit comments

Comments
 (0)