fix: keep transient ticket failures reconnecting (#740)

* fix: keep transient ticket failures reconnecting

* fix: recover rejected websocket sends

* fix: serialize websocket reconnect scheduling

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

@@ -14,6 +14,7 @@ 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
@@ -33,6 +34,7 @@ import okhttp3.RequestBody.Companion.toRequestBody
 import okhttp3.Response
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
+import okio.utf8Size
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicBoolean
@@ -65,17 +67,26 @@ object HermesWsClient {
     private const val INITIAL_BACKOFF_MS = 1_000L
     private const val MAX_BACKOFF_MS = 30_000L
     private const val BACKOFF_MULTIPLIER = 2.0
+    private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024
+    private const val OUTBOUND_DRAIN_TIMEOUT_MS = 60_000L
 
     // ── Internal state (all access through synchronized / atomic) ────────
 
     private val requestId = AtomicInteger(0)
+    private val connectionGeneration = AtomicInteger(0)
     private val connected = AtomicBoolean(false)
     private val intentionalClose = AtomicBoolean(false)
+    private val acceptQueuedMessages = AtomicBoolean(true)
     private val messageQueue = ConcurrentLinkedQueue<String>()
+    private val queuedMessagesById = ConcurrentHashMap<String, String>()
+    private val outboundLock = Any()
 
     @Volatile
     private var webSocket: WebSocket? = null
 
+    @Volatile
+    private var closingSocket: WebSocket? = null
+
     @Volatile
     private var currentBackoff = INITIAL_BACKOFF_MS
 
@@ -84,6 +95,9 @@ object HermesWsClient {
     @Volatile
     private var reconnectJob: Job? = null
 
+    @Volatile
+    private var outboundDrainJob: Job? = null
+
     // ── Health and Ping/Pong tracking ────────────────────────────────────
 
     @Volatile
@@ -161,10 +175,19 @@ object HermesWsClient {
         // Monitor network state to trigger immediate reconnect when network is restored
         wsScope.launch {
             NetworkMonitor.isConnected.collect { connected ->
+                val shouldOpen =
+                    synchronized(outboundLock) {
                         if (connected && !isConnected && !intentionalClose.get() && AuthManager.isAutoReconnect()) {
                             Log.d(TAG, "Network restored — triggering immediate reconnect")
                             currentBackoff = INITIAL_BACKOFF_MS
                             reconnectJob?.cancel()
+                            reconnectJob = null
+                            true
+                        } else {
+                            false
+                        }
+                    }
+                if (shouldOpen) {
                     openSocket()
                 }
             }
@@ -193,6 +216,7 @@ object HermesWsClient {
 
     /** Open a WebSocket connection using settings from [AuthManager]. */
     fun connect() {
+        acceptQueuedMessages.set(true)
         if (connected.get()) {
             Log.d(TAG, "Already connected — skipping")
             return
@@ -217,7 +241,9 @@ object HermesWsClient {
             return
         }
         intentionalClose.set(false)
+        synchronized(outboundLock) {
             currentBackoff = INITIAL_BACKOFF_MS
+        }
         _connectionStatus.value = ConnectionStatus.CONNECTING
         openSocket()
     }
@@ -268,6 +294,7 @@ object HermesWsClient {
                     client.newCall(request).execute()
                 }
 
+            response.use {
                 if (response.isSuccessful) {
                     val body = response.body.string()
                     val ticketMatch = Regex("""\"ticket\":\"([^\"]+)\"""").find(body)
@@ -278,32 +305,54 @@ object HermesWsClient {
                         return true
                     } else {
                         Log.w(TAG, "WS ticket refresh failed: response body did not contain ticket")
-                    _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED
+                        return handleWsTicketRefreshFailure(ConnectionStatus.DISCONNECTED)
-                    return false
                     }
                 } else {
                     Log.w(TAG, "WS ticket refresh failed: HTTP ${response.code}")
-                _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED
+                    val status =
-                return false
+                        when {
+                            response.code == 401 || response.code == 403 -> ConnectionStatus.AUTH_EXPIRED
+                            response.code == 408 || response.code == 429 || response.code >= 500 ->
+                                ConnectionStatus.RECONNECTING
+                            else -> ConnectionStatus.DISCONNECTED
+                        }
+                    return handleWsTicketRefreshFailure(status)
+                }
             }
         } catch (e: Exception) {
             Log.w(TAG, "WS ticket refresh failed: ${e.javaClass.simpleName}")
-            _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED
+            return handleWsTicketRefreshFailure(ConnectionStatus.RECONNECTING)
-            return false
         }
     }
 
+    private fun handleWsTicketRefreshFailure(status: ConnectionStatus): Boolean {
+        _connectionStatus.value = status
+        if (status == ConnectionStatus.RECONNECTING) scheduleReconnect()
+        return false
+    }
+
     /** Cleanly close the WebSocket and stop auto-reconnect. */
-    fun disconnect() {
+    fun disconnect(clearPendingMessages: Boolean = false) {
+        synchronized(outboundLock) {
             intentionalClose.set(true)
+            acceptQueuedMessages.set(!clearPendingMessages)
+            connectionGeneration.incrementAndGet()
             reconnectJob?.cancel()
             reconnectJob = null
+            outboundDrainJob?.cancel()
+            outboundDrainJob = null
             stopHealthTracking()
             webSocket?.close(1000, "Client closed")
             webSocket = null
+            closingSocket = null
+            if (clearPendingMessages) {
+                messageQueue.clear()
+                queuedMessagesById.clear()
+            }
             connected.set(false)
             _connectionStatus.value = ConnectionStatus.DISCONNECTED
         }
+    }
 
     // ── Awaited RPC request layer (issue #526) ─────────────────────────
     // Mirrors desktop apps/shared JsonRpcGatewayClient.request(): an in-flight
@@ -368,6 +417,7 @@ object HermesWsClient {
         error: JsonRpcError?,
     ) {
         val call = pendingCalls.remove(id) ?: return
+        removeQueuedMessage(id)
         call.timeoutJob?.cancel()
         if (error != null) {
             call.deferred.completeExceptionally(HermesRpcException(error.message))
@@ -390,6 +440,7 @@ object HermesWsClient {
         val snapshot = pendingCalls.toList()
         pendingCalls.clear()
         for ((id, call) in snapshot) {
+            removeQueuedMessage(id)
             Log.w(TAG, "Rejecting pending request on disconnect: ${call.method} (id=$id)")
             call.timeoutJob?.cancel()
             call.deferred.completeExceptionally(error)
@@ -412,16 +463,79 @@ object HermesWsClient {
         val request = JsonRpcRequest(id = id, method = method, params = params.mapValues { it.value.toJsonElement() })
         val json = OkHttpProvider.json.encodeToString(request)
         if (BuildConfig.DEBUG) Log.d(TAG, "→ $json")
+        synchronized(outboundLock) {
             val ws = webSocket
             if (ws != null && connected.get()) {
-            ws.send(json)
+                if (!ws.send(json)) {
+                    if (webSocket !== ws || !acceptQueuedMessages.get()) return@synchronized
+                    if (isRetryableMessage(json)) {
+                        Log.w(TAG, "WS rejected outgoing message — queuing for reconnect")
+                        queueMessage(id, json)
+                        recoverRejectedSocket(ws)
                     } else {
+                        Log.w(TAG, "WS rejected oversized outgoing message — not retrying")
+                    }
+                }
+            } else if (acceptQueuedMessages.get()) {
+                if (isRetryableMessage(json)) {
                     Log.d(TAG, "WS disconnected — queuing message")
-            messageQueue.add(json)
+                    queueMessage(id, json)
+                } else {
+                    Log.w(TAG, "WS disconnected with oversized outgoing message — not queueing")
+                }
+            }
         }
         return id
     }
 
+    private fun isRetryableMessage(json: String): Boolean {
+        if (json.length > MAX_OUTBOUND_MESSAGE_BYTES) return false
+        if (json.length <= MAX_OUTBOUND_MESSAGE_BYTES / 4) return true
+        return json.utf8Size() <= MAX_OUTBOUND_MESSAGE_BYTES.toLong()
+    }
+
+    private fun queueMessage(
+        id: String,
+        json: String,
+    ) {
+        messageQueue.add(json)
+        queuedMessagesById[id] = json
+    }
+
+    private fun removeQueuedMessage(id: String) {
+        synchronized(outboundLock) {
+            queuedMessagesById.remove(id)?.let(messageQueue::remove)
+        }
+    }
+
+    private fun markQueuedMessageSent(json: String) {
+        queuedMessagesById.entries.removeIf { it.value == json }
+    }
+
+    private fun recoverRejectedSocket(ws: WebSocket) {
+        connected.set(false)
+        if (closingSocket === ws) return
+        _connectionStatus.value = ConnectionStatus.RECONNECTING
+        if (ws.queueSize() == 0L) {
+            ws.cancel()
+            scheduleReconnect()
+            return
+        }
+        if (intentionalClose.get()) return
+        outboundDrainJob?.cancel()
+        outboundDrainJob =
+            wsScope.launch {
+                delay(OUTBOUND_DRAIN_TIMEOUT_MS)
+                synchronized(outboundLock) {
+                    if (!connected.get() && !intentionalClose.get() && webSocket === ws) {
+                        ws.cancel()
+                        outboundDrainJob = null
+                        scheduleReconnect()
+                    }
+                }
+            }
+    }
+
     /** Convenience: submit a user prompt to an existing session. */
     fun sendMessage(
         sessionId: String,
@@ -463,11 +577,20 @@ object HermesWsClient {
         if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl")
 
         val request = Request.Builder().url(url).build()
-        webSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl())
+        val generation = connectionGeneration.incrementAndGet()
+        val newSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl(generation))
+        synchronized(outboundLock) {
+            if (connectionGeneration.get() == generation && !intentionalClose.get()) {
+                webSocket = newSocket
+            } else {
+                newSocket.cancel()
+            }
+        }
     }
 
     private fun scheduleReconnect() {
-        if (intentionalClose.get()) return
+        synchronized(outboundLock) {
+            if (intentionalClose.get() || reconnectJob?.isActive == true) return
             if (!AuthManager.isAutoReconnect()) {
                 if (BuildConfig.DEBUG) Log.d(TAG, "Auto-reconnect disabled")
                 _connectionStatus.value = ConnectionStatus.DISCONNECTED
@@ -478,41 +601,72 @@ object HermesWsClient {
                 _connectionStatus.value = ConnectionStatus.NO_NETWORK
                 return
             }
-        val delay = currentBackoff
+            val reconnectDelay = currentBackoff
             currentBackoff =
                 (currentBackoff * BACKOFF_MULTIPLIER)
                     .toLong()
                     .coerceAtMost(MAX_BACKOFF_MS)
-        if (BuildConfig.DEBUG) Log.d(TAG, "Reconnecting in ${delay}ms …")
+            if (BuildConfig.DEBUG) Log.d(TAG, "Reconnecting in ${reconnectDelay}ms …")
 
-        reconnectJob?.cancel()
             reconnectJob =
                 wsScope.launch {
-                delay(delay)
+                    delay(reconnectDelay)
-                if (!intentionalClose.get() && !connected.get()) {
+                    val owner = currentCoroutineContext()[Job]
-                    openSocket()
+                    val shouldOpen =
+                        synchronized(outboundLock) {
+                            if (reconnectJob !== owner) {
+                                false
+                            } else {
+                                reconnectJob = null
+                                !intentionalClose.get() && !connected.get()
+                            }
+                        }
+                    if (shouldOpen) openSocket()
                 }
         }
     }
 
     // ── Listener ─────────────────────────────────────────────────────────
 
-    private class WsListenerImpl : WebSocketListener() {
+    private class WsListenerImpl(
+        private val generation: Int,
+    ) : WebSocketListener() {
+        private fun isCurrent(): Boolean = connectionGeneration.get() == generation && !intentionalClose.get()
+
         override fun onOpen(
             webSocket: WebSocket,
             response: Response,
         ) {
+            synchronized(outboundLock) {
+                if (!isCurrent()) {
+                    webSocket.close(1000, "Superseded")
+                    return
+                }
+                HermesWsClient.webSocket = webSocket
+                closingSocket = null
+                outboundDrainJob?.cancel()
+                outboundDrainJob = null
                 Log.i(TAG, "WebSocket opened")
                 connected.set(true)
                 _connectionStatus.value = ConnectionStatus.CONNECTED
                 currentBackoff = INITIAL_BACKOFF_MS
                 startHealthTracking()
 
-            while (messageQueue.isNotEmpty()) {
+                while (true) {
-                val msg = messageQueue.poll()
+                    val msg = messageQueue.peek() ?: break
-                if (msg != null) {
+                    if (!isRetryableMessage(msg)) {
+                        Log.w(TAG, "Dropping oversized queued message")
+                        messageQueue.poll()
+                        markQueuedMessageSent(msg)
+                        continue
+                    }
                     if (BuildConfig.DEBUG) Log.d(TAG, "→ (queued) $msg")
-                    webSocket.send(msg)
+                    if (!webSocket.send(msg)) {
+                        recoverRejectedSocket(webSocket)
+                        break
+                    }
+                    messageQueue.poll()
+                    markQueuedMessageSent(msg)
                 }
             }
         }
@@ -521,6 +675,7 @@ object HermesWsClient {
             webSocket: WebSocket,
             text: String,
         ) {
+            if (!isCurrent() || HermesWsClient.webSocket !== webSocket) return
             if (BuildConfig.DEBUG) Log.d(TAG, "← $text")
             lastPongTimestamp = System.currentTimeMillis()
             // Resolve any in-flight `request()` awaiting this RPC result/error
@@ -549,16 +704,36 @@ object HermesWsClient {
             code: Int,
             reason: String,
         ) {
+            synchronized(outboundLock) {
+                if (!isCurrent()) {
+                    webSocket.close(code, reason)
+                    return
+                }
+                closingSocket = webSocket
                 // Do NOT log [reason] — it may carry server-side context.
                 Log.d(TAG, "WebSocket closing: $code")
+                if (code == 4001 || code == 4401 ||
+                    reason.contains("unauthorized", ignoreCase = true) ||
+                    reason.startsWith("auth:", ignoreCase = true)
+                ) {
+                    _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED
+                }
                 webSocket.close(code, reason)
             }
+        }
 
         override fun onClosed(
             webSocket: WebSocket,
             code: Int,
             reason: String,
         ) {
+            synchronized(outboundLock) {
+                if (!isCurrent()) return
+                connectionGeneration.incrementAndGet()
+                closingSocket = null
+                outboundDrainJob?.cancel()
+                outboundDrainJob = null
+                if (HermesWsClient.webSocket === webSocket) HermesWsClient.webSocket = null
                 // Do NOT log [reason] — it may carry server-side context. The
                 // reason is still inspected internally to detect auth failures.
                 Log.i(TAG, "WebSocket closed: $code")
@@ -574,12 +749,20 @@ object HermesWsClient {
                     scheduleReconnect()
                 }
             }
+        }
 
         override fun onFailure(
             webSocket: WebSocket,
             t: Throwable,
             response: Response?,
         ) {
+            synchronized(outboundLock) {
+                if (!isCurrent()) return
+                connectionGeneration.incrementAndGet()
+                closingSocket = null
+                outboundDrainJob?.cancel()
+                outboundDrainJob = null
+                if (HermesWsClient.webSocket === webSocket) HermesWsClient.webSocket = null
                 // Log the exception class only — [Throwable.message] can leak URLs
                 // or headers. The message is still inspected internally for auth
                 // detection.
@@ -598,4 +781,5 @@ object HermesWsClient {
                 }
             }
         }
+    }
 }

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

@@ -121,9 +121,7 @@ fun ChatConnectionBanner(
                         }
                     }
 
-                    ConnectionStatus.RECONNECTING,
+                    ConnectionStatus.AUTH_EXPIRED -> {
-                    ConnectionStatus.AUTH_EXPIRED,
-                    -> {
                         TextButton(
                             onClick = onReloginClick,
                             colors =
@@ -135,6 +133,8 @@ fun ChatConnectionBanner(
                         }
                     }
 
+                    ConnectionStatus.RECONNECTING -> {}
+
                     else -> {}
                 }
             }

app/src/main/java/com/m57/hermescontrol/ui/settings/SettingsViewModel.kt

@@ -299,7 +299,7 @@ class SettingsViewModel(
         AuthManager.setToken(null)
         AuthManager.setSessionCookie(null)
         AuthManager.setWsAuthParam("token")
-        HermesWsClient.disconnect()
+        HermesWsClient.disconnect(clearPendingMessages = true)
         // Don't rebuild ApiClient here — let the navigation complete first
     }
 

app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt

@@ -4,6 +4,7 @@ import android.util.Log
 import com.m57.hermescontrol.data.local.AuthManager
 import com.m57.hermescontrol.data.remote.CleartextPolicy
 import com.m57.hermescontrol.data.remote.CookieManager
+import com.m57.hermescontrol.data.remote.NetworkMonitor
 import com.m57.hermescontrol.data.remote.ServerEndpoint
 import com.m57.hermescontrol.data.remote.buildFakePersistentCookieJar
 import io.mockk.every
@@ -11,6 +12,7 @@ import io.mockk.mockk
 import io.mockk.mockkObject
 import io.mockk.mockkStatic
 import io.mockk.unmockkAll
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
@@ -24,9 +26,11 @@ import org.junit.After
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertSame
 import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Test
+import java.io.IOException
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 
@@ -72,12 +76,20 @@ class HermesWsClientTest {
         @Suppress("UNCHECKED_CAST")
         (statusField.get(HermesWsClient) as MutableStateFlow<ConnectionStatus>).value = ConnectionStatus.DISCONNECTED
 
-        HermesWsClient.disconnect() // Ensure it starts clean
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        (queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>).clear()
+
+        HermesWsClient.disconnect(clearPendingMessages = true) // Ensure it starts clean
+        val acceptQueuedMessagesField = HermesWsClient::class.java.getDeclaredField("acceptQueuedMessages")
+        acceptQueuedMessagesField.isAccessible = true
+        (acceptQueuedMessagesField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true)
     }
 
     @After
     fun tearDown() {
-        HermesWsClient.disconnect()
+        HermesWsClient.disconnect(clearPendingMessages = true)
         // Wait a bit to allow internal OkHttp coroutines to clean up before shutting down MockWebServer
         // Increased from 100ms for OkHttp 5.x — needs more time for the WS close handshake
         Thread.sleep(500)
@@ -136,6 +148,395 @@ class HermesWsClientTest {
         assertTrue(msg.contains(id))
     }
 
+    @Test
+    fun testFailedSocketSendQueuesMessageForReconnect() {
+        val staleSocket = mockk<WebSocket>(relaxed = true)
+        every { staleSocket.send(any<String>()) } returns false
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, staleSocket)
+
+        val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
+        connectedField.isAccessible = true
+        (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val statusField = HermesWsClient::class.java.getDeclaredField("_connectionStatus")
+        statusField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        (statusField.get(HermesWsClient) as MutableStateFlow<ConnectionStatus>).value = ConnectionStatus.CONNECTED
+
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello"))
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        assertFalse(HermesWsClient.isConnected)
+        assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value)
+        assertEquals(1, queue.size)
+    }
+
+    @Test
+    fun testReconnectSchedulingIsIdempotentWhilePending() {
+        mockkObject(NetworkMonitor)
+        every { AuthManager.isAutoReconnect() } returns true
+        every { NetworkMonitor.isConnected } returns MutableStateFlow(true)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff")
+        backoffField.isAccessible = true
+        backoffField.setLong(HermesWsClient, 1_000L)
+
+        val reconnectMethod = HermesWsClient::class.java.getDeclaredMethod("scheduleReconnect")
+        reconnectMethod.isAccessible = true
+        reconnectMethod.invoke(HermesWsClient)
+        reconnectMethod.invoke(HermesWsClient)
+
+        assertEquals(2_000L, backoffField.getLong(HermesWsClient))
+    }
+
+    @Test
+    fun testCompletedReconnectDoesNotClearReplacementJob() {
+        mockkObject(NetworkMonitor)
+        every { AuthManager.isAutoReconnect() } returns true
+        every { NetworkMonitor.isConnected } returns MutableStateFlow(true)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
+        connectedField.isAccessible = true
+        (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff")
+        backoffField.isAccessible = true
+        backoffField.setLong(HermesWsClient, 0L)
+
+        val reconnectJobField = HermesWsClient::class.java.getDeclaredField("reconnectJob")
+        reconnectJobField.isAccessible = true
+        val replacementJob = mockk<Job>(relaxed = true)
+
+        val lockField = HermesWsClient::class.java.getDeclaredField("outboundLock")
+        lockField.isAccessible = true
+        val lock = lockField.get(HermesWsClient)
+        val reconnectMethod = HermesWsClient::class.java.getDeclaredMethod("scheduleReconnect")
+        reconnectMethod.isAccessible = true
+
+        synchronized(lock) {
+            reconnectMethod.invoke(HermesWsClient)
+            reconnectJobField.set(HermesWsClient, replacementJob)
+        }
+
+        Thread.sleep(100)
+        assertSame(replacementJob, reconnectJobField.get(HermesWsClient))
+    }
+
+    @Test
+    fun testOversizedRejectedMessageIsNotQueued() {
+        val staleSocket = mockk<WebSocket>(relaxed = true)
+        every { staleSocket.send(any<String>()) } returns false
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, staleSocket)
+
+        val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
+        connectedField.isAccessible = true
+        (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true)
+
+        HermesWsClient.send(
+            WsMethods.PROMPT_SUBMIT,
+            mapOf("session_id" to "s1", "text" to "x".repeat(16 * 1024 * 1024 + 1)),
+        )
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        assertTrue(queue.isEmpty())
+        io.mockk.verify(exactly = 0) { staleSocket.cancel() }
+    }
+
+    @Test
+    fun testDisconnectedOversizedMessageIsNotQueued() {
+        HermesWsClient.send(
+            WsMethods.PROMPT_SUBMIT,
+            mapOf("session_id" to "s1", "text" to "x".repeat(16 * 1024 * 1024 + 1)),
+        )
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        assertTrue(queue.isEmpty())
+    }
+
+    @Test
+    fun testQueuePressureDoesNotCancelAcceptedFrames() {
+        val pressuredSocket = mockk<WebSocket>(relaxed = true)
+        every { pressuredSocket.send(any<String>()) } returns false
+        every { pressuredSocket.queueSize() } returns 1024L
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, pressuredSocket)
+
+        val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
+        connectedField.isAccessible = true
+        (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello"))
+
+        io.mockk.verify(exactly = 0) { pressuredSocket.cancel() }
+        val drainJobField = HermesWsClient::class.java.getDeclaredField("outboundDrainJob")
+        drainJobField.isAccessible = true
+        assertNotNull(drainJobField.get(HermesWsClient))
+    }
+
+    @Test
+    fun testFailureDuringOutboundDrainSchedulesReconnect() {
+        mockkObject(NetworkMonitor)
+        every { AuthManager.isAutoReconnect() } returns true
+        every { NetworkMonitor.isConnected } returns MutableStateFlow(true)
+
+        val pressuredSocket = mockk<WebSocket>(relaxed = true)
+        every { pressuredSocket.send(any<String>()) } returns false
+        every { pressuredSocket.queueSize() } returns 1024L
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, pressuredSocket)
+
+        val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
+        connectedField.isAccessible = true
+        (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration")
+        generationField.isAccessible = true
+        (generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger).set(1)
+
+        val backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff")
+        backoffField.isAccessible = true
+        backoffField.setLong(HermesWsClient, 1_000L)
+
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello"))
+
+        val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl")
+        val constructor = listenerClass.declaredConstructors.single()
+        constructor.isAccessible = true
+        val listener = constructor.newInstance(1) as WebSocketListener
+        listener.onFailure(pressuredSocket, IOException("connection lost"), null)
+
+        assertEquals(2_000L, backoffField.getLong(HermesWsClient))
+    }
+
+    @Test
+    fun testReplayPressureSchedulesRecoveryWithoutDroppingHead() {
+        val pressuredSocket = mockk<WebSocket>(relaxed = true)
+        every { pressuredSocket.send(any<String>()) } returnsMany listOf(true, false)
+        every { pressuredSocket.queueSize() } returns 1024L
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        queue.add("{\"jsonrpc\":\"2.0\",\"id\":\"1\"}")
+        queue.add("{\"jsonrpc\":\"2.0\",\"id\":\"2\"}")
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration")
+        generationField.isAccessible = true
+        (generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger).set(1)
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, pressuredSocket)
+
+        val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl")
+        val constructor = listenerClass.declaredConstructors.single()
+        constructor.isAccessible = true
+        val listener = constructor.newInstance(1) as WebSocketListener
+        listener.onOpen(pressuredSocket, mockk(relaxed = true))
+
+        assertFalse(HermesWsClient.isConnected)
+        assertEquals(1, queue.size)
+        assertTrue(queue.peek()?.contains("\"2\"") == true)
+        io.mockk.verify(exactly = 0) { pressuredSocket.cancel() }
+        val drainJobField = HermesWsClient::class.java.getDeclaredField("outboundDrainJob")
+        drainJobField.isAccessible = true
+        assertNotNull(drainJobField.get(HermesWsClient))
+    }
+
+    @Test
+    fun testStaleOnOpenDoesNotReplayQueue() {
+        val staleSocket = mockk<WebSocket>(relaxed = true)
+        val currentSocket = mockk<WebSocket>(relaxed = true)
+        every { currentSocket.send(any<String>()) } returns true
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, currentSocket)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration")
+        generationField.isAccessible = true
+        (generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger).set(2)
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        queue.add("{\"jsonrpc\":\"2.0\",\"id\":\"1\"}")
+
+        val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl")
+        val constructor = listenerClass.declaredConstructors.single()
+        constructor.isAccessible = true
+        val staleListener = constructor.newInstance(1) as WebSocketListener
+        val currentListener = constructor.newInstance(2) as WebSocketListener
+        val response = mockk<okhttp3.Response>(relaxed = true)
+
+        staleListener.onOpen(staleSocket, response)
+        currentListener.onOpen(currentSocket, response)
+
+        io.mockk.verify(exactly = 0) { staleSocket.send(any<String>()) }
+        io.mockk.verify(exactly = 1) { currentSocket.send(any<String>()) }
+        assertTrue(queue.isEmpty())
+    }
+
+    @Test
+    fun testRejectedSendAfterDisconnectIsNotQueued() {
+        val staleSocket = mockk<WebSocket>(relaxed = true)
+        every { staleSocket.send(any<String>()) } answers {
+            HermesWsClient.disconnect(clearPendingMessages = true)
+            false
+        }
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, staleSocket)
+
+        val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
+        connectedField.isAccessible = true
+        (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello"))
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        assertTrue(queue.isEmpty())
+        assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value)
+    }
+
+    @Test
+    fun testDisconnectPreservesQueuedMessagesUnlessExplicitlyCleared() {
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello"))
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        assertEquals(1, queue.size)
+
+        HermesWsClient.disconnect()
+        assertEquals(1, queue.size)
+
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "during reconnect"))
+        assertEquals(2, queue.size)
+
+        HermesWsClient.disconnect(clearPendingMessages = true)
+        assertTrue(queue.isEmpty())
+
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "after logout"))
+        assertTrue(queue.isEmpty())
+    }
+
+    @Test
+    fun testRejectAllPendingRemovesQueuedAwaitedRpc() {
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val deferred = HermesWsClient.request(WsMethods.PROCESS_LIST, mapOf("session_id" to "s1"))
+
+        val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue")
+        queueField.isAccessible = true
+        @Suppress("UNCHECKED_CAST")
+        val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue<String>
+        assertEquals(1, queue.size)
+
+        HermesWsClient.rejectAllPending()
+
+        assertTrue(deferred.isCompleted)
+        assertTrue(queue.isEmpty())
+    }
+
+    @Test
+    fun testAuthClosingCodeSurvivesRejectedSendRecovery() {
+        val socket = mockk<WebSocket>(relaxed = true)
+        every { socket.send(any<String>()) } returns false
+        every { socket.queueSize() } returns 0L
+
+        val socketField = HermesWsClient::class.java.getDeclaredField("webSocket")
+        socketField.isAccessible = true
+        socketField.set(HermesWsClient, socket)
+
+        val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
+        connectedField.isAccessible = true
+        (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true)
+
+        val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose")
+        intentionalCloseField.isAccessible = true
+        (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false)
+
+        val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration")
+        generationField.isAccessible = true
+        (generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger).set(1)
+
+        val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl")
+        val constructor = listenerClass.declaredConstructors.single()
+        constructor.isAccessible = true
+        val listener = constructor.newInstance(1) as WebSocketListener
+
+        listener.onClosing(socket, 4401, "expired")
+        HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello"))
+
+        assertEquals(ConnectionStatus.AUTH_EXPIRED, HermesWsClient.connectionStatus.value)
+        io.mockk.verify(exactly = 0) { socket.cancel() }
+    }
+
     @Test
     fun testReceiveMessage() {
         var serverWebSocket: WebSocket? = null
@@ -211,7 +612,7 @@ class HermesWsClientTest {
         runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
         assertTrue(serverLatch.await(5, TimeUnit.SECONDS))
 
-        HermesWsClient.disconnect()
+        HermesWsClient.disconnect(clearPendingMessages = true)
         assertFalse(HermesWsClient.isConnected)
         assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value)
 
@@ -378,7 +779,7 @@ class HermesWsClientTest {
         runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
 
         // Disconnect — this sets intentionalClose = true and cancels reconnect
-        HermesWsClient.disconnect()
+        HermesWsClient.disconnect(clearPendingMessages = true)
 
         assertFalse(HermesWsClient.isConnected)
         assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value)
@@ -482,7 +883,7 @@ class HermesWsClientTest {
         runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
 
         // Disconnect (sets intentionalClose) — after this, reconnect should be prevented
-        HermesWsClient.disconnect()
+        HermesWsClient.disconnect(clearPendingMessages = true)
         assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value)
         assertFalse(HermesWsClient.isConnected)
     }
@@ -557,4 +958,54 @@ class HermesWsClientTest {
 
         ticketServer.shutdown()
     }
+
+    @Test
+    fun testGatedMode_permanentTicketFailureStopsRetrying() {
+        every { AuthManager.isAutoReconnect() } returns true
+        every { AuthManager.serverStore } returns
+            mockk<com.m57.hermescontrol.data.config.ServerStore>().also {
+                every { it.getLatestState() } returns
+                    com.m57.hermescontrol.data.config
+                        .ServerStoreState(wsAuthParam = "ticket")
+            }
+
+        val ticketServer = MockWebServer()
+        ticketServer.start()
+        ticketServer.enqueue(MockResponse().setResponseCode(400))
+        every { AuthManager.endpointForBuild() } returns
+            ServerEndpoint.parse(
+                ticketServer.url("/").toString(),
+                CleartextPolicy.ALLOW_WITH_WARNING,
+            )
+
+        HermesWsClient.connect()
+
+        assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value)
+        ticketServer.shutdown()
+    }
+
+    @Test
+    fun testGatedMode_transientTicketFailureKeepsReconnectFlow() {
+        every { AuthManager.isAutoReconnect() } returns true
+        every { AuthManager.serverStore } returns
+            mockk<com.m57.hermescontrol.data.config.ServerStore>().also {
+                every { it.getLatestState() } returns
+                    com.m57.hermescontrol.data.config
+                        .ServerStoreState(wsAuthParam = "ticket")
+            }
+
+        val unavailableTicketServer = MockWebServer()
+        unavailableTicketServer.start()
+        val unavailableEndpoint = unavailableTicketServer.url("/").toString()
+        unavailableTicketServer.shutdown()
+        every { AuthManager.endpointForBuild() } returns
+            ServerEndpoint.parse(
+                unavailableEndpoint,
+                CleartextPolicy.ALLOW_WITH_WARNING,
+            )
+
+        HermesWsClient.connect()
+
+        assertEquals(ConnectionStatus.RECONNECTING, HermesWsClient.connectionStatus.value)
+    }
 }