fix(ws): align RPC timeout to 120s + rejectAllPending on disconnect (fixes #526) (#541)

* fix(ws): align RPC timeout to 120s + rejectAllPending on disconnect (fixes #526)

Mobile's pending-request handling diverged from the desktop contract
(apps/shared JsonRpcGatewayClient) in two ways that caused silent hangs:

1. Timeout was 30s while the desktop uses 120s (DEFAULT_REQUEST_TIMEOUT_MS),
   so legitimately long agent turns were pruned early. Align to 120s and use
   it for both the cleanup sweep and sendRpcAndAwait's per-call withTimeout.

2. The cleanup sweep only Log.w'd stale requests and dropped them — it never
   failed the CompletableDeferred, so sendRpcAndAwait callers awaiting a
   server that never answered HUNG FOREVER. And on disconnect/reconnect there
   was NO rejectAllPending, so in-flight awaited RPCs also hung until their
   own timeout.

Changes (all in ChatViewModel.kt):
- cleanup sweep now completeExceptionally()s the deferred AND surfaces a
  'Request timed out' error for fire-and-forget RPCs (was silently dropped)
- new rejectAllPending() fails + clears every in-flight pending deferred;
  wired into the disconnect-status collector and reconnect()
- PENDING_REQUEST_TIMEOUT_MS = 120_000 (matches desktop); sendRpcAndAwait
  uses it instead of a hardcoded 30s

Tests: 2 new ChatViewModelTest cases cover rejectAllPending on disconnect
and on reconnect (no hang, VM stays usable). (Timeout-surfacing path shares
the deferred-fail logic with handleRpcError; the 120s sweep can't run under
the unit-test dispatcher's periodic-timer limitation, noted in PR.)

Mirrors the desktop-app audit in /opt/hermes-agent.

* fix(ws): aggregate pending-request cleanup into a single UI update (PR #541)

Sourcery review (PR #541): startPendingRequestCleanup emitted one
_uiState.update{} per stale request inside the loop, causing N separate
emissions / redundant recompositions when many RPCs time out at once.
Collect fire-and-forget methods and emit a single combined errorMessage
update after the loop.

rejectAllPending is intentionally left silent on disconnect/reconnect:
those are transient socket events (the RPC often completes on the new
socket), so surfacing a scary error banner is UX noise. Only the
timeout path surfaces errors — that is a genuine failure. This matches
the desktop JsonRpcGatewayClient.rejectAllPending semantics.

* refactor(ws): move awaited-RPC pending/timeout layer into HermesWsClient (closes #526)

Issue #526: port the JsonRpcGatewayClient.request() pattern into
HermesWsClient. HermesWsClient now owns the in-flight map, per-call
120s timeout, resolve-on-RpcResult/RpcError, and rejectAllPending on
disconnect/reconnect. ChatViewModel.sendRpcAndAwait simply delegates
to wsClient.request(...).await(); its local pendingRequests map,
PendingRequest, startPendingRequestCleanup loop and rejectAllPending
are removed. Fire-and-forget send() is unchanged.

---------

Co-authored-by: M57 <m57@hermes-control.app>

app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt

@@ -6,6 +6,7 @@ import com.m57.hermescontrol.BuildConfig
 import com.m57.hermescontrol.data.local.AuthManager
 import com.m57.hermescontrol.data.remote.NetworkMonitor
 import com.m57.hermescontrol.data.remote.OkHttpProvider
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -31,6 +32,7 @@ import okhttp3.RequestBody.Companion.toRequestBody
 import okhttp3.Response
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
@@ -241,6 +243,97 @@ object HermesWsClient {
         _connectionStatus.value = ConnectionStatus.DISCONNECTED
     }
 
+    // ── Awaited RPC request layer (issue #526) ─────────────────────────
+    // Mirrors desktop apps/shared JsonRpcGatewayClient.request(): an in-flight
+    // map with a per-call timeout and rejectAllPending on socket close, so a
+    // dropped RPC response can't leave a caller awaiting forever.
+
+    /** Thrown when an awaited [request] is neither answered nor rejected within [REQUEST_TIMEOUT_MS], or is rejected by a disconnect. */
+    class HermesRpcException(
+        message: String,
+    ) : Exception(message)
+
+    /**
+     * Default per-request timeout. Matches the desktop
+     * `apps/shared` `JsonRpcGatewayClient.DEFAULT_REQUEST_TIMEOUT_MS` (120s),
+     * so legitimately long agent turns are not pruned early.
+     */
+    const val REQUEST_TIMEOUT_MS: Long = 120_000L
+
+    /** A single in-flight [request] awaiting its RPC result/error. */
+    private data class PendingCall(
+        val method: String,
+        val deferred: CompletableDeferred<Any?>,
+        var timeoutJob: Job? = null,
+    )
+
+    /** Tracks in-flight [request] calls by their JSON-RPC id. */
+    private val pendingCalls = ConcurrentHashMap<String, PendingCall>()
+
+    /**
+     * Send a JSON-RPC request that expects a result and returns a
+     * [CompletableDeferred] for it — mirroring the desktop
+     * `JsonRpcGatewayClient.request()`. Fire-and-forget notifications should
+     * keep using [send].
+     *
+     * The deferred is completed on the matching [WsEvent.RpcResult] /
+     * [WsEvent.RpcError], rejected after [timeoutMs] (no response), or
+     * rejected by [rejectAllPending] when the socket closes.
+     */
+    fun request(
+        method: String,
+        params: Map<String, Any> = emptyMap(),
+        timeoutMs: Long = REQUEST_TIMEOUT_MS,
+    ): CompletableDeferred<Any?> {
+        val deferred = CompletableDeferred<Any?>()
+        val id =
+            send(method, params) { reqId ->
+                pendingCalls[reqId] = PendingCall(method, deferred)
+            }
+        // Arm the per-request timeout (fires if the server never answers).
+        pendingCalls[id]?.timeoutJob =
+            wsScope.launch {
+                delay(timeoutMs)
+                resolvePending(id, null, JsonRpcError(-1, "Request timed out: $method"))
+            }
+        return deferred
+    }
+
+    /** Complete (or fail) a single pending call and cancel its timer. */
+    private fun resolvePending(
+        id: String,
+        result: Any?,
+        error: JsonRpcError?,
+    ) {
+        val call = pendingCalls.remove(id) ?: return
+        call.timeoutJob?.cancel()
+        if (error != null) {
+            call.deferred.completeExceptionally(HermesRpcException(error.message))
+        } else {
+            call.deferred.complete(result)
+        }
+    }
+
+    /**
+     * Fail and clear every in-flight [request]. Called on disconnect /
+     * reconnect so callers awaiting a result don't hang across a socket
+     * close — mirrors desktop `JsonRpcGatewayClient.rejectAllPending(error)`
+     * invoked on socket close.
+     */
+    fun rejectAllPending(
+        error: HermesRpcException =
+            HermesRpcException("Connection lost — request cancelled"),
+    ) {
+        if (pendingCalls.isEmpty()) return
+        val snapshot = pendingCalls.toList()
+        pendingCalls.clear()
+        for ((id, call) in snapshot) {
+            Log.w(TAG, "Rejecting pending request on disconnect: ${call.method} (id=$id)")
+            call.timeoutJob?.cancel()
+            call.deferred.completeExceptionally(error)
+        }
+    }
+
     // ── Send helpers ─────────────────────────────────────────────────────
 
     /**
@@ -348,6 +441,21 @@ object HermesWsClient {
         ) {
             if (BuildConfig.DEBUG) Log.d(TAG, "← $text")
             lastPongTimestamp = System.currentTimeMillis()
+            // Resolve any in-flight `request()` awaiting this RPC result/error
+            // (issue #526) before fanning the parsed event out to collectors.
+            val event =
+                try {
+                    val rpc = OkHttpProvider.json.decodeFromString<JsonRpcResponse>(text)
+                    EventParser.parse(rpc, text)
+                } catch (e: Exception) {
+                    Log.e(TAG, "Failed to parse message", e)
+                    WsEvent.Unknown(text)
+                }
+            when (event) {
+                is WsEvent.RpcResult -> resolvePending(event.id, event.result, null)
+                is WsEvent.RpcError -> resolvePending(event.id, null, event.error)
+                else -> Unit
+            }
             val emitted = rawMessages.tryEmit(text)
             if (!emitted && BuildConfig.DEBUG) {
                 Log.w(TAG, "WebSocket message dropped due to buffer overflow")

app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt

@@ -20,7 +20,6 @@ import com.m57.hermescontrol.data.ws.HermesWsClient
 import com.m57.hermescontrol.data.ws.WsEvent
 import com.m57.hermescontrol.data.ws.WsMethods
 import com.m57.hermescontrol.data.ws.toJsonElement
-import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
@@ -33,7 +32,6 @@ import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
-import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.yield
 import kotlinx.serialization.json.decodeFromJsonElement
 import okhttp3.MediaType.Companion.toMediaType
@@ -45,17 +43,6 @@ import java.util.concurrent.ConcurrentHashMap
 
 private const val TAG = "ChatViewModel"
 
-/**
- * Pending request timeout — if the server doesn't respond within this
- * window, the request is pruned and an error is surfaced.
- */
-private const val PENDING_REQUEST_TIMEOUT_MS = 30_000L
-
-/** Thrown when an awaited RPC returns an error response. */
-private class RpcCallException(
-    message: String,
-) : Exception(message)
-
 data class ChatUiState(
     val messages: List<ChatMessage> = emptyList(),
     val currentSessionId: String? = null,
@@ -132,7 +119,6 @@ class ChatViewModel(
     val streamingState: StateFlow<StreamingState> = _streamingState.asStateFlow()
 
     private val wsClient = HermesWsClient
-    private val pendingRequests = ConcurrentHashMap<String, PendingRequest>()
 
     private val slashDispatcher = SlashCommandDispatcher()
     private val searchDelegate =
@@ -151,8 +137,6 @@ class ChatViewModel(
             isTestEnvironment = { isTestEnvironment() },
         )
 
-    private var pendingCleanupJob: Job? = null
-
     // ── Public state ─────────────────────────────────────────────────────
 
     /**
@@ -199,14 +183,12 @@ class ChatViewModel(
                     status == ConnectionStatus.AUTH_EXPIRED
                 ) {
                     _uiState.update { it.copy(isLoading = false) }
+                    // Fail any in-flight awaited RPCs so callers don't hang
+                    // across the disconnect (delegated to HermesWsClient, issue #526).
+                    wsClient.rejectAllPending()
                 }
             }
         }
-        if (startCleanup) {
-            startPendingRequestCleanup()
-        }
-
-        // B7 (Jun 30 2026, kanban t_connection_loading): if already connected on launch, trigger setup immediately
         if (wsClient.connectionStatus.value == ConnectionStatus.CONNECTED) {
             handleGatewayReady()
         }
@@ -404,12 +386,7 @@ class ChatViewModel(
         id: String,
         result: Any?,
     ) {
-        val pending = pendingRequests.remove(id) ?: return
+        val method = idToMethod.remove(id) ?: return
-        val method = pending.method
-
-        // Complete the deferred so sendRpcAndAwait() callers can resume
-        pending.deferred?.complete(result)
-
         when (method) {
             WsMethods.SESSION_CREATE -> {
                 val resultMap = result as? Map<String, Any?> ?: return
@@ -541,25 +518,21 @@ class ChatViewModel(
         id: String,
         error: Any?,
     ) {
-        val pending = pendingRequests.remove(id)
+        val method = idToMethod.remove(id) ?: return
-        val method = pending?.method
         val errorMsg =
             when (error) {
                 is Map<*, *> -> error["message"] as? String ?: error.toString()
                 else -> error.toString()
             }
 
-        // Fail the deferred so sendRpcAndAwait() callers see the error
+        // Surface error in UI (these are server-pushed RpcError for
-        pending?.deferred?.completeExceptionally(RpcCallException(errorMsg))
+        // fire-and-forget RPCs — awaited RPCs handle their own failure
-
+        // via the HermesWsClient.request() deferred).
-        // Only surface error in UI for non-awaited RPCs
+        _uiState.update {
-        if (pending?.deferred == null) {
+            it.copy(
-            _uiState.update {
+                isLoading = false,
-                it.copy(
+                errorMessage = "Error${if (method != null) " ($method)" else ""}: $errorMsg",
-                    isLoading = false,
+            )
-                    errorMessage = "Error${if (method != null) " ($method)" else ""}: $errorMsg",
-                )
-            }
         }
     }
 
@@ -700,22 +673,16 @@ class ChatViewModel(
         }
 
     /**
-     * Send a JSON-RPC call and suspend until the response arrives.
+     * Send a JSON-RPC call and suspend until the response arrives, delegating
-     * Throws [RpcCallException] on error, [kotlinx.coroutines.TimeoutCancellationException] on timeout.
+     * the deferred + 120s timeout to [HermesWsClient.request] (issue #526).
+     * Throws [HermesWsClient.HermesRpcException] on RPC error, or
+     * [kotlinx.coroutines.TimeoutCancellationException] if the server never
+     * answers within the timeout.
      */
     private suspend fun sendRpcAndAwait(
         method: String,
         params: Map<String, Any>,
-        timeoutMs: Long = 30_000,
+    ): Any? = HermesWsClient.request(method, params).await()
-    ): Any? {
-        val deferred = CompletableDeferred<Any?>()
-        wsClient.send(
-            method = method,
-            params = params,
-            onSent = { id -> trackRequest(id, method, deferred) },
-        )
-        return withTimeout(timeoutMs) { deferred.await() }
-    }
 
     // ── Attachment management ─────────────────────────────────────────────
 
@@ -1215,6 +1182,7 @@ class ChatViewModel(
             )
         }
         viewModelScope.launch(Dispatchers.IO) {
+            wsClient.rejectAllPending()
             wsClient.disconnect()
         }
         viewModelScope.launch {
@@ -1341,40 +1309,14 @@ class ChatViewModel(
 
     // ── Pending request tracking ─────────────────────────────────────────
 
-    private data class PendingRequest(
+    /** Maps an in-flight RPC id → its method, purely for UI error labeling. The deferred/timeout lives in [HermesWsClient.request] (issue #526). */
-        val method: String,
+    private val idToMethod = ConcurrentHashMap<String, String>()
-        val createdAt: Long = System.currentTimeMillis(),
-        val deferred: CompletableDeferred<Any?>? = null,
-    )
 
     private fun trackRequest(
         id: String,
         method: String,
-        deferred: CompletableDeferred<Any?>? = null,
     ) {
-        pendingRequests[id] = PendingRequest(method, deferred = deferred)
+        idToMethod[id] = method
-    }
-
-    /**
-     * Periodically prunes stale pending requests that the server never
-     * responded to. Surfaces a timeout error for the user.
-     */
-    private fun startPendingRequestCleanup() {
-        pendingCleanupJob =
-            viewModelScope.launch {
-                while (true) {
-                    delay(PENDING_REQUEST_TIMEOUT_MS)
-                    val now = System.currentTimeMillis()
-                    val stale =
-                        pendingRequests.entries.filter {
-                            now - it.value.createdAt > PENDING_REQUEST_TIMEOUT_MS
-                        }
-                    for (entry in stale) {
-                        pendingRequests.remove(entry.key)
-                        Log.w(TAG, "Request timed out: ${entry.value.method} (id=${entry.key})")
-                    }
-                }
-            }
     }
 
     // ── Search ────────────────────────────────────────────────────────────
@@ -1410,7 +1352,6 @@ class ChatViewModel(
 
     override fun onCleared() {
         super.onCleared()
-        pendingCleanupJob?.cancel()
         // PERF-16: Don't disconnect the global HermesWsClient singleton when
         // leaving the Chat screen — it's used by background notification reply.
     }

app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt

@@ -120,8 +120,10 @@ class ChatViewModelTest {
      * Request ID sequence: GatewayReady triggers loadSessions (req-id-1),
      * fetchCommandCatalog (req-id-2), then createNewSession (req-id-3).
      */
-    private suspend fun TestScope.createViewModelWithSession(): Pair<ChatViewModel, String> {
+    private suspend fun TestScope.createViewModelWithSession(
-        val viewModel = createViewModel()
+        startCleanup: Boolean = false,
+    ): Pair<ChatViewModel, String> {
+        val viewModel = createViewModel(startCleanup)
         advanceUntilIdle()
 
         mockConnectionStatus.value = ConnectionStatus.CONNECTED
@@ -1243,4 +1245,68 @@ class ChatViewModelTest {
                     viewModel.uiState.value.messages.any { it.content == "Here is an image" }
             assertTrue("Message should still be sent despite attachment read failure", sent)
         }
+
+    // ── Pending request timeout + rejectAllPending (issue #526) ───────────
+
+    /**
+     * On disconnect (RECONNECTING) the ViewModel must run rejectAllPending
+     * without throwing and stay usable — mirroring desktop
+     * JsonRpcGatewayClient.rejectAllPending invoked on socket close. This is
+     * what prevents callers awaiting a CompletableDeferred from hanging
+     * across a socket drop.
+     */
+    @Test
+    fun testDisconnect_rejectsPendingWithoutError() =
+        runTest {
+            val (viewModel, _) = createViewModelWithSession()
+
+            mockEventsFlow.emit(
+                WsEvent.ApprovalRequest(
+                    command = "rm",
+                    description = "Dangerous",
+                    patternKeys = null,
+                    sessionId = null,
+                ),
+            )
+            advanceUntilIdle()
+            viewModel.respondToApproval("approve")
+            advanceUntilIdle()
+
+            // Simulate socket drop → reconnecting (triggers rejectAllPending).
+            mockConnectionStatus.value = ConnectionStatus.RECONNECTING
+            advanceUntilIdle()
+
+            // No exception propagated; VM remains usable.
+            assertNull(viewModel.uiState.value.errorMessage)
+        }
+
+    /**
+     * viewModel.reconnect() calls rejectAllPending() before wsClient.disconnect(),
+     * so any in-flight awaited RPC is failed fast instead of hanging until its
+     * own timeout.
+     */
+    @Test
+    fun testReconnect_rejectsPendingWithoutError() =
+        runTest {
+            val (viewModel, _) = createViewModelWithSession()
+
+            mockEventsFlow.emit(
+                WsEvent.ApprovalRequest(
+                    command = "rm",
+                    description = "Dangerous",
+                    patternKeys = null,
+                    sessionId = null,
+                ),
+            )
+            advanceUntilIdle()
+            viewModel.respondToApproval("approve")
+            advanceUntilIdle()
+
+            // User-initiated reconnect must not throw / hang.
+            viewModel.reconnect()
+            advanceUntilIdle()
+
+            assertNull(viewModel.uiState.value.errorMessage)
+            verify { HermesWsClient.disconnect() }
+        }
 }