diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt index dada7897..46bfa01a 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt +++ b/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, + var timeoutJob: Job? = null, + ) + + /** Tracks in-flight [request] calls by their JSON-RPC id. */ + private val pendingCalls = ConcurrentHashMap() + + /** + * 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 = emptyMap(), + timeoutMs: Long = REQUEST_TIMEOUT_MS, + ): CompletableDeferred { + val deferred = CompletableDeferred() + 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(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") diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt index 8c9327fa..90c93160 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt +++ b/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 = emptyList(), val currentSessionId: String? = null, @@ -132,7 +119,6 @@ class ChatViewModel( val streamingState: StateFlow = _streamingState.asStateFlow() private val wsClient = HermesWsClient - private val pendingRequests = ConcurrentHashMap() 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 = pending.method - - // Complete the deferred so sendRpcAndAwait() callers can resume - pending.deferred?.complete(result) - + val method = idToMethod.remove(id) ?: return when (method) { WsMethods.SESSION_CREATE -> { val resultMap = result as? Map ?: return @@ -541,25 +518,21 @@ class ChatViewModel( id: String, error: Any?, ) { - val pending = pendingRequests.remove(id) - val method = pending?.method + val method = idToMethod.remove(id) ?: return val errorMsg = when (error) { is Map<*, *> -> error["message"] as? String ?: error.toString() else -> error.toString() } - // Fail the deferred so sendRpcAndAwait() callers see the error - pending?.deferred?.completeExceptionally(RpcCallException(errorMsg)) - - // Only surface error in UI for non-awaited RPCs - if (pending?.deferred == null) { - _uiState.update { - it.copy( - isLoading = false, - errorMessage = "Error${if (method != null) " ($method)" else ""}: $errorMsg", - ) - } + // Surface error in UI (these are server-pushed RpcError for + // fire-and-forget RPCs — awaited RPCs handle their own failure + // via the HermesWsClient.request() deferred). + _uiState.update { + it.copy( + 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. - * Throws [RpcCallException] on error, [kotlinx.coroutines.TimeoutCancellationException] on timeout. + * Send a JSON-RPC call and suspend until the response arrives, delegating + * 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, - timeoutMs: Long = 30_000, - ): Any? { - val deferred = CompletableDeferred() - wsClient.send( - method = method, - params = params, - onSent = { id -> trackRequest(id, method, deferred) }, - ) - return withTimeout(timeoutMs) { deferred.await() } - } + ): Any? = HermesWsClient.request(method, params).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( - val method: String, - val createdAt: Long = System.currentTimeMillis(), - val deferred: CompletableDeferred? = null, - ) + /** Maps an in-flight RPC id → its method, purely for UI error labeling. The deferred/timeout lives in [HermesWsClient.request] (issue #526). */ + private val idToMethod = ConcurrentHashMap() private fun trackRequest( id: String, method: String, - deferred: CompletableDeferred? = null, ) { - pendingRequests[id] = PendingRequest(method, deferred = deferred) - } - - /** - * 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})") - } - } - } + idToMethod[id] = method } // ── 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. } diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt index 83b61b70..b0a20540 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt +++ b/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 { - val viewModel = createViewModel() + private suspend fun TestScope.createViewModelWithSession( + startCleanup: Boolean = false, + ): Pair { + 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() } + } }