[codex] Cancel Android gateway pending RPCs on close (#98067)

* fix: cancel Android gateway pending RPCs on close

* fix(android): isolate pending RPCs per connection

* style(android): separate RPC waiter invariant

* chore(android): align native i18n inventory

---------

Co-authored-by: NianJiuZst <180004567+users.noreply.github.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>

apps/.i18n/native-source.json

@@ -259,7 +259,7 @@
     },
     {
       "kind": "conditional-branch",
-      "line": 1037,
+      "line": 1054,
       "path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
       "source": "Connecting…",
       "surface": "android",
@@ -267,7 +267,7 @@
     },
     {
       "kind": "conditional-branch",
-      "line": 1037,
+      "line": 1054,
       "path": "apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt",
       "source": "Reconnecting…",
       "surface": "android",

apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt

@@ -180,7 +180,6 @@ class GatewaySession(
 
   private val json = Json { ignoreUnknownKeys = true }
   private val writeLock = Mutex()
-  private val pending = ConcurrentHashMap<String, CompletableDeferred<RpcResponse>>()
 
   @Volatile private var pluginSurfaceUrls: Map<String, String> = emptyMap()
 
@@ -389,6 +388,11 @@ class GatewaySession(
     private var socket: WebSocket? = null
     private val loggerTag = "OpenClawGateway"
     private val incomingMessages = Channel<String>(Channel.UNLIMITED)
+
+    // RPC waiters belong to this socket generation. Closing it must not touch a replacement connection.
+    private val pending = ConcurrentHashMap<String, CompletableDeferred<RpcResponse>>()
+
+    private val pendingLock = Any()
     private val messagePumpJob =
       scope.launch(Dispatchers.IO) {
         for (text in incomingMessages) {
@@ -424,19 +428,14 @@ class GatewaySession(
       timeoutMs: Long,
     ): RpcResponse {
       val id = UUID.randomUUID().toString()
-      val deferred = CompletableDeferred<RpcResponse>()
-      pending[id] = deferred
+      val deferred = registerPending(id)
       try {
         sendJson(buildRequestFrame(id = id, method = method, params = params))
-      } catch (err: Throwable) {
-        pending.remove(id)
-        throw err
-      }
-      return try {
-        withTimeout(timeoutMs) { deferred.await() }
+        return withTimeout(timeoutMs) { deferred.await() }
       } catch (err: TimeoutCancellationException) {
-        pending.remove(id)
         throw IllegalStateException("request timeout")
+      } finally {
+        pending.remove(id)
       }
     }
 
@@ -447,8 +446,7 @@ class GatewaySession(
       onError: (ErrorShape) -> Unit,
     ) {
       val id = UUID.randomUUID().toString()
-      val deferred = CompletableDeferred<RpcResponse>()
-      pending[id] = deferred
+      val deferred = registerPending(id)
       try {
         sendJson(buildRequestFrame(id = id, method = method, params = params))
       } catch (err: Throwable) {
@@ -456,20 +454,35 @@ class GatewaySession(
         throw err
       }
       scope.launch(Dispatchers.IO) {
-        val response =
-          try {
-            withTimeout(timeoutMs) { deferred.await() }
-          } catch (_: TimeoutCancellationException) {
-            pending.remove(id)
-            onError(ErrorShape("UNAVAILABLE", "request timeout"))
-            return@launch
+        try {
+          val response =
+            try {
+              withTimeout(timeoutMs) { deferred.await() }
+            } catch (_: TimeoutCancellationException) {
+              onError(ErrorShape("UNAVAILABLE", "request timeout"))
+              return@launch
+            } catch (_: CancellationException) {
+              return@launch
+            }
+          if (!response.ok) {
+            onError(response.error ?: ErrorShape("UNAVAILABLE", "request failed"))
           }
-        if (!response.ok) {
-          onError(response.error ?: ErrorShape("UNAVAILABLE", "request failed"))
+        } finally {
+          pending.remove(id)
         }
       }
     }
 
+    private fun registerPending(id: String): CompletableDeferred<RpcResponse> {
+      val deferred = CompletableDeferred<RpcResponse>()
+      // Registration and the close drain are one lifecycle decision; no waiter may slip between them.
+      synchronized(pendingLock) {
+        if (isClosed.get()) throw IllegalStateException("Gateway closed")
+        pending[id] = deferred
+      }
+      return deferred
+    }
+
     suspend fun sendJson(obj: JsonObject) {
       val jsonString = obj.toString()
       writeLock.withLock {
@@ -500,6 +513,7 @@ class GatewaySession(
         if (!connectDeferred.isCompleted) {
           connectDeferred.completeExceptionally(IllegalStateException("Gateway closed"))
         }
+        failPending()
         socket?.close(1000, "bye")
         socket = null
         closedDeferred.complete(Unit)
@@ -1011,10 +1025,13 @@ class GatewaySession(
     }
 
     private fun failPending() {
-      for ((_, waiter) in pending) {
+      val waiters =
+        synchronized(pendingLock) {
+          pending.values.toList().also { pending.clear() }
+        }
+      for (waiter in waiters) {
         waiter.cancel()
       }
-      pending.clear()
     }
   }
 

apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt

@@ -1,11 +1,14 @@
 package ai.openclaw.app.gateway
 
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.withTimeoutOrNull
@@ -125,6 +128,65 @@ class GatewaySessionInvokeTest {
       }
     }
 
+  @Test
+  fun disconnectCancelsPendingRpcWithoutWaitingForRequestTimeout() {
+    runBlocking {
+      val json = testJson()
+      val connected = CompletableDeferred<Unit>()
+      val slowRequestSeen = CompletableDeferred<Unit>()
+      val requestResult = CompletableDeferred<Result<GatewaySession.RpcResult>>()
+      val lastDisconnect = AtomicReference("")
+      val serverWebSocket = AtomicReference<WebSocket?>(null)
+      val server =
+        startGatewayServer(json) { webSocket, id, method, _ ->
+          serverWebSocket.set(webSocket)
+          when (method) {
+            "connect" -> webSocket.send(connectResponseFrame(id))
+            "slow.method" -> {
+              if (!slowRequestSeen.isCompleted) slowRequestSeen.complete(Unit)
+            }
+          }
+        }
+
+      val harness =
+        createNodeHarness(
+          connected = connected,
+          lastDisconnect = lastDisconnect,
+        ) { GatewaySession.InvokeResult.ok("""{"handled":true}""") }
+      var requestJob: Job? = null
+
+      try {
+        connectNodeSession(harness.session, server.port)
+        awaitConnectedOrThrow(connected, lastDisconnect, server)
+        requestJob =
+          launch {
+            requestResult.complete(
+              runCatching {
+                harness.session.requestDetailed("slow.method", null, timeoutMs = 30_000)
+              },
+            )
+          }
+        withTimeout(TEST_TIMEOUT_MS) { slowRequestSeen.await() }
+
+        harness.session.disconnect()
+
+        val result = withTimeout(2_000) { requestResult.await() }
+        assertEquals(true, result.exceptionOrNull() is CancellationException)
+        serverWebSocket.get()?.close(1000, "done")
+        withTimeoutOrNull(2_000) {
+          while (lastDisconnect.get().isEmpty()) delay(10)
+        }
+      } finally {
+        requestJob?.cancelAndJoin()
+        runCatching { serverWebSocket.get()?.close(1000, "done") }
+        delay(100)
+        harness.session.disconnect()
+        harness.sessionJob.cancelAndJoin()
+        server.shutdown()
+      }
+    }
+  }
+
   @Test
   fun eventsAreDispatchedInWebSocketFrameOrder() =
     runBlocking {

apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionReconnectTest.kt

@@ -5,9 +5,11 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.jsonPrimitive
@@ -28,6 +30,7 @@ import org.robolectric.RobolectricTestRunner
 import org.robolectric.RuntimeEnvironment
 import org.robolectric.annotation.Config
 import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicInteger
 
 private const val LIFECYCLE_TEST_TIMEOUT_MS = 8_000L
 private const val LIFECYCLE_CONNECT_CHALLENGE_FRAME =
@@ -79,6 +82,75 @@ private data class ReconnectServer(
 @RunWith(RobolectricTestRunner::class)
 @Config(sdk = [34])
 class GatewaySessionReconnectTest {
+  @Test
+  fun staleConnectionDrainCannotCancelReplacementRpc() =
+    runBlocking {
+      val json = Json { ignoreUnknownKeys = true }
+      val firstConnected = CompletableDeferred<Unit>()
+      val secondConnected = CompletableDeferred<Unit>()
+      val replacementRequest = CompletableDeferred<Pair<WebSocket, String>>()
+      val connectionCount = AtomicInteger(0)
+      val firstServer =
+        startGatewayServer(json = json) { webSocket, id, method ->
+          if (method == "connect") webSocket.send(connectResponseFrame(id))
+        }
+      val secondServer =
+        startGatewayServer(json = json) { webSocket, id, method ->
+          when (method) {
+            "connect" -> webSocket.send(connectResponseFrame(id))
+            "slow.method" -> replacementRequest.complete(webSocket to id)
+          }
+        }
+      val harness =
+        createReconnectHarness(
+          onConnected = {
+            when (connectionCount.incrementAndGet()) {
+              1 -> firstConnected.complete(Unit)
+              2 -> secondConnected.complete(Unit)
+            }
+          },
+        )
+
+      try {
+        connectNodeSession(harness.session, firstServer.port)
+        withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { firstConnected.await() }
+        val oldConnection = readField<Any>(harness.session, "currentConnection")
+
+        connectNodeSession(harness.session, secondServer.port)
+        withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { secondConnected.await() }
+        val newRequest =
+          async {
+            harness.session.requestDetailed("slow.method", null, timeoutMs = 30_000)
+          }
+        val (replacementSocket, requestId) =
+          withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { replacementRequest.await() }
+
+        val failPending = oldConnection.javaClass.getDeclaredMethod("failPending")
+        failPending.isAccessible = true
+        failPending.invoke(oldConnection)
+
+        assertNull(withTimeoutOrNull(200) { newRequest.await() })
+        replacementSocket.send(
+          """{"type":"res","id":"$requestId","ok":true,"payload":{"connection":2}}""",
+        )
+        val newResult = withTimeout(LIFECYCLE_TEST_TIMEOUT_MS) { newRequest.await() }
+        assertTrue(newResult.ok)
+        assertEquals("""{"connection":2}""", newResult.payloadJson)
+      } finally {
+        shutdownReconnectHarness(harness, firstServer, secondServer)
+      }
+    }
+
+  @Suppress("UNCHECKED_CAST")
+  private fun <T> readField(
+    target: Any,
+    name: String,
+  ): T {
+    val field = target.javaClass.getDeclaredField(name)
+    field.isAccessible = true
+    return field.get(target) as T
+  }
+
   @Test
   fun connectToNewGatewayClosesActiveConnectionAndStartsReplacement() =
     runBlocking {
@@ -366,6 +438,7 @@ class GatewaySessionReconnectTest {
     }
 
   private fun createReconnectHarness(
+    onConnected: () -> Unit = {},
     onConnectFailure: (GatewaySession.ErrorShape, Boolean) -> Unit = { _, _ -> },
   ): ReconnectHarness {
     val app = RuntimeEnvironment.getApplication()
@@ -375,7 +448,7 @@ class GatewaySessionReconnectTest {
         scope = CoroutineScope(sessionJob + Dispatchers.Default),
         identityStore = DeviceIdentityStore(app),
         deviceAuthStore = ReconnectDeviceAuthStore(),
-        onConnected = {},
+        onConnected = { onConnected() },
         onDisconnected = { _ -> },
         onConnectFailure = onConnectFailure,
         onEvent = { _, _ -> },