From 1cdc0dbfc042b70e9a355aeed69bacdd957e3796 Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Sat, 12 Sep 2026 00:19:13 +0300 Subject: [PATCH] fix(ws): gateway liveness, replay correctness, and streaming flush (#1085) --- .../hermescontrol/data/ws/HermesWsClient.kt | 192 ++++++++++++------ .../com/m57/hermescontrol/data/ws/WsEvent.kt | 7 + .../m57/hermescontrol/data/ws/WsMethods.kt | 3 + .../ui/chat/ChatStreamingController.kt | 157 ++++++++++---- .../hermescontrol/ui/chat/ChatViewModel.kt | 13 +- .../ui/chat/ChatWsEventReducer.kt | 4 + .../data/ws/HermesWsClientTest.kt | 95 ++++++++- .../ui/chat/ChatStreamingControllerTest.kt | 19 ++ 8 files changed, 372 insertions(+), 118 deletions(-) diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt index 80dde78e..cca84223 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt @@ -13,6 +13,7 @@ import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.currentCoroutineContext @@ -25,6 +26,7 @@ import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.filterIsInstance import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext import kotlinx.serialization.decodeFromString import kotlinx.serialization.encodeToString import kotlinx.serialization.json.JsonElement @@ -62,34 +64,48 @@ enum class ConnectionStatus { object HermesWsClient { private const val TAG = "HermesWsClient" - // ── Backoff settings ───────────────────────────────────────────────── + // ── Monotonic time helper (immune to NTP and wall-clock jumps) ───────── + private fun monotonicTimeMs(): Long = System.nanoTime() / 1_000_000L + + // ── Backoff settings (desktop parity: reconnect-backoff.ts full-jitter) ─ private var initialBackoffMs = 1_000L + private const val BACKOFF_BASE_MS = 300L + private const val BACKOFF_CAP_MS = 15_000L private const val MAX_BACKOFF_MS = 30_000L private const val BACKOFF_MULTIPLIER = 2.0 + private var testBackoffOverrideMs: Long? = null /** * Test hook: the initial reconnect backoff, overridable so reconnect * tests are deterministic instead of racing real time (CI runners - * routinely exceed fixed latch windows). Production keeps 1s. + * routinely exceed fixed latch windows). Production keeps full-jitter backoff. */ @VisibleForTesting internal fun setReconnectBackoffForTest(initialMillis: Long) { initialBackoffMs = initialMillis currentBackoff = initialMillis + testBackoffOverrideMs = initialMillis + } + + private fun nextBackoffDelayMs(): Long { + val override = testBackoffOverrideMs + if (override != null) return override + val ceiling = currentBackoff + if (ceiling <= 0L) return 0L + return (Math.random() * ceiling).toLong().coerceAtLeast(BACKOFF_BASE_MS).coerceAtMost(ceiling) } private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024 private const val OUTBOUND_DRAIN_TIMEOUT_MS = 60_000L /** - * No inbound frame for this long means the link is dead even though the - * TCP session may still look open (silent NAT/VPN drop). OkHttp pings - * every 30s (OkHttpProvider.websocket) and any inbound frame refreshes - * [lastPongTimestamp], so a healthy link never exceeds ~30s of silence. - * 90s tolerates exactly one missed ping cycle before acting. + * Liveness thresholds aligned with reference clients (desktop/web): + * Ping every 15s using gateway.ping; 45s inbound silence deadline. */ - private const val STALE_THRESHOLD_MS = 90_000L + private const val HEARTBEAT_INTERVAL_MS = 15_000L + private const val STALE_THRESHOLD_MS = 45_000L + private const val LIVENESS_PROBE_TIMEOUT_MS = 5_000L // ── Internal state (all access through synchronized / atomic) ──────── @@ -150,7 +166,7 @@ object HermesWsClient { private set val isHealthy: Boolean - get() = isConnected && (System.currentTimeMillis() - lastPongTimestamp < 60_000L) + get() = isConnected && (monotonicTimeMs() - lastPongTimestamp < STALE_THRESHOLD_MS) // ── Observable latency tracking (issue #1017) ───────────────────────── private val _lastLatencyMs = MutableStateFlow(null) @@ -162,14 +178,14 @@ object HermesWsClient { private fun startHealthTracking() { healthJob?.cancel() - lastPongTimestamp = System.currentTimeMillis() + lastPongTimestamp = monotonicTimeMs() healthJob = wsScope.launch { while (connected.get()) { - delay(30_000L) + delay(HEARTBEAT_INTERVAL_MS) if (connected.get()) { try { - ping(timeoutMs = 10_000L) + ping(timeoutMs = LIVENESS_PROBE_TIMEOUT_MS) } catch (e: Exception) { Log.w(TAG, "Health check ping failed: ${e.message}") } @@ -181,14 +197,15 @@ object HermesWsClient { /** * Ultra-lightweight JSON-RPC liveness check and RTT measurement (issue #1017). - * Bypasses agent queues on the backend; measures round-trip time in milliseconds. + * Uses [WsMethods.GATEWAY_PING] which is answered inline in the WebSocket read + * loop (ws.py:350) and never queues behind worker dispatch pools. * Updates [lastLatencyMs] and [lastPongTimestamp]. */ - suspend fun ping(timeoutMs: Long = 5_000L): Long { - val start = System.currentTimeMillis() - request(WsMethods.PING, emptyMap(), timeoutMs = timeoutMs).await() - val latency = (System.currentTimeMillis() - start).coerceAtLeast(0L) - lastPongTimestamp = System.currentTimeMillis() + suspend fun ping(timeoutMs: Long = LIVENESS_PROBE_TIMEOUT_MS): Long { + val start = monotonicTimeMs() + request(WsMethods.GATEWAY_PING, emptyMap(), timeoutMs = timeoutMs).await() + val latency = (monotonicTimeMs() - start).coerceAtLeast(0L) + lastPongTimestamp = monotonicTimeMs() _lastLatencyMs.value = latency return latency } @@ -207,7 +224,7 @@ object HermesWsClient { */ private fun runHealthCheckPass(): Boolean { if (!connected.get()) return false - val staleMs = System.currentTimeMillis() - lastPongTimestamp + val staleMs = monotonicTimeMs() - lastPongTimestamp if (staleMs <= STALE_THRESHOLD_MS) return true Log.w(TAG, "WebSocket stale (${staleMs / 1000}s without frames) — cancelling to trigger reconnect") synchronized(outboundLock) { @@ -220,11 +237,11 @@ object HermesWsClient { /** * Test hook: backdate liveness past the staleness threshold and force one * synchronous watchdog pass, so reconnect tests are deterministic instead - * of racing the real 30s/90s cadence. + * of racing the real 15s/45s cadence. */ @VisibleForTesting internal fun forceHealthCheckForTest(staleMillis: Long) { - lastPongTimestamp = System.currentTimeMillis() - staleMillis + lastPongTimestamp = monotonicTimeMs() - staleMillis runHealthCheckPass() } @@ -238,7 +255,7 @@ object HermesWsClient { private val parsedEvents = MutableSharedFlow( - extraBufferCapacity = 512, + extraBufferCapacity = 2048, onBufferOverflow = BufferOverflow.DROP_OLDEST, ) @@ -346,9 +363,34 @@ object HermesWsClient { @VisibleForTesting val isConnected: Boolean get() = connected.get() + private var foregroundProbeJob: Job? = null + fun setAppForeground(foreground: Boolean) { appInForeground.set(foreground) - if (!foreground) disconnectIfIdleInBackground() + if (!foreground) { + disconnectIfIdleInBackground() + return + } + probeLivenessOnWake() + } + + private fun probeLivenessOnWake(deferredOnce: Boolean = false) { + if (!connected.get()) return + foregroundProbeJob?.cancel() + foregroundProbeJob = + wsScope.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.w(TAG, "Foreground liveness probe failed — cancelling socket") + synchronized(outboundLock) { + if (connected.get()) webSocket?.cancel() + } + } } fun acquireExternalActivityConnectionLease() { @@ -629,6 +671,8 @@ object HermesWsClient { reconnectJob = null outboundDrainJob?.cancel() outboundDrainJob = null + foregroundProbeJob?.cancel() + foregroundProbeJob = null stopHealthTracking() webSocket?.close(1000, "Client closed") webSocket = null @@ -890,21 +934,23 @@ object HermesWsClient { // ── Sequence tracking & replay on reconnect (issue #1016) ───────────── private fun triggerReplay() { + // Arm the hold SYNCHRONOUSLY (still under onOpen's outboundLock) so a + // live frame racing the replay coroutine is parked instead of advancing + // lastSeenSeq past events the replay would then skip-drop (G2). if (lastSeenSeq.isEmpty() || replayInFlight) return - replayJob?.cancel() - replayJob = - wsScope.launch { - fetchReplay() - } - } - - private suspend fun fetchReplay() { - if (replayInFlight || lastSeenSeq.isEmpty()) return replayInFlight = true val sessionsToReplay = lastSeenSeq.keys().toList() for (sid in sessionsToReplay) { replayHold[sid] = Collections.synchronizedList(mutableListOf()) } + replayJob?.cancel() + replayJob = + wsScope.launch { + fetchReplay(sessionsToReplay) + } + } + + private suspend fun fetchReplay(sessionsToReplay: List = lastSeenSeq.keys().toList()) { try { for (sid in sessionsToReplay) { val lastSeen = lastSeenSeq[sid] ?: continue @@ -926,13 +972,21 @@ object HermesWsClient { } if (resultMap != null) { val epoch = resultMap["epoch"] as? String - if (!epoch.isNullOrEmpty()) { - if (replayEpoch != null && replayEpoch != epoch) { - replayEpoch = epoch - lastSeenSeq.clear() - continue - } - replayEpoch = epoch + val epochChanged = + !epoch.isNullOrEmpty() && replayEpoch != null && replayEpoch != epoch + val truncated = resultMap["truncated"] == true + val latestSeq = (resultMap["latest_seq"] as? Number)?.toInt() + + 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 + parsedEvents.tryEmit(WsEvent.TranscriptResyncRequired(sid)) + continue } val eventsList = (resultMap["events"] as? List<*>)?.filterIsInstance>() if (eventsList != null) { @@ -962,25 +1016,28 @@ object HermesWsClient { } } catch (e: Exception) { Log.w(TAG, "Replay failed for session $sid: ${e.message}") + parsedEvents.tryEmit(WsEvent.TranscriptResyncRequired(sid)) } } } finally { - 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 + 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 + } + parsedEvents.tryEmit(heldEvent) } - parsedEvents.tryEmit(heldEvent) } } } + replayInFlight = false } - replayInFlight = false } } @@ -1065,7 +1122,7 @@ object HermesWsClient { _connectionStatus.value = ConnectionStatus.NO_NETWORK return } - val reconnectDelay = currentBackoff + val reconnectDelay = nextBackoffDelayMs() currentBackoff = (currentBackoff * BACKOFF_MULTIPLIER) .toLong() @@ -1092,6 +1149,14 @@ object HermesWsClient { } } + private fun isTerminalAuthClose( + code: Int, + reason: String, + ): Boolean = + code == 4401 || code == 4403 || + reason.contains("unauthorized", ignoreCase = true) || + reason.startsWith("auth:", ignoreCase = true) + // ── Listener ───────────────────────────────────────────────────────── private class WsListenerImpl( @@ -1145,7 +1210,7 @@ object HermesWsClient { ) { if (!isCurrent() || HermesWsClient.webSocket !== webSocket) return if (BuildConfig.DEBUG) Log.d(TAG, "← $text") - lastPongTimestamp = System.currentTimeMillis() + lastPongTimestamp = monotonicTimeMs() // Resolve any in-flight `request()` awaiting this RPC result/error // (issue #526) before fanning the parsed event out to collectors. val event = @@ -1173,6 +1238,7 @@ object HermesWsClient { if (!sid.isNullOrBlank() && seq != null) { val prev = lastSeenSeq[sid] ?: 0 if (seq <= prev) { + // Frame is already at or behind watermark — drop duplicate return } val parsed = EventParser.parse(rpc, text) @@ -1219,17 +1285,17 @@ object HermesWsClient { pendingPromptSubmits.isEmpty() ) { pendingReply = false + disconnectIfIdleInBackground() } } removeQueuedMessage(event.id) resolvePending(event.id, null, event.error) - disconnectIfIdleInBackground() } else -> {} } // tryEmit on a DROP_OLDEST flow only returns false when the - // buffer is full AND no subscriber is draining; with extraBuffer=512 + // buffer is full AND no subscriber is draining; with extraBuffer=2048 // and always-on init collectors this is unreachable in practice. parsedEvents.tryEmit(event) } @@ -1247,10 +1313,7 @@ object HermesWsClient { 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) - ) { + if (isTerminalAuthClose(code, reason)) { _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED } webSocket.close(code, reason) @@ -1275,10 +1338,7 @@ object HermesWsClient { connected.set(false) ActiveSessionHolder.clear() stopHealthTracking() - if (code == 4001 || code == 4401 || - reason.contains("unauthorized", ignoreCase = true) || - reason.startsWith("auth:", ignoreCase = true) - ) { + if (isTerminalAuthClose(code, reason)) { _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED } else if (_connectionStatus.value != ConnectionStatus.AUTH_EXPIRED) { _connectionStatus.value = ConnectionStatus.RECONNECTING @@ -1307,9 +1367,11 @@ object HermesWsClient { ActiveSessionHolder.clear() stopHealthTracking() val code = response?.code ?: 0 - if (code == 401 || t.message?.contains( - "401", - ) == true || t.message?.contains("unauthorized", ignoreCase = true) == true + 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) { diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt index 7f1cff0f..21d61b6a 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt @@ -355,6 +355,13 @@ sealed class WsEvent { val kind: String = "", ) : WsEvent() + // ── Replay resync (internal) ────────────────────────────────────────── + + /** Internal: replay could not cover the reconnect gap (truncated or epoch change) — UI must refetch history. */ + data class TranscriptResyncRequired( + val sessionId: String, + ) : WsEvent() + // ── Fallback ───────────────────────────────────────────────────────── data class Unknown( diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt index c8e00fb4..e2d8a1da 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt @@ -48,6 +48,9 @@ object WsMethods { /** Ultra-lightweight liveness check and round-trip latency measurement. */ const val PING = "ping" + /** Inline WebSocket read-loop ping, bypasses server thread pool dispatch. */ + const val GATEWAY_PING = "gateway.ping" + // ── Commands catalog ────────────────────────────────────────────────── const val COMMANDS_CATALOG = "commands.catalog" const val COMMAND_DISPATCH = "command.dispatch" diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt index 0d06ccc5..9652cf5f 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt @@ -2,8 +2,11 @@ package com.m57.hermescontrol.ui.chat import com.m57.hermescontrol.data.ws.WsEvent import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.update +import kotlinx.coroutines.launch /** * Holds chat message-streaming state and logic, extracted from [ChatViewModel] @@ -27,14 +30,23 @@ class ChatStreamingController( private val isCurrentSession: (String?) -> Boolean, private val isTestEnvironment: () -> Boolean, ) { + private companion object { + const val FLUSH_INTERVAL_MS = 33L + } + private val streamingBuffer = java.lang.StringBuilder() private var lastFlushMs = 0L + private var trailingMessageFlushJob: Job? = null private val thinkingBuffer = java.lang.StringBuilder() private var lastThinkingFlushMs = 0L + private var trailingThinkingFlushJob: Job? = null private val reasoningBuffer = java.lang.StringBuilder() private var lastReasoningFlushMs = 0L + private var trailingReasoningFlushJob: Job? = null + + private fun nowMs(): Long = System.nanoTime() / 1_000_000L /** * Resets all streaming buffers. Centralizes the clear logic that was @@ -42,6 +54,9 @@ class ChatStreamingController( * ToolStart, ClarifyRequest, session switches, and interrupt handling. */ fun resetStreaming() { + trailingMessageFlushJob?.cancel() + trailingThinkingFlushJob?.cancel() + trailingReasoningFlushJob?.cancel() flushReasoning() streamingBuffer.clear() thinkingBuffer.clear() @@ -73,6 +88,9 @@ class ChatStreamingController( * buffer-only clear is safe and prevents stale deltas from re-flushing. */ fun clearStreamingBuffers() { + trailingMessageFlushJob?.cancel() + trailingThinkingFlushJob?.cancel() + trailingReasoningFlushJob?.cancel() streamingBuffer.clear() thinkingBuffer.clear() reasoningBuffer.clear() @@ -94,6 +112,7 @@ class ChatStreamingController( * synchronously so the finalized message carries the latest reasoning. */ fun flushPendingReasoning() { + trailingReasoningFlushJob?.cancel() flushReasoning() } @@ -111,8 +130,10 @@ class ChatStreamingController( * tokens synchronously first makes the sealed orphan complete. */ fun flushPendingTokens() { + trailingMessageFlushJob?.cancel() val currentContent = streamingBuffer.toString() if (currentContent.isEmpty()) return + lastFlushMs = nowMs() streamingState.update { state -> val current = state.streamingMessage ?: return@update state state.copy( @@ -130,63 +151,97 @@ class ChatStreamingController( if (!isCurrentSession(event.sessionId)) return streamingBuffer.append(event.token) - val now = System.currentTimeMillis() + val now = nowMs() // Always flush in tests, or if enough time has passed - val shouldFlush = (now - lastFlushMs >= 33L) || lastFlushMs == 0L || isTestEnvironment() + val shouldFlush = (now - lastFlushMs >= FLUSH_INTERVAL_MS) || lastFlushMs == 0L || isTestEnvironment() if (shouldFlush) { - val currentContent = streamingBuffer.toString() - lastFlushMs = now - streamingState.update { state -> - val current = state.streamingMessage - if (current != null) { - val currentReasoning = - if (current.reasoningText.isNotBlank()) current.reasoningText else state.reasoningText - state.copy( - streamingMessage = - current.copy( - content = currentContent, - reasoningText = currentReasoning, - ), - isThinking = false, - ) - } else { - // Fallback: no MessageStart was received — create one now - val msg = - ChatMessage( - role = MessageRole.ASSISTANT, - content = currentContent, - reasoningText = state.reasoningText, - isStreaming = true, - ) - state.copy( - streamingMessage = msg, - isThinking = false, - ) - } - } - uiState.update { it.copy(isAgentTyping = true) } + trailingMessageFlushJob?.cancel() + flushTokens(now) + } else { + scheduleTrailingMessageFlush() } } + private fun scheduleTrailingMessageFlush() { + if (trailingMessageFlushJob?.isActive == true) return + trailingMessageFlushJob = + scope.launch { + delay(FLUSH_INTERVAL_MS) + flushTokens(nowMs()) + } + } + + private fun flushTokens(now: Long) { + val currentContent = streamingBuffer.toString() + if (currentContent.isEmpty()) return + lastFlushMs = now + streamingState.update { state -> + val current = state.streamingMessage + if (current != null) { + val currentReasoning = + if (current.reasoningText.isNotBlank()) current.reasoningText else state.reasoningText + state.copy( + streamingMessage = + current.copy( + content = currentContent, + reasoningText = currentReasoning, + ), + isThinking = false, + ) + } else { + // Fallback: no MessageStart was received — create one now + val msg = + ChatMessage( + role = MessageRole.ASSISTANT, + content = currentContent, + reasoningText = state.reasoningText, + isStreaming = true, + ) + state.copy( + streamingMessage = msg, + isThinking = false, + ) + } + } + uiState.update { it.copy(isAgentTyping = true) } + } + fun handleThinkingDelta(event: WsEvent.ThinkingDelta) { if (!isCurrentSession(event.sessionId)) return thinkingBuffer.append(event.token) - val now = System.currentTimeMillis() + val now = nowMs() // Always flush in tests, or if enough time has passed val shouldFlush = - (now - lastThinkingFlushMs >= 33L) || lastThinkingFlushMs == 0L || isTestEnvironment() + (now - lastThinkingFlushMs >= FLUSH_INTERVAL_MS) || lastThinkingFlushMs == 0L || isTestEnvironment() if (shouldFlush) { - val currentContent = thinkingBuffer.toString() - lastThinkingFlushMs = now - streamingState.update { state -> - state.copy( - isThinking = true, - thinkingText = currentContent, - ) + trailingThinkingFlushJob?.cancel() + flushThinking(now) + } else { + scheduleTrailingThinkingFlush() + } + } + + private fun scheduleTrailingThinkingFlush() { + if (trailingThinkingFlushJob?.isActive == true) return + trailingThinkingFlushJob = + scope.launch { + delay(FLUSH_INTERVAL_MS) + flushThinking(nowMs()) } + } + + private fun flushThinking(now: Long) { + val currentContent = thinkingBuffer.toString() + if (currentContent.isEmpty()) return + lastThinkingFlushMs = now + streamingState.update { state -> + state.copy( + isThinking = true, + thinkingText = currentContent, + ) } } @@ -194,17 +249,29 @@ class ChatStreamingController( if (!isCurrentSession(event.sessionId)) return reasoningBuffer.append(event.token) - val now = System.currentTimeMillis() + val now = nowMs() // Always flush in tests, or if enough time has passed val shouldFlush = - (now - lastReasoningFlushMs >= 33L) || lastReasoningFlushMs == 0L || isTestEnvironment() + (now - lastReasoningFlushMs >= FLUSH_INTERVAL_MS) || lastReasoningFlushMs == 0L || isTestEnvironment() if (shouldFlush) { + trailingReasoningFlushJob?.cancel() flushReasoning(now) + } else { + scheduleTrailingReasoningFlush() } } - private fun flushReasoning(now: Long = System.currentTimeMillis()) { + private fun scheduleTrailingReasoningFlush() { + if (trailingReasoningFlushJob?.isActive == true) return + trailingReasoningFlushJob = + scope.launch { + delay(FLUSH_INTERVAL_MS) + flushReasoning(nowMs()) + } + } + + private fun flushReasoning(now: Long = nowMs()) { val currentContent = reasoningBuffer.toString() if (currentContent.isEmpty()) return lastReasoningFlushMs = now diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt index 7282b615..3e24796e 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt @@ -12,7 +12,6 @@ import com.m57.hermescontrol.data.model.Attachment import com.m57.hermescontrol.data.model.ModelCapabilities import com.m57.hermescontrol.data.model.ModelProvider import com.m57.hermescontrol.data.model.PinnedModel -import com.m57.hermescontrol.data.model.ProfileInfo import com.m57.hermescontrol.data.model.parseContextBreakdown import com.m57.hermescontrol.data.model.parseUsageSnapshot import com.m57.hermescontrol.data.remote.ApiClient @@ -45,10 +44,6 @@ import kotlinx.serialization.json.JsonObject import kotlinx.serialization.json.decodeFromJsonElement import kotlinx.serialization.json.intOrNull import kotlinx.serialization.json.jsonPrimitive -import okhttp3.MediaType.Companion.toMediaType -import okhttp3.Request -import okhttp3.RequestBody.Companion.toRequestBody -import org.json.JSONObject import java.io.File import java.util.concurrent.ConcurrentHashMap @@ -838,6 +833,14 @@ class ChatViewModel( loadSessions() } + is WsEvent.TranscriptResyncRequired -> { + val current = runtimeSessionId ?: _uiState.value.currentSessionId + if (current != null && (current == event.sessionId || event.sessionId.isEmpty())) { + val storageId = ActiveSessionHolder.resolveStoredSessionId(current) ?: current + loadSessionMessages(storageId, sessionGeneration) + } + } + is WsEvent.ClarifyRequest -> { _uiState.update { it.copy( diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt index d4f84abe..d8db9c15 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt @@ -74,6 +74,7 @@ object ChatWsEventReducer { is WsEvent.ReviewSummary -> event.sessionId is WsEvent.BtwComplete -> event.sessionId is WsEvent.SessionUsage -> event.sessionId + is WsEvent.TranscriptResyncRequired -> event.sessionId else -> null } if (eventSessionId != null && currentSessionId != null && eventSessionId != currentSessionId) { @@ -174,6 +175,9 @@ object ChatWsEventReducer { // Change events (issue #784) are consumed by their screens' ViewModels is WsEvent.ChangeEvent -> ReducerResult(state = state, streamingState = streamingState) + + // TranscriptResyncRequired is handled by ChatViewModel directly (refetch history on replay gap) + is WsEvent.TranscriptResyncRequired -> ReducerResult(state = state, streamingState = streamingState) } // ── GatewayReady ────────────────────────────────────────────────── diff --git a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt index 7994b464..bfd30a21 100644 --- a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt @@ -15,6 +15,8 @@ import io.mockk.mockkObject import io.mockk.mockkStatic import io.mockk.unmockkAll import io.mockk.verify +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first @@ -35,6 +37,7 @@ import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test import java.io.IOException +import java.util.Collections import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger @@ -1576,7 +1579,8 @@ class HermesWsClientTest { // Send duplicate seq 1 -> should be dropped ws.send( - """{"method":"event","params":{"type":"message.token","session_id":"s1","seq":1,"payload":{"text":"A-dup"}}}""", + """{"method":"event","params":{"type":"message.token","session_id":"s1","seq":1,""" + + """"payload":{"text":"A-dup"}}}""", ) Thread.sleep(100) assertEquals(1, HermesWsClient.getSeqWatermarks()["s1"]) @@ -1664,7 +1668,9 @@ class HermesWsClientTest { // Extract ID from JSON-RPC request val id = Regex(""""id":"([^"]+)"""").find(text)?.groupValues?.get(1) ?: "1" webSocket.send( - """{"jsonrpc":"2.0","id":"$id","result":{"epoch":"ep1","events":[{"type":"message.token","session_id":"s1","seq":6,"payload":{"text":"replayed"}}]}}""", + """{"jsonrpc":"2.0","id":"$id","result":{"epoch":"ep1","events":""" + + """[{"type":"message.token","session_id":"s1","seq":6,""" + + """"payload":{"text":"replayed"}}]}}""", ) requestLatch.countDown() } @@ -1689,9 +1695,92 @@ class HermesWsClientTest { assertEquals(6, HermesWsClient.getSeqWatermarks()["s1"]) } + @Test + fun testTerminalCloseCode4403SetsAuthExpired() { + val socket = mockk(relaxed = 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, 4403, "forbidden origin") + assertEquals(ConnectionStatus.AUTH_EXPIRED, HermesWsClient.connectionStatus.value) + } + + @Test + fun testReplayTruncatedEmitsTranscriptResyncRequired() { + var serverWebSocket: WebSocket? = null + val serverLatch = CountDownLatch(1) + val requestLatch = CountDownLatch(1) + var receivedMethod: String? = null + val receivedEvents = Collections.synchronizedList(mutableListOf()) + + val collectJob = + CoroutineScope(Dispatchers.IO).launch { + HermesWsClient.events.collect { receivedEvents.add(it) } + } + + mockWebServer.enqueue( + MockResponse().withWebSocketUpgrade( + object : WebSocketListener() { + override fun onOpen( + webSocket: WebSocket, + response: okhttp3.Response, + ) { + serverWebSocket = webSocket + serverLatch.countDown() + } + + override fun onMessage( + webSocket: WebSocket, + text: String, + ) { + if (text.contains(WsMethods.SESSION_EVENTS_SINCE)) { + receivedMethod = WsMethods.SESSION_EVENTS_SINCE + val id = Regex(""""id":"([^"]+)"""").find(text)?.groupValues?.get(1) ?: "1" + webSocket.send( + """{"jsonrpc":"2.0","id":"$id","result":""" + + """{"epoch":"ep1","truncated":true,"latest_seq":100,"events":[]}}""", + ) + requestLatch.countDown() + } + } + }, + ), + ) + + HermesWsClient.setSeqWatermark("s1", 5) + HermesWsClient.connect() + runBlocking { + withTimeout(5000) { + HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } + } + } + assertTrue(serverLatch.await(5, TimeUnit.SECONDS)) + assertTrue(requestLatch.await(5, TimeUnit.SECONDS)) + assertEquals(WsMethods.SESSION_EVENTS_SINCE, receivedMethod) + + // Wait for replay processing + Thread.sleep(300) + assertEquals(100, HermesWsClient.getSeqWatermarks()["s1"]) + val resyncEvent = receivedEvents.filterIsInstance().firstOrNull() + assertNotNull(resyncEvent) + assertEquals("s1", resyncEvent?.sessionId) + collectJob.cancel() + } + @Test fun testPingMethodConstant() { assertEquals("ping", WsMethods.PING) + assertEquals("gateway.ping", WsMethods.GATEWAY_PING) } @Test @@ -1716,7 +1805,7 @@ class HermesWsClientTest { webSocket: WebSocket, text: String, ) { - if (text.contains(""""method":"ping"""")) { + if (text.contains(""""method":"ping"""") || text.contains(""""method":"gateway.ping"""")) { pingReceived = true val id = Regex(""""id":"([^"]+)"""").find(text)?.groupValues?.get(1) ?: "1" webSocket.send( diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatStreamingControllerTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatStreamingControllerTest.kt index 3bc70e35..6efcd0a2 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatStreamingControllerTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatStreamingControllerTest.kt @@ -149,6 +149,25 @@ class ChatStreamingControllerTest { assertEquals("real reasoning trace", streamingState.value.streamingMessage?.reasoningText) } + @Test + fun handleMessageToken_schedulesTrailingFlushForBufferedTail() = + runTest { + val uiState = MutableStateFlow(ChatUiState()) + val streamingState = MutableStateFlow(StreamingState()) + val controller = controller(this, uiState, streamingState, isTestEnvironment = { false }) + + controller.handleMessageToken(WsEvent.MessageToken("First", "session")) + assertEquals("First", streamingState.value.streamingMessage?.content) + + // Second token arrives immediately (<33ms) -> held in buffer + controller.handleMessageToken(WsEvent.MessageToken(" Second", "session")) + assertEquals("First", streamingState.value.streamingMessage?.content) + + // Advance virtual time by 35ms -> trailing flush executes + testScheduler.advanceTimeBy(35L) + assertEquals("First Second", streamingState.value.streamingMessage?.content) + } + private fun controller( scope: CoroutineScope, uiState: MutableStateFlow,