fix(notification): disconnect idle background socket (#858)

* fix(notification): disconnect idle background socket

* fix(notification): clear rejected prompt state

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

@@ -14,20 +14,15 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharedFlow
-import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
-import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.filterIsInstance
-import kotlinx.coroutines.flow.flowOn
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.launch
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
@@ -91,6 +86,7 @@ object HermesWsClient {
     private val connected = AtomicBoolean(false)
     private val intentionalClose = AtomicBoolean(false)
     private val acceptQueuedMessages = AtomicBoolean(true)
+    private val appInForeground = AtomicBoolean(true)
     private val messageQueue = ConcurrentLinkedQueue<String>()
     private val queuedMessagesById = ConcurrentHashMap<String, String>()
     private val outboundLock = Any()
@@ -153,26 +149,14 @@ object HermesWsClient {
 
     // ── Public observable stream ─────────────────────────────────────────
 
-    private val rawMessages =
-        MutableSharedFlow<String>(
+    private val parsedEvents =
+        MutableSharedFlow<WsEvent>(
             extraBufferCapacity = 512,
             onBufferOverflow = BufferOverflow.DROP_OLDEST,
         )
 
     /** Collect this from ViewModels to receive all parsed [WsEvent]s. */
-    val events: SharedFlow<WsEvent> =
-        rawMessages
-            .buffer(Channel.BUFFERED)
-            .map { text ->
-                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)
-                }
-            }.flowOn(Dispatchers.Default) // CPU-bound
-            .shareIn(wsScope, SharingStarted.Eagerly)
+    val events: SharedFlow<WsEvent> = parsedEvents.asSharedFlow()
 
     // ── Connection status flow ──────────────────────────────────────────
     private val _connectionStatus = MutableStateFlow(ConnectionStatus.DISCONNECTED)
@@ -189,6 +173,8 @@ object HermesWsClient {
     var pendingReply: Boolean = false
         private set
 
+    private val pendingPromptSubmits = ConcurrentHashMap.newKeySet<String>()
+
     // ── Credential warning (issue #534) ─────────────────────────────────
     // Backend surfaces `credential_warning` in `gateway.ready` / `session.info`
     // WS payloads (desktop `requestDesktopOnboarding`). Mobile has no equivalent
@@ -233,7 +219,10 @@ object HermesWsClient {
                     is WsEvent.ToolStart,
                     -> pendingReply = true
 
-                    is WsEvent.MessageComplete -> pendingReply = false
+                    is WsEvent.MessageComplete -> {
+                        pendingReply = false
+                        disconnectIfIdleInBackground()
+                    }
                     else -> {}
                 }
             }
@@ -253,6 +242,23 @@ object HermesWsClient {
     @VisibleForTesting
     val isConnected: Boolean get() = connected.get()
 
+    fun setAppForeground(foreground: Boolean) {
+        appInForeground.set(foreground)
+        if (!foreground) disconnectIfIdleInBackground()
+    }
+
+    private fun disconnectIfIdleInBackground() {
+        synchronized(outboundLock) {
+            if (!appInForeground.get() &&
+                !pendingReply &&
+                pendingCalls.isEmpty() &&
+                messageQueue.isEmpty()
+            ) {
+                disconnect()
+            }
+        }
+    }
+
     /** Open a WebSocket connection using settings from [AuthManager]. */
     fun connect() {
         if (networkCollectorStarted.compareAndSet(false, true)) {
@@ -501,6 +507,8 @@ object HermesWsClient {
             if (clearPendingMessages) {
                 messageQueue.clear()
                 queuedMessagesById.clear()
+                pendingPromptSubmits.clear()
+                pendingReply = false
             }
             connected.set(false)
             _connectionStatus.value = ConnectionStatus.DISCONNECTED
@@ -577,6 +585,7 @@ object HermesWsClient {
         } else {
             call.deferred.complete(result)
         }
+        disconnectIfIdleInBackground()
     }
 
     /**
@@ -621,7 +630,12 @@ object HermesWsClient {
             )
         val json = OkHttpProvider.json.encodeToString(request)
         if (BuildConfig.DEBUG) Log.d(TAG, "→ $json")
+        var reconnect = false
         synchronized(outboundLock) {
+            if (method == WsMethods.PROMPT_SUBMIT) {
+                pendingPromptSubmits.add(id)
+                pendingReply = true
+            }
             onSent?.invoke(id)
             val ws = webSocket
             if (ws != null && connected.get()) {
@@ -640,11 +654,13 @@ object HermesWsClient {
                 if (isRetryableMessage(json)) {
                     Log.d(TAG, "WS disconnected — queuing message")
                     queueMessage(id, json)
+                    reconnect = true
                 } else {
                     Log.w(TAG, "WS disconnected with oversized outgoing message — not queueing")
                 }
             }
         }
+        if (reconnect) connect()
         return id
     }
 
@@ -701,10 +717,6 @@ object HermesWsClient {
         text: String,
         onSent: ((String) -> Unit)? = null,
     ): String {
-        // A reply is now pending — even before the first MessageStart event
-        // (the agent may be in its tool phase), so backgrounding the app in
-        // that window still arms the notification service (issue #794).
-        pendingReply = true
         return send(
             method = WsMethods.PROMPT_SUBMIT,
             params = mapOf("session_id" to sessionId, "text" to text),
@@ -848,6 +860,7 @@ object HermesWsClient {
                     markQueuedMessageSent(msg)
                 }
             }
+            disconnectIfIdleInBackground()
         }
 
         override fun onMessage(
@@ -862,25 +875,41 @@ object HermesWsClient {
             val event =
                 try {
                     val rpc = OkHttpProvider.json.decodeFromString<JsonRpcResponse>(text)
-                    EventParser.parse(rpc, text)
+                    val parsed = EventParser.parse(rpc, text)
+                    if (parsed is WsEvent.MessageComplete) {
+                        parsed.copy(
+                            storedSessionId = ActiveSessionHolder.resolveStoredSessionId(parsed.sessionId),
+                        )
+                    } else {
+                        parsed
+                    }
                 } catch (e: Exception) {
                     Log.e(TAG, "Failed to parse message", e)
                     WsEvent.Unknown(text)
                 }
             when (event) {
                 is WsEvent.RpcResult -> {
+                    synchronized(outboundLock) { pendingPromptSubmits.remove(event.id) }
                     removeQueuedMessage(event.id)
                     resolvePending(event.id, event.result, null)
                 }
 
                 is WsEvent.RpcError -> {
+                    synchronized(outboundLock) {
+                        if (pendingPromptSubmits.remove(event.id) &&
+                            pendingPromptSubmits.isEmpty()
+                        ) {
+                            pendingReply = false
+                        }
+                    }
                     removeQueuedMessage(event.id)
                     resolvePending(event.id, null, event.error)
+                    disconnectIfIdleInBackground()
                 }
 
                 else -> Unit
             }
-            val emitted = rawMessages.tryEmit(text)
+            val emitted = parsedEvents.tryEmit(event)
             if (!emitted && BuildConfig.DEBUG) {
                 Log.w(TAG, "WebSocket message dropped due to buffer overflow")
             }

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

@@ -57,6 +57,8 @@ sealed class WsEvent {
          * (e.g. a tool call wiped the streaming buffer mid-turn).
          */
         val reasoning: String? = null,
+        /** Stored session id captured before background disconnect clears the active mapping. */
+        val storedSessionId: String? = null,
     ) : WsEvent()
 
     data class MessageDone(