refactor(#1376): extract WebSocket client components (#1424)

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

@@ -0,0 +1,140 @@
+package com.m57.hermescontrol.data.ws
+
+import androidx.annotation.VisibleForTesting
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.launch
+
+internal class ConnectionHealth(
+    private val scope: CoroutineScope,
+    private val isConnected: () -> Boolean,
+    private val pingRequest: suspend (Long) -> Unit,
+    private val cancelSocket: () -> Unit,
+    private val pendingReply: () -> Boolean,
+    private val nowMs: () -> Long = { System.nanoTime() / 1_000_000L },
+    private val log: (String) -> Unit = {},
+    private val debugLog: (String) -> Unit = {},
+) {
+    companion object {
+        internal const val HEARTBEAT_INTERVAL_MS = 15_000L
+        internal const val STALE_THRESHOLD_MS = 30_000L
+        internal const val LIVENESS_PROBE_TIMEOUT_MS = 5_000L
+    }
+
+    @Volatile
+    var lastPongTimestamp: Long = 0L
+        internal set
+
+    private val _lastLatencyMs = MutableStateFlow<Long?>(null)
+    val lastLatencyMs: StateFlow<Long?> = _lastLatencyMs.asStateFlow()
+
+    val isHealthy: Boolean
+        get() = isConnected() && (nowMs() - lastPongTimestamp < STALE_THRESHOLD_MS)
+
+    private var healthJob: Job? = null
+    private var foregroundProbeJob: Job? = null
+    private var transportProbeJob: Job? = null
+
+    fun start() {
+        healthJob?.cancel()
+        lastPongTimestamp = nowMs()
+        healthJob =
+            scope.launch {
+                while (isConnected()) {
+                    delay(HEARTBEAT_INTERVAL_MS)
+                    if (isConnected()) {
+                        try {
+                            ping(timeoutMs = LIVENESS_PROBE_TIMEOUT_MS)
+                        } catch (e: Exception) {
+                            log("Health check ping failed: ${e.message}")
+                        }
+                    }
+                    if (!runHealthCheckPass()) break
+                }
+            }
+    }
+
+    fun stop() {
+        healthJob?.cancel()
+        healthJob = null
+        _lastLatencyMs.value = null
+    }
+
+    fun onInboundFrame() {
+        lastPongTimestamp = nowMs()
+    }
+
+    suspend fun ping(timeoutMs: Long = LIVENESS_PROBE_TIMEOUT_MS): Long {
+        val start = nowMs()
+        pingRequest(timeoutMs)
+        val latency = (nowMs() - start).coerceAtLeast(0L)
+        lastPongTimestamp = nowMs()
+        _lastLatencyMs.value = latency
+        return latency
+    }
+
+    @VisibleForTesting
+    internal fun setLastLatencyForTest(latency: Long?) {
+        _lastLatencyMs.value = latency
+    }
+
+    private fun runHealthCheckPass(): Boolean {
+        if (!isConnected()) return false
+        val staleMs = nowMs() - lastPongTimestamp
+        if (staleMs <= STALE_THRESHOLD_MS) return true
+        log("WebSocket stale (${staleMs / 1000}s without frames) — cancelling to trigger reconnect")
+        cancelSocket()
+        return false
+    }
+
+    @VisibleForTesting
+    internal fun forceHealthCheckForTest(staleMillis: Long) {
+        lastPongTimestamp = nowMs() - staleMillis
+        runHealthCheckPass()
+    }
+
+    fun probeLivenessOnWake(deferredOnce: Boolean = false) {
+        if (!isConnected()) return
+        foregroundProbeJob?.cancel()
+        foregroundProbeJob =
+            scope.launch {
+                val alive = runCatching { ping(LIVENESS_PROBE_TIMEOUT_MS) }.isSuccess
+                if (alive) return@launch
+                if (pendingReply() && !deferredOnce) {
+                    delay(3_000L)
+                    probeLivenessOnWake(deferredOnce = true)
+                    return@launch
+                }
+                log("Foreground liveness probe failed — cancelling socket")
+                cancelSocket()
+            }
+    }
+
+    fun cancelForegroundProbe() {
+        foregroundProbeJob?.cancel()
+        foregroundProbeJob = null
+    }
+
+    fun probeLivenessOnTransportChange(
+        timeoutMs: Long = LIVENESS_PROBE_TIMEOUT_MS,
+        onFailureAction: () -> Unit = cancelSocket,
+    ) {
+        if (!isConnected()) return
+        transportProbeJob?.cancel()
+        transportProbeJob =
+            scope.launch {
+                debugLog("Network transport changed — probing WebSocket liveness")
+                val alive = runCatching { ping(timeoutMs) }.isSuccess
+                if (alive) {
+                    debugLog("WebSocket liveness probe succeeded on new transport")
+                    return@launch
+                }
+                log("Transport change liveness probe failed — cancelling socket")
+                onFailureAction()
+            }
+    }
+}

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

@@ -0,0 +1,39 @@
+package com.m57.hermescontrol.data.ws
+
+internal class ReconnectPolicy(
+    private val random: () -> Double = { Math.random() },
+) {
+    @Volatile
+    var currentBackoff: Long = INITIAL_BACKOFF_MS
+    private var initialBackoffMs: Long = INITIAL_BACKOFF_MS
+    private var testBackoffOverrideMs: Long? = null
+
+    fun nextDelayMs(): Long {
+        val override = testBackoffOverrideMs
+        if (override != null) return override
+        val ceiling = currentBackoff
+        if (ceiling <= 0L) return 0L
+        return (random() * ceiling).toLong().coerceAtLeast(BACKOFF_BASE_MS).coerceAtMost(ceiling)
+    }
+
+    fun advance() {
+        currentBackoff = (currentBackoff * BACKOFF_MULTIPLIER).toLong().coerceAtMost(MAX_BACKOFF_MS)
+    }
+
+    fun reset() {
+        currentBackoff = initialBackoffMs
+    }
+
+    fun setBackoffForTest(initialMillis: Long) {
+        initialBackoffMs = initialMillis
+        currentBackoff = initialMillis
+        testBackoffOverrideMs = initialMillis
+    }
+
+    companion object {
+        private const val INITIAL_BACKOFF_MS = 1_000L
+        private const val BACKOFF_BASE_MS = 300L
+        private const val MAX_BACKOFF_MS = 30_000L
+        private const val BACKOFF_MULTIPLIER = 2.0
+    }
+}

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

@@ -0,0 +1,245 @@
+package com.m57.hermescontrol.data.ws
+
+import androidx.annotation.VisibleForTesting
+import com.m57.hermescontrol.data.ws.contract.SessionEventsSinceParams
+import com.m57.hermescontrol.data.ws.contract.SessionEventsSinceResult
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+import kotlinx.serialization.json.JsonObject
+import java.util.Collections
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * Tracks session event sequence watermarks and orchestrates event replay on reconnect.
+ *
+ * Extracted from [HermesWsClient] (issue #1016, issue #1376).
+ */
+internal class ReplayTracker(
+    private val scope: CoroutineScope,
+    private val fetchEventsSince: suspend (SessionEventsSinceParams) -> SessionEventsSinceResult,
+    private val emitEvent: (WsEvent) -> Unit,
+    private val resolveStoredSessionId: (String?) -> String? = { it },
+    private val onReplayFailure: (String, Exception) -> Unit = { _, _ -> },
+) {
+    private val lastSeenSeq = ConcurrentHashMap<String, Int>()
+
+    @Volatile
+    private var replayEpoch: String? = null
+
+    @Volatile
+    private var replayInFlight: Boolean = false
+
+    private val replayHold = ConcurrentHashMap<String, MutableList<Pair<Int?, WsEvent>>>()
+
+    @Volatile
+    private var replayJob: Job? = null
+
+    /**
+     * Synchronously arms the hold buffers for all known sessions and launches replay in [scope].
+     * Arming synchronously ensures racing live frames are held instead of advancing watermarks.
+     */
+    fun triggerReplay() {
+        if (lastSeenSeq.isEmpty() || replayInFlight) return
+        replayInFlight = true
+        val sessionsToReplay = lastSeenSeq.keys().toList()
+        for (sid in sessionsToReplay) {
+            replayHold[sid] = Collections.synchronizedList(mutableListOf())
+        }
+        replayJob?.cancel()
+        replayJob =
+            scope.launch {
+                fetchReplay(sessionsToReplay)
+            }
+    }
+
+    /**
+     * Updates replay epoch from gateway.ready.
+     * When [epoch] is non-empty and changes from a previous non-null value, resets sequence watermarks.
+     */
+    fun updateEpoch(epoch: String?) {
+        if (!epoch.isNullOrEmpty()) {
+            if (replayEpoch != null && replayEpoch != epoch) {
+                lastSeenSeq.clear()
+            }
+            replayEpoch = epoch
+        }
+    }
+
+    /**
+     * Checks if an incoming event is a duplicate against the current watermark for [sessionId].
+     * Returns true if [seq] <= watermark.
+     */
+    fun isDuplicate(
+        sessionId: String,
+        seq: Int,
+    ): Boolean {
+        val prev = lastSeenSeq[sessionId] ?: 0
+        return seq <= prev
+    }
+
+    /**
+     * Processes an incoming live event with sequence tracking.
+     * Evaluates [eventFactory] only after checking for duplicates.
+     * If duplicate, drops it without calling [eventFactory].
+     * If replay is in flight for [sessionId], buffers it in replayHold.
+     * Otherwise advances watermark and emits the event.
+     */
+    fun acceptLiveEvent(
+        sessionId: String,
+        seq: Int,
+        eventFactory: () -> WsEvent,
+    ) {
+        if (isDuplicate(sessionId, seq)) {
+            return
+        }
+        val finalEvent = decorateEvent(eventFactory())
+        if (replayInFlight && replayHold.containsKey(sessionId)) {
+            replayHold[sessionId]?.add(seq to finalEvent)
+            return
+        }
+        lastSeenSeq[sessionId] = seq
+        emitEvent(finalEvent)
+    }
+
+    /**
+     * Processes an incoming live event with sequence tracking.
+     * If duplicate, drops it.
+     * If replay is in flight for [sessionId], buffers it in replayHold.
+     * Otherwise advances watermark and emits the event.
+     */
+    fun acceptLiveEvent(
+        sessionId: String,
+        seq: Int,
+        event: WsEvent,
+    ) {
+        acceptLiveEvent(sessionId, seq) { event }
+    }
+
+    private fun decorateEvent(event: WsEvent): WsEvent =
+        if (event is WsEvent.MessageComplete) {
+            event.copy(
+                storedSessionId = resolveStoredSessionId(event.sessionId),
+            )
+        } else {
+            event
+        }
+
+    private suspend fun fetchReplay(sessionsToReplay: List<String> = lastSeenSeq.keys().toList()) {
+        try {
+            for (sid in sessionsToReplay) {
+                val lastSeen = lastSeenSeq[sid] ?: continue
+                try {
+                    val res =
+                        fetchEventsSince(
+                            SessionEventsSinceParams(sessionId = sid, lastSeen = lastSeen),
+                        )
+
+                    val epoch = res.epoch
+                    val epochChanged =
+                        !epoch.isNullOrEmpty() && replayEpoch != null && replayEpoch != epoch
+                    val truncated = res.truncated == true
+                    val latestSeq = res.latestSeq
+
+                    if (epochChanged) replayEpoch = epoch
+
+                    if (epochChanged || truncated) {
+                        // Ring buffer (512 events) could not cover the gap or epoch changed (gateway restarted).
+                        // Partial replay would silently hole the transcript — fast-forward watermark
+                        // and request full transcript resync.
+                        if (epochChanged) lastSeenSeq.clear()
+                        if (latestSeq != null && !epochChanged) lastSeenSeq[sid] = latestSeq
+                        emitEvent(WsEvent.TranscriptResyncRequired(sid))
+                        continue
+                    }
+                    val eventsList = res.events
+                    if (eventsList != null) {
+                        for (element in eventsList) {
+                            @Suppress("UNCHECKED_CAST")
+                            val eventMap = (element as? JsonObject)?.toAny() as? Map<String, Any?> ?: continue
+                            val eventSeq = (eventMap["seq"] as? Number)?.toInt()
+                            val eventSid = (eventMap["session_id"] as? String) ?: sid
+                            if (eventSeq != null && eventSid.isNotEmpty()) {
+                                val prev = lastSeenSeq[eventSid] ?: 0
+                                if (eventSeq <= prev) continue
+                                lastSeenSeq[eventSid] = eventSeq
+                            }
+                            val parsedEvent = EventParser.parseParams(eventMap)
+                            val finalEvent = decorateEvent(parsedEvent)
+                            emitEvent(finalEvent)
+                        }
+                    }
+                } catch (e: Exception) {
+                    onReplayFailure(sid, e)
+                    emitEvent(WsEvent.TranscriptResyncRequired(sid))
+                }
+            }
+        } finally {
+            withContext(NonCancellable) {
+                for (sid in sessionsToReplay) {
+                    val held = replayHold.remove(sid)
+                    if (held != null) {
+                        synchronized(held) {
+                            for ((heldSeq, heldEvent) in held) {
+                                if (heldSeq != null) {
+                                    val prev = lastSeenSeq[sid] ?: 0
+                                    if (heldSeq <= prev) continue
+                                    lastSeenSeq[sid] = heldSeq
+                                }
+                                emitEvent(heldEvent)
+                            }
+                        }
+                    }
+                }
+                replayInFlight = false
+            }
+        }
+    }
+
+    /**
+     * Complete teardown matching disconnect (issue #1016).
+     * Cancels any active replay job and clears all holds, watermarks, and epoch.
+     */
+    fun clear() {
+        lastSeenSeq.clear()
+        replayHold.clear()
+        replayEpoch = null
+        replayJob?.cancel()
+        replayJob = null
+        replayInFlight = false
+    }
+
+    @VisibleForTesting
+    internal fun getSeqWatermarks(): Map<String, Int> = HashMap(lastSeenSeq)
+
+    @VisibleForTesting
+    internal fun setSeqWatermark(
+        sessionId: String,
+        seq: Int,
+    ) {
+        lastSeenSeq[sessionId] = seq
+    }
+
+    @VisibleForTesting
+    internal fun clearSeqWatermarks() {
+        lastSeenSeq.clear()
+        replayHold.clear()
+        replayEpoch = null
+        replayInFlight = false
+    }
+
+    @VisibleForTesting
+    internal fun isReplayInFlight(): Boolean = replayInFlight
+
+    @VisibleForTesting
+    internal fun setReplayEpochForTest(epoch: String?) {
+        replayEpoch = epoch
+    }
+
+    @VisibleForTesting
+    internal suspend fun fetchReplayForTest() {
+        fetchReplay()
+    }
+}

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

@@ -0,0 +1,353 @@
+package com.m57.hermescontrol.data.ws
+
+import com.m57.hermescontrol.data.ws.contract.RpcMethod
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.serialization.decodeFromString
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
+import kotlinx.serialization.json.buildJsonObject
+import kotlinx.serialization.json.put
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger
+
+/**
+ * Socket-independent JSON-RPC 2.0 request correlation and response dispatcher.
+ * Manages request IDs, in-flight pending calls, timeouts, cancellation cleanup,
+ * and server response frame formatting.
+ */
+internal class RpcChannel(
+    private val scope: CoroutineScope,
+    private val json: Json,
+    private val sendRequest: (method: String, params: Map<String, Any>, onSent: ((String) -> Unit)?) -> String,
+    private val sendFrame: (String) -> Boolean,
+    private val removeQueuedMessage: (String) -> Unit,
+    private val onIdleCheck: () -> Unit,
+    private val logRejected: (method: String, id: String) -> Unit = { _, _ -> },
+    private val outboundLock: Any = Any(),
+    private val onResult: (String) -> Unit = {},
+    private val onError: (String) -> Unit = {},
+    private val resolveStoredSessionId: (String?) -> String? = { it },
+    private val logParseFailure: (Exception) -> Unit = {},
+    private val decorateParams: (String, Map<String, Any>) -> Map<String, Any> = { _, p -> p },
+) {
+    /** 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,
+        val suppressErrorEvent: Boolean = false,
+    )
+
+    private val requestId = AtomicInteger(0)
+    private val pendingCalls = ConcurrentHashMap<String, PendingCall>()
+    private val capabilityRequestIds = ConcurrentHashMap.newKeySet<String>()
+
+    fun nextId(): String = requestId.incrementAndGet().toString()
+
+    fun hasPendingCalls(): Boolean = pendingCalls.isNotEmpty()
+
+    fun pendingCallIdsForTest(): Set<String> = pendingCalls.keys.toSet()
+
+    fun pendingCallTimeoutForTest(id: String): Job? = pendingCalls[id]?.timeoutJob
+
+    fun isErrorEventSuppressed(id: String): Boolean = pendingCalls[id]?.suppressErrorEvent == true
+
+    fun registerCapabilityRequest(id: String) {
+        capabilityRequestIds.add(id)
+    }
+
+    fun clearCapabilityRequests() {
+        capabilityRequestIds.clear()
+    }
+
+    fun consumeCapabilityResponse(event: WsEvent): Boolean {
+        val id =
+            when (event) {
+                is WsEvent.RpcResult -> event.id
+                is WsEvent.RpcError -> event.id
+                else -> return false
+            }
+        val isCapabilityResponse =
+            synchronized(outboundLock) {
+                capabilityRequestIds.remove(id)
+            }
+        if (!isCapabilityResponse) return false
+
+        removeQueuedMessage(id)
+        return true
+    }
+
+    fun createRequest(
+        method: String,
+        params: Map<String, Any>,
+    ): JsonRpcRequest {
+        val id = nextId()
+        val decorated = decorateParams(method, params)
+        return JsonRpcRequest(
+            id = id,
+            method = method,
+            params = decorated.mapValues { it.value.toJsonElement() },
+        )
+    }
+
+    fun encodeRequest(request: JsonRpcRequest): String = json.encodeToString(request)
+
+    /**
+     * Send a JSON-RPC request that expects a result and returns a [CompletableDeferred] for it.
+     */
+    fun request(
+        method: String,
+        params: Map<String, Any> = emptyMap(),
+        timeoutMs: Long = HermesWsClient.REQUEST_TIMEOUT_MS,
+        suppressErrorEvent: Boolean = false,
+    ): CompletableDeferred<Any?> {
+        val deferred = CompletableDeferred<Any?>()
+        val id =
+            sendRequest(method, params) { reqId ->
+                pendingCalls[reqId] = PendingCall(method, deferred, suppressErrorEvent = suppressErrorEvent)
+            }
+        deferred.invokeOnCompletion { cause ->
+            if (cause is CancellationException) {
+                pendingCalls.remove(id)?.let { call ->
+                    call.timeoutJob?.cancel()
+                    removeQueuedMessage(id)
+                    onIdleCheck()
+                }
+            }
+        }
+        // Arm the per-request timeout (fires if the server never answers).
+        pendingCalls[id]?.timeoutJob =
+            scope.launch {
+                delay(timeoutMs)
+                resolvePending(id, null, JsonRpcError(-1, "Request timed out: $method"))
+            }
+        return deferred
+    }
+
+    /**
+     * Non-suspending typed twin of [request] for callers that must enqueue inside a
+     * non-suspending owner check and await the raw result themselves.
+     */
+    fun <P, R> requestTyped(
+        method: RpcMethod<P, R>,
+        params: P,
+        timeoutMs: Long = HermesWsClient.REQUEST_TIMEOUT_MS,
+        suppressErrorEvent: Boolean = false,
+    ): CompletableDeferred<Any?> = request(method.name, encodeParams(method, params), timeoutMs, suppressErrorEvent)
+
+    /**
+     * Send a typed JSON-RPC request and await its deserialized result.
+     */
+    suspend fun <P, R> call(
+        method: RpcMethod<P, R>,
+        params: P,
+        timeoutMs: Long = HermesWsClient.REQUEST_TIMEOUT_MS,
+        suppressErrorEvent: Boolean = false,
+    ): R = awaitResult(method, requestTyped(method, params, timeoutMs, suppressErrorEvent))
+
+    suspend fun <P, R> awaitResult(
+        method: RpcMethod<P, R>,
+        deferred: CompletableDeferred<Any?>,
+    ): R {
+        val result =
+            try {
+                deferred.await()
+            } catch (e: CancellationException) {
+                deferred.cancel(e)
+                throw e
+            }
+        val element = result.toJsonElement()
+        return json.decodeFromJsonElement(method.result, element)
+    }
+
+    fun <P, R> encodeParams(
+        method: RpcMethod<P, R>,
+        params: P,
+    ): JsonObject {
+        val encoded = json.encodeToJsonElement(method.params, params)
+        require(encoded is JsonObject) {
+            "RPC params for ${method.name} must serialize to a JsonObject, got ${encoded::class.simpleName}"
+        }
+        return encoded
+    }
+
+    /** Complete (or fail) a single pending call and cancel its timer. */
+    fun resolvePending(
+        id: String,
+        result: Any?,
+        error: JsonRpcError?,
+    ) {
+        val call = pendingCalls.remove(id) ?: return
+        removeQueuedMessage(id)
+        call.timeoutJob?.cancel()
+        if (error != null) {
+            call.deferred.completeExceptionally(
+                HermesWsClient.HermesRpcException(
+                    message = error.message,
+                    code = error.code,
+                    data = error.data,
+                ),
+            )
+        } else {
+            call.deferred.complete(result)
+        }
+        onIdleCheck()
+    }
+
+    /**
+     * Fail and clear every in-flight [request].
+     */
+    fun rejectAllPending(
+        error: HermesWsClient.HermesRpcException =
+            HermesWsClient.HermesRpcException("Connection lost — request cancelled"),
+    ) {
+        if (pendingCalls.isEmpty()) return
+        val snapshot = pendingCalls.toList()
+        pendingCalls.clear()
+        for ((id, call) in snapshot) {
+            removeQueuedMessage(id)
+            logRejected(call.method, id)
+            call.timeoutJob?.cancel()
+            call.deferred.completeExceptionally(error)
+        }
+    }
+
+    /**
+     * Answer a gateway-originated JSON-RPC request.
+     */
+    fun respondToServerRequest(
+        id: String,
+        result: JsonElement,
+    ): Boolean {
+        if (id.isBlank()) return false
+        val frameJson =
+            buildJsonObject {
+                put("jsonrpc", "2.0")
+                put("id", id)
+                put("result", result)
+            }.toString()
+        return sendFrame(frameJson)
+    }
+
+    /** Answer a gateway-originated request with a standard JSON-RPC error. */
+    fun respondToServerRequestError(
+        id: String,
+        code: Int,
+        message: String,
+    ): Boolean {
+        if (id.isBlank()) return false
+        val frameJson =
+            buildJsonObject {
+                put("jsonrpc", "2.0")
+                put("id", id)
+                put(
+                    "error",
+                    buildJsonObject {
+                        put("code", code)
+                        put("message", message)
+                    },
+                )
+            }.toString()
+        return sendFrame(frameJson)
+    }
+
+    /**
+     * Handles an incoming WebSocket text frame: JSON decoding, request correlation,
+     * server request dispatch, sequence deduplication via [replayTracker], and event emission.
+     */
+    fun handleIncomingFrame(
+        text: String,
+        replayTracker: ReplayTracker,
+        emitEvent: (WsEvent) -> Unit,
+    ) {
+        val event =
+            try {
+                val rpc = json.decodeFromString<JsonRpcResponse>(text)
+                // Resolve pending request() calls with the RAW JsonElement from
+                // rpc.result, BEFORE EventParser.parse() converts it via toAny().
+                // This preserves integer/float type fidelity needed by callers
+                // that re-deserialize the result into typed data classes.
+                val rpcId = rpc.id
+                if (rpcId != null && rpc.error == null && rpc.result != null) {
+                    onResult(rpcId)
+                    removeQueuedMessage(rpcId)
+                    resolvePending(rpcId, rpc.result, null)
+                }
+
+                // Resume/replay responses carry still-open server requests
+                // separately from the event ring. Re-deliver them through the
+                // same generic path before exposing the RPC result.
+                @Suppress("UNCHECKED_CAST")
+                val openRequests =
+                    (rpc.result?.toAny() as? Map<*, *>)?.get("open_requests") as? List<Map<*, *>>
+                openRequests?.forEach { openRequest ->
+                    val openId = openRequest["id"] as? String ?: return@forEach
+                    val method = openRequest["method"] as? String ?: return@forEach
+
+                    @Suppress("UNCHECKED_CAST")
+                    val params = openRequest["params"] as? Map<String, Any?> ?: emptyMap()
+                    emitEvent(WsEvent.ServerRequest(openId, method, params, replayed = true))
+                }
+
+                if (rpc.id == null) {
+                    // Issue #1163: inspect scalars, not a throwaway copy of the whole params tree.
+                    val sid = rpc.params?.eventSessionId()
+                    val seq = ((rpc.params?.get("seq") as? JsonPrimitive)?.toAny() as? Number)?.toInt()
+
+                    if (!sid.isNullOrBlank() && seq != null) {
+                        replayTracker.acceptLiveEvent(sid, seq) {
+                            EventParser.parse(rpc, text)
+                        }
+                        return
+                    }
+                }
+
+                val parsed = EventParser.parse(rpc, text)
+                if (parsed is WsEvent.MessageComplete) {
+                    parsed.copy(
+                        storedSessionId = resolveStoredSessionId(parsed.sessionId),
+                    )
+                } else {
+                    parsed
+                }
+            } catch (e: Exception) {
+                logParseFailure(e)
+                WsEvent.Unknown(text)
+            }
+        if (consumeCapabilityResponse(event)) return
+        when (event) {
+            is WsEvent.RpcResult -> {
+                onResult(event.id)
+                removeQueuedMessage(event.id)
+                resolvePending(event.id, event.result, null)
+            }
+
+            is WsEvent.RpcError -> {
+                onError(event.id)
+                if (pendingCalls[event.id]?.suppressErrorEvent == true) {
+                    // Opt-in suppression: the caller already handles the
+                    // failure through the CompletableDeferred, so the event
+                    // copy would only surface a duplicate UI banner for
+                    // optional features (e.g. "subagent.list" on gateways
+                    // without the method, issue #1089). Default request()
+                    // behavior still emits the event for shared consumers.
+                    resolvePending(event.id, null, event.error)
+                    return
+                }
+                removeQueuedMessage(event.id)
+                resolvePending(event.id, null, event.error)
+            }
+
+            else -> {}
+        }
+        emitEvent(event)
+    }
+}

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

@@ -0,0 +1,912 @@
+package com.m57.hermescontrol.data.ws
+
+import android.util.Log
+import androidx.annotation.VisibleForTesting
+import com.m57.hermescontrol.BuildConfig
+import com.m57.hermescontrol.data.local.AuthManager
+import com.m57.hermescontrol.data.remote.CookieManager
+import com.m57.hermescontrol.data.remote.DashboardSessionTokenRefresher
+import com.m57.hermescontrol.data.remote.NetworkMonitor
+import com.m57.hermescontrol.data.remote.OkHttpProvider
+import com.m57.hermescontrol.data.remote.await
+import com.m57.hermescontrol.data.session.ActiveSessionHolder
+import com.m57.hermescontrol.data.ws.contract.PromptSubmitParams
+import com.m57.hermescontrol.data.ws.contract.RpcMethods
+import com.m57.hermescontrol.data.ws.contract.SessionCorrectionParams
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+import kotlinx.serialization.decodeFromString
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
+import okhttp3.Request
+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
+import java.util.concurrent.atomic.AtomicInteger
+
+/**
+ * Socket lifecycle, authentication, outbound queueing, and network transport management.
+ *
+ * Extracted from [HermesWsClient] (issue #1376).
+ */
+internal class WsTransport(
+    private val scope: CoroutineScope,
+    private val outboundLock: Any,
+    private val reconnectPolicy: ReconnectPolicy,
+    private val connectionHealth: ConnectionHealth,
+    private val replayTracker: ReplayTracker,
+    private val rpcChannel: RpcChannel,
+    private val onRawMessage: (String) -> Unit,
+) {
+    companion object {
+        private const val TAG = "HermesWsClient"
+        private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024
+        private const val OUTBOUND_DRAIN_TIMEOUT_MS = 60_000L
+    }
+
+    private val connectionGeneration = AtomicInteger(0)
+    private val connected = AtomicBoolean(false)
+    private val intentionalClose = AtomicBoolean(false)
+    private val acceptQueuedMessages = AtomicBoolean(true)
+    private val appInForeground = AtomicBoolean(true)
+    private val externalActivityConnectionLease = AtomicBoolean(false)
+    private val backgroundConnectionLease = AtomicBoolean(false)
+    private val messageQueue = ConcurrentLinkedQueue<String>()
+    private val queuedMessagesById = ConcurrentHashMap<String, String>()
+
+    @Volatile
+    private var webSocket: WebSocket? = null
+
+    @Volatile
+    private var closingSocket: WebSocket? = null
+
+    @Volatile
+    private var reconnectJob: Job? = null
+
+    @Volatile
+    private var outboundDrainJob: Job? = null
+
+    private val networkCollectorStarted = AtomicBoolean(false)
+
+    private val _connectionStatus = MutableStateFlow(ConnectionStatus.DISCONNECTED)
+    val connectionStatus: StateFlow<ConnectionStatus> = _connectionStatus.asStateFlow()
+
+    @Volatile
+    var pendingReply: Boolean = false
+        internal set
+
+    private val pendingPromptSubmits = ConcurrentHashMap.newKeySet<String>()
+
+    val isConnected: Boolean get() = connected.get()
+
+    fun setAppForeground(foreground: Boolean) {
+        appInForeground.set(foreground)
+        if (!foreground) {
+            disconnectIfIdleInBackground()
+            return
+        }
+        connectionHealth.probeLivenessOnWake()
+    }
+
+    fun acquireExternalActivityConnectionLease() {
+        externalActivityConnectionLease.set(true)
+    }
+
+    fun releaseExternalActivityConnectionLease() {
+        externalActivityConnectionLease.set(false)
+        disconnectIfIdleInBackground()
+    }
+
+    fun acquireBackgroundConnectionLease() {
+        backgroundConnectionLease.set(true)
+    }
+
+    fun releaseBackgroundConnectionLease() {
+        backgroundConnectionLease.set(false)
+        disconnectIfIdleInBackground()
+    }
+
+    internal fun hasBackgroundConnectionLease(): Boolean = backgroundConnectionLease.get()
+
+    fun disconnectIfIdleInBackground() {
+        synchronized(outboundLock) {
+            if (!appInForeground.get() &&
+                !externalActivityConnectionLease.get() &&
+                !backgroundConnectionLease.get() &&
+                !pendingReply &&
+                !rpcChannel.hasPendingCalls() &&
+                messageQueue.isEmpty()
+            ) {
+                disconnect()
+            }
+        }
+    }
+
+    fun connect() {
+        if (networkCollectorStarted.compareAndSet(false, true)) {
+            scope.launch {
+                NetworkMonitor.networkChanges.collect { networkAvailable ->
+                    reconnectForNetworkChange(networkAvailable)
+                }
+            }
+            scope.launch {
+                NetworkMonitor.transportChanges.collect {
+                    connectionHealth.probeLivenessOnTransportChange(
+                        onFailureAction = { cancelSocket() },
+                    )
+                }
+            }
+        }
+        var socketToCancel: WebSocket? = null
+        val shouldOpen =
+            synchronized(outboundLock) {
+                acceptQueuedMessages.set(true)
+                if (_connectionStatus.value == ConnectionStatus.AUTH_EXPIRED) {
+                    connectionGeneration.incrementAndGet()
+                    reconnectJob?.cancel()
+                    reconnectJob = null
+                    outboundDrainJob?.cancel()
+                    outboundDrainJob = null
+                    socketToCancel = webSocket
+                    webSocket = null
+                    closingSocket = null
+                    connected.set(false)
+                    connectionHealth.stop()
+                }
+                if (connected.get()) {
+                    Log.d(TAG, "Already connected — skipping")
+                    return@synchronized false
+                }
+                if (_connectionStatus.value == ConnectionStatus.CONNECTING ||
+                    _connectionStatus.value == ConnectionStatus.RECONNECTING
+                ) {
+                    Log.d(TAG, "Connection already in flight (${_connectionStatus.value}) — skipping")
+                    return@synchronized false
+                }
+                // An explicit connect follows successful authentication and is
+                // therefore allowed to leave the terminal AUTH_EXPIRED state.
+                intentionalClose.set(false)
+                _connectionStatus.value = ConnectionStatus.CONNECTING
+                reconnectPolicy.reset()
+                true
+            }
+        if (socketToCancel != null) {
+            rpcChannel.rejectAllPending()
+            socketToCancel.cancel()
+        }
+        if (shouldOpen) openSocket()
+    }
+
+    fun reconnectForNetworkChange(
+        networkAvailable: Boolean = NetworkMonitor.isConnected.value,
+        openReplacement: () -> Unit = ::openSocket,
+    ) {
+        val socketToCancel: WebSocket?
+        val shouldOpen: Boolean
+        synchronized(outboundLock) {
+            if (intentionalClose.get() || _connectionStatus.value == ConnectionStatus.AUTH_EXPIRED) {
+                return
+            }
+            Log.d(TAG, "Default network changed — reconnecting WebSocket")
+            reconnectPolicy.reset()
+            reconnectJob?.cancel()
+            reconnectJob = null
+            outboundDrainJob?.cancel()
+            outboundDrainJob = null
+            connectionGeneration.incrementAndGet()
+            socketToCancel = webSocket
+            webSocket = null
+            closingSocket = null
+            connected.set(false)
+            connectionHealth.stop()
+            shouldOpen = networkAvailable && AuthManager.isAutoReconnect()
+            _connectionStatus.value =
+                when {
+                    !networkAvailable -> ConnectionStatus.NO_NETWORK
+                    shouldOpen -> ConnectionStatus.RECONNECTING
+                    else -> ConnectionStatus.DISCONNECTED
+                }
+            rpcChannel.rejectAllPending()
+        }
+        socketToCancel?.cancel()
+        if (shouldOpen) openReplacement()
+    }
+
+    suspend fun refreshWsTicketIfNeeded(generation: Int): Boolean {
+        val isGated =
+            try {
+                AuthManager.serverStore.getLatestState().wsAuthParam == "ticket"
+            } catch (_: IllegalStateException) {
+                // serverStore not initialized yet (transient early call); treat
+                // as loopback so a stale-token refresh still runs. Log so a real
+                // misconfiguration isn't silently swallowed.
+                Log.w(TAG, "serverStore uninitialized during WS handshake; assuming non-gated")
+                false
+            }
+        if (!isGated) {
+            // The loopback dashboard token is regenerated on every server
+            // restart. Refresh it before each WebSocket handshake so automatic
+            // reconnect does not get stuck in AUTH_EXPIRED with a stale token.
+            val token =
+                runCatching {
+                    DashboardSessionTokenRefresher.refreshAsync()
+                }.getOrNull()
+            synchronized(outboundLock) {
+                if (isCurrentGeneration(generation) && token != null) {
+                    runCatching { AuthManager.setToken(token) }
+                }
+            }
+            return true
+        }
+
+        val ticketResult = requestWsTicket()
+        if (!ticketResult.ticket.isNullOrBlank()) {
+            synchronized(outboundLock) {
+                if (!isCurrentGeneration(generation)) return false
+                AuthManager.setToken(ticketResult.ticket)
+            }
+            if (BuildConfig.DEBUG) Log.d(TAG, "WS ticket refreshed")
+            return true
+        }
+        val status =
+            when {
+                ticketResult.exception -> ConnectionStatus.RECONNECTING
+                ticketResult.httpCode == 401 || ticketResult.httpCode == 403 -> ConnectionStatus.AUTH_EXPIRED
+                ticketResult.httpCode == 408 || ticketResult.httpCode == 429 -> ConnectionStatus.RECONNECTING
+                ticketResult.httpCode != null && ticketResult.httpCode >= 500 -> ConnectionStatus.RECONNECTING
+                else -> ConnectionStatus.DISCONNECTED
+            }
+        return handleWsTicketRefreshFailure(status, generation)
+    }
+
+    suspend fun mintWsTicket(): String? {
+        val isGated =
+            try {
+                AuthManager.serverStore.getLatestState().wsAuthParam == "ticket"
+            } catch (_: IllegalStateException) {
+                false
+            }
+        if (!isGated) {
+            DashboardSessionTokenRefresher.refreshAsync()
+            return AuthManager.getToken()
+        }
+        return requestWsTicket().ticket
+    }
+
+    private data class TicketRequestResult(
+        val ticket: String?,
+        val httpCode: Int?,
+        val exception: Boolean = false,
+    )
+
+    private suspend fun requestWsTicket(): TicketRequestResult {
+        try {
+            val client = OkHttpProvider.probe
+            val request =
+                Request
+                    .Builder()
+                    .url(AuthManager.endpointForBuild().resolve("api/auth/ws-ticket").toString())
+                    .post("{}".toRequestBody())
+                    .build()
+
+            val (code, body) =
+                withContext(Dispatchers.IO) {
+                    if (CookieManager.isInitialized()) {
+                        CookieManager.useStore(CookieManager.cookieJar.currentServer())
+                    }
+                    client.newCall(request).await().use { resp ->
+                        resp.code to resp.body.string()
+                    }
+                }
+
+            if (code in 200..299) {
+                // Real JSON decode — the ticket is base64url and may contain
+                // characters a naive [^"]+ regex cannot extract (see
+                // testGatedMode_parsesEscapedTicketFromRealJsonShape).
+                val parsed =
+                    runCatching {
+                        OkHttpProvider.json.decodeFromString<WsTicketResponse>(body)
+                    }.getOrNull()
+                val ticket = parsed?.ticket
+                if (!ticket.isNullOrBlank()) {
+                    return TicketRequestResult(ticket, null)
+                }
+                Log.w(TAG, "WS ticket mint failed: unparseable ticket response")
+            } else {
+                Log.w(TAG, "WS ticket mint failed: HTTP $code")
+                return TicketRequestResult(null, code)
+            }
+        } catch (e: Exception) {
+            // Include the stack: the exception class alone (e.g.
+            // NetworkOnMainThreadException) can't show WHICH caller on what
+            // thread tripped the mint (main-scope kanban events connect vs
+            // the WS reconnect path).
+            Log.w(TAG, "WS ticket mint failed: ${e.javaClass.simpleName}", e)
+            return TicketRequestResult(null, null, exception = true)
+        }
+        return TicketRequestResult(null, null)
+    }
+
+    private fun handleWsTicketRefreshFailure(
+        status: ConnectionStatus,
+        generation: Int,
+    ): Boolean {
+        synchronized(outboundLock) {
+            if (!isCurrentGeneration(generation)) return false
+            _connectionStatus.value = status
+            if (status == ConnectionStatus.RECONNECTING) scheduleReconnect()
+        }
+        return false
+    }
+
+    fun disconnect(clearPendingMessages: Boolean = false) {
+        ActiveSessionHolder.clear()
+        synchronized(outboundLock) {
+            intentionalClose.set(true)
+            acceptQueuedMessages.set(!clearPendingMessages)
+            connectionGeneration.incrementAndGet()
+            reconnectJob?.cancel()
+            reconnectJob = null
+            outboundDrainJob?.cancel()
+            outboundDrainJob = null
+            connectionHealth.cancelForegroundProbe()
+            connectionHealth.stop()
+            webSocket?.close(1000, "Client closed")
+            webSocket = null
+            closingSocket = null
+            rpcChannel.clearCapabilityRequests()
+            if (clearPendingMessages) {
+                backgroundConnectionLease.set(false)
+                messageQueue.clear()
+                queuedMessagesById.clear()
+                pendingPromptSubmits.clear()
+                pendingReply = false
+                replayTracker.clear()
+            }
+            connected.set(false)
+            _connectionStatus.value = ConnectionStatus.DISCONNECTED
+        }
+    }
+
+    fun cancelSocket() {
+        synchronized(outboundLock) {
+            if (connected.get()) webSocket?.cancel()
+        }
+    }
+
+    fun sendDirectFrame(frameJson: String): Boolean =
+        synchronized(outboundLock) {
+            val ws = webSocket
+            if (ws == null || !connected.get()) false else ws.send(frameJson)
+        }
+
+    fun send(
+        request: JsonRpcRequest,
+        onSent: ((String) -> Unit)? = null,
+        queueIfDisconnected: Boolean = true,
+    ): String {
+        val id = request.id
+        val method = request.method
+        val json = rpcChannel.encodeRequest(request)
+        if (BuildConfig.DEBUG) {
+            Log.d(TAG, formatSafeOutgoingFrameLog(id, method, request.params.keys, json.length, queued = false))
+        }
+        var reconnect = false
+        synchronized(outboundLock) {
+            if (method == WsMethods.PROMPT_SUBMIT) {
+                pendingPromptSubmits.add(id)
+                pendingReply = true
+            }
+            val ws = webSocket
+            if (ws != null && connected.get()) {
+                // Never replay send(true): the server may have executed it even if its response is lost.
+                if (ws.send(json)) {
+                    onSent?.invoke(id)
+                } else {
+                    if (webSocket !== ws || !acceptQueuedMessages.get()) return@synchronized
+                    if (queueIfDisconnected && isRetryableMessage(json)) {
+                        onSent?.invoke(id)
+                        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() && queueIfDisconnected) {
+                if (isRetryableMessage(json)) {
+                    onSent?.invoke(id)
+                    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
+    }
+
+    fun removeQueuedMessage(id: String) {
+        synchronized(outboundLock) {
+            queuedMessagesById.remove(id)?.let(messageQueue::remove)
+        }
+    }
+
+    fun onRpcResult(id: String) {
+        synchronized(outboundLock) {
+            pendingPromptSubmits.remove(id)
+        }
+    }
+
+    fun onRpcError(id: String) {
+        synchronized(outboundLock) {
+            if (pendingPromptSubmits.remove(id) && pendingPromptSubmits.isEmpty()) {
+                pendingReply = false
+                disconnectIfIdleInBackground()
+            }
+        }
+    }
+
+    fun observeEvent(event: WsEvent) {
+        when (event) {
+            is WsEvent.MessageStart,
+            is WsEvent.MessageToken,
+            is WsEvent.ThinkingDelta,
+            is WsEvent.ReasoningDelta,
+            is WsEvent.ToolStart,
+            -> {
+                pendingReply = true
+            }
+
+            is WsEvent.MessageComplete -> {
+                pendingReply = false
+                disconnectIfIdleInBackground()
+            }
+
+            else -> {}
+        }
+    }
+
+    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,
+    ) {
+        if (queuedMessagesById.putIfAbsent(id, json) == null) messageQueue.add(json)
+    }
+
+    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 =
+            scope.launch {
+                delay(OUTBOUND_DRAIN_TIMEOUT_MS)
+                synchronized(outboundLock) {
+                    if (!connected.get() && !intentionalClose.get() && webSocket === ws) {
+                        ws.cancel()
+                        outboundDrainJob = null
+                        scheduleReconnect()
+                    }
+                }
+            }
+    }
+
+    fun sendMessage(
+        sessionId: String,
+        text: String,
+        onSent: ((String) -> Unit)? = null,
+        queued: Boolean = false,
+    ): String =
+        send(
+            rpcChannel.createRequest(
+                method = RpcMethods.PROMPT_SUBMIT.name,
+                params =
+                    PromptSubmitParams(
+                        sessionId = sessionId,
+                        text = text,
+                        queued = queued.takeIf { it },
+                    ).let { rpcChannel.encodeParams(RpcMethods.PROMPT_SUBMIT, it) },
+            ),
+            onSent = onSent,
+        )
+
+    fun sendRedirect(
+        sessionId: String,
+        text: String,
+        onSent: ((String) -> Unit)? = null,
+    ): String =
+        send(
+            rpcChannel.createRequest(
+                method = RpcMethods.SESSION_REDIRECT.name,
+                params =
+                    SessionCorrectionParams(
+                        sessionId = sessionId,
+                        text = text,
+                    ).let { rpcChannel.encodeParams(RpcMethods.SESSION_REDIRECT, it) },
+            ),
+            onSent = onSent,
+        )
+
+    internal fun openSocket() {
+        val generation =
+            synchronized(outboundLock) {
+                if (intentionalClose.get()) return
+                connectionGeneration.incrementAndGet()
+            }
+        scope.launch(Dispatchers.IO) {
+            if (!refreshWsTicketIfNeeded(generation)) {
+                Log.w(TAG, "Aborting openSocket: WS ticket refresh failed")
+                return@launch
+            }
+            if (!isCurrentGeneration(generation)) return@launch
+            val url = AuthManager.wsUrl()
+            val safeUrl = url.replace(Regex("token=[^&]+"), "token=REDACTED")
+            if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl")
+
+            val request = Request.Builder().url(url).build()
+            val newSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl(generation))
+            synchronized(outboundLock) {
+                if (connectionGeneration.get() == generation && !intentionalClose.get()) {
+                    webSocket = newSocket
+                } else {
+                    newSocket.cancel()
+                }
+            }
+        }
+    }
+
+    private fun isCurrentGeneration(generation: Int): Boolean =
+        connectionGeneration.get() == generation && !intentionalClose.get()
+
+    private fun scheduleReconnect() {
+        synchronized(outboundLock) {
+            if (intentionalClose.get() ||
+                _connectionStatus.value == ConnectionStatus.AUTH_EXPIRED ||
+                reconnectJob?.isActive == true
+            ) {
+                return
+            }
+            if (!AuthManager.isAutoReconnect()) {
+                if (BuildConfig.DEBUG) Log.d(TAG, "Auto-reconnect disabled")
+                _connectionStatus.value = ConnectionStatus.DISCONNECTED
+                return
+            }
+            if (!NetworkMonitor.isConnected.value) {
+                Log.d(TAG, "No network available — delaying reconnect scheduling")
+                _connectionStatus.value = ConnectionStatus.NO_NETWORK
+                return
+            }
+            val reconnectDelay = reconnectPolicy.nextDelayMs()
+            reconnectPolicy.advance()
+            if (BuildConfig.DEBUG) Log.d(TAG, "Reconnecting in ${reconnectDelay}ms …")
+
+            reconnectJob =
+                scope.launch {
+                    delay(reconnectDelay)
+                    val owner = currentCoroutineContext()[Job]
+                    val shouldOpen =
+                        synchronized(outboundLock) {
+                            if (reconnectJob !== owner) {
+                                false
+                            } else {
+                                reconnectJob = null
+                                !intentionalClose.get() &&
+                                    !connected.get() &&
+                                    _connectionStatus.value != ConnectionStatus.AUTH_EXPIRED
+                            }
+                        }
+                    if (shouldOpen) openSocket()
+                }
+        }
+    }
+
+    private fun isTerminalAuthClose(
+        code: Int,
+        reason: String,
+    ): Boolean =
+        code == 4401 || code == 4403 ||
+            reason.contains("unauthorized", ignoreCase = true) ||
+            reason.startsWith("auth:", ignoreCase = true)
+
+    private inner class WsListenerImpl(
+        private val generation: Int,
+    ) : WebSocketListener() {
+        private fun isCurrent(): Boolean = isCurrentGeneration(generation)
+
+        override fun onOpen(
+            webSocket: WebSocket,
+            response: Response,
+        ) {
+            synchronized(outboundLock) {
+                if (!isCurrent()) {
+                    webSocket.close(1000, "Superseded")
+                    return
+                }
+                this@WsTransport.webSocket = webSocket
+                closingSocket = null
+                outboundDrainJob?.cancel()
+                outboundDrainJob = null
+                Log.i(TAG, "WebSocket opened")
+                connected.set(true)
+                _connectionStatus.value = ConnectionStatus.CONNECTED
+                reconnectPolicy.reset()
+                connectionHealth.start()
+
+                while (true) {
+                    val msg = messageQueue.peek() ?: break
+                    if (!isRetryableMessage(msg)) {
+                        Log.w(TAG, "Dropping oversized queued message")
+                        messageQueue.poll()
+                        markQueuedMessageSent(msg)
+                        continue
+                    }
+                    if (BuildConfig.DEBUG) Log.d(TAG, formatSafeQueuedFrameLog(msg))
+                    if (!webSocket.send(msg)) {
+                        recoverRejectedSocket(webSocket)
+                        break
+                    }
+                    messageQueue.poll()
+                    markQueuedMessageSent(msg)
+                }
+                replayTracker.triggerReplay()
+            }
+            disconnectIfIdleInBackground()
+        }
+
+        override fun onMessage(
+            webSocket: WebSocket,
+            text: String,
+        ) {
+            if (!isCurrent() || this@WsTransport.webSocket !== webSocket) return
+            if (BuildConfig.DEBUG) Log.d(TAG, formatSafeIncomingFrameLog(text))
+            connectionHealth.onInboundFrame()
+            onRawMessage(text)
+        }
+
+        override fun onClosing(
+            webSocket: WebSocket,
+            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 (isTerminalAuthClose(code, reason)) {
+                    _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 (this@WsTransport.webSocket === webSocket) this@WsTransport.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")
+                connected.set(false)
+                ActiveSessionHolder.clear()
+                connectionHealth.stop()
+                rpcChannel.clearCapabilityRequests()
+                if (isTerminalAuthClose(code, reason)) {
+                    _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED
+                } else if (_connectionStatus.value != ConnectionStatus.AUTH_EXPIRED) {
+                    _connectionStatus.value = ConnectionStatus.RECONNECTING
+                    scheduleReconnect()
+                }
+            }
+        }
+
+        override fun onFailure(
+            webSocket: WebSocket,
+            t: Throwable,
+            response: Response?,
+        ) {
+            synchronized(outboundLock) {
+                if (!isCurrent()) return
+                connectionGeneration.incrementAndGet()
+                closingSocket = null
+                outboundDrainJob?.cancel()
+                outboundDrainJob = null
+                if (this@WsTransport.webSocket === webSocket) this@WsTransport.webSocket = null
+                // Log the exception class only — [Throwable.message] can leak URLs
+                // or headers. The message is still inspected internally for auth
+                // detection.
+                Log.e(TAG, "WebSocket failure: ${t.javaClass.simpleName}", t)
+                connected.set(false)
+                ActiveSessionHolder.clear()
+                connectionHealth.stop()
+                rpcChannel.clearCapabilityRequests()
+                val code = response?.code ?: 0
+                if (code == 401 || code == 4401 || code == 4403 ||
+                    t.message?.contains("401") == true ||
+                    t.message?.contains("4401") == true ||
+                    t.message?.contains("4403") == true ||
+                    t.message?.contains("unauthorized", ignoreCase = true) == true
+                ) {
+                    _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED
+                } else if (_connectionStatus.value != ConnectionStatus.AUTH_EXPIRED) {
+                    _connectionStatus.value = ConnectionStatus.RECONNECTING
+                    scheduleReconnect()
+                }
+            }
+        }
+    }
+
+    // ── Test seams (issue #1376 extraction support) ──────────────────────
+
+    @VisibleForTesting
+    internal val connectedForTest: AtomicBoolean
+        get() = connected
+
+    @VisibleForTesting
+    internal val intentionalCloseForTest: AtomicBoolean
+        get() = intentionalClose
+
+    @VisibleForTesting
+    internal val connectionGenerationForTest: AtomicInteger
+        get() = connectionGeneration
+
+    @VisibleForTesting
+    internal val messageQueueForTest: ConcurrentLinkedQueue<String>
+        get() = messageQueue
+
+    @VisibleForTesting
+    internal val outboundLockForTest: Any
+        get() = outboundLock
+
+    @VisibleForTesting
+    internal val acceptQueuedMessagesForTest: AtomicBoolean
+        get() = acceptQueuedMessages
+
+    @VisibleForTesting
+    internal val connectionStatusForTest: MutableStateFlow<ConnectionStatus>
+        get() = _connectionStatus
+
+    @VisibleForTesting
+    internal var webSocketForTest: WebSocket?
+        get() = webSocket
+        set(value) {
+            webSocket = value
+        }
+
+    @VisibleForTesting
+    internal var currentBackoffForTest: Long
+        get() = synchronized(outboundLock) { reconnectPolicy.currentBackoff }
+        set(value) {
+            synchronized(outboundLock) {
+                reconnectPolicy.currentBackoff = value
+            }
+        }
+
+    @VisibleForTesting
+    internal var reconnectJobForTest: Job?
+        get() = reconnectJob
+        set(value) {
+            reconnectJob = value
+        }
+
+    @VisibleForTesting
+    internal var outboundDrainJobForTest: Job?
+        get() = outboundDrainJob
+        set(value) {
+            outboundDrainJob = value
+        }
+
+    @VisibleForTesting
+    internal fun setReconnectBackoffForTest(initialMillis: Long) {
+        synchronized(outboundLock) {
+            reconnectPolicy.setBackoffForTest(initialMillis)
+        }
+    }
+
+    @VisibleForTesting
+    internal fun scheduleReconnectForTest() {
+        scheduleReconnect()
+    }
+
+    @VisibleForTesting
+    internal fun createListenerForTest(generation: Int): WebSocketListener = WsListenerImpl(generation)
+
+    @VisibleForTesting
+    internal fun setConnectionStatusForTest(status: ConnectionStatus) {
+        _connectionStatus.value = status
+    }
+
+    @VisibleForTesting
+    internal fun formatSafeOutgoingFrameLog(
+        id: String,
+        method: String,
+        paramsKeys: Collection<String>?,
+        byteLength: Int,
+        queued: Boolean = false,
+    ): String {
+        val prefix = if (queued) "→ (queued)" else "→"
+        val keys = paramsKeys?.joinToString(",", prefix = "[", postfix = "]") ?: "[]"
+        return "$prefix id=$id method=$method paramsKeys=$keys bytes=$byteLength"
+    }
+
+    @VisibleForTesting
+    internal fun formatSafeQueuedFrameLog(msg: String): String =
+        try {
+            val req = OkHttpProvider.json.decodeFromString<JsonRpcRequest>(msg)
+            formatSafeOutgoingFrameLog(req.id, req.method, req.params.keys, msg.length, queued = true)
+        } catch (_: Throwable) {
+            "→ (queued) frame bytes=${msg.length}"
+        }
+
+    @VisibleForTesting
+    internal fun formatSafeIncomingFrameLog(text: String): String =
+        try {
+            val element = OkHttpProvider.json.parseToJsonElement(text)
+            if (element is JsonObject) {
+                val id = (element["id"] as? JsonPrimitive)?.content
+                val method = (element["method"] as? JsonPrimitive)?.content
+                val hasResult = element.containsKey("result")
+                val resultKeys =
+                    (element["result"] as? JsonObject)?.keys?.joinToString(
+                        ",",
+                        prefix = "[",
+                        postfix = "]",
+                    )
+                val errorObj = element["error"] as? JsonObject
+                val errorCode = (errorObj?.get("code") as? JsonPrimitive)?.content
+                val sb = StringBuilder("← ")
+                if (id != null) sb.append("id=").append(id).append(" ")
+                if (method != null) sb.append("method=").append(method).append(" ")
+                if (hasResult) sb.append("result=").append(resultKeys ?: "present").append(" ")
+                if (errorCode != null) sb.append("errorCode=").append(errorCode).append(" ")
+                sb.append("bytes=").append(text.length).toString()
+            } else {
+                "← non-object frame bytes=${text.length}"
+            }
+        } catch (_: Throwable) {
+            "← frame bytes=${text.length}"
+        }
+}