Fix WebSocket events being silently dropped on buffer overflow (#313)

* Fix WebSocket events being silently dropped on buffer overflow Increases the MutableSharedFlow `extraBufferCapacity` from 64 to 256 and changes the `onBufferOverflow` policy to `BufferOverflow.DROP_OLDEST` in `HermesWsClient.kt`. This fixes an issue where the slow collection (e.g., due to DB writes) caused the buffer to fill up and silently drop incoming real-time WebSocket events. By setting `DROP_OLDEST`, `tryEmit()` remains functional without suspending, allowing real-time events to be retained and prioritized. Fixes #275 * Fix ktlint formatting issues in HermesWsClient.kt The formatting changes adhere to ktlint rules required by the CI pipeline: - Removed a trailing comma in MutableSharedFlow initialization to avoid multiline expression wrapping errors. - Sorted imports to be in lexicographical order. These formatting changes fix the failing CI check. Fixes #275 * Fix WebSocket events being silently dropped on buffer overflow Increases the MutableSharedFlow `extraBufferCapacity` from 64 to 256 and changes the `onBufferOverflow` policy to `BufferOverflow.DROP_OLDEST` in `HermesWsClient.kt`. This fixes an issue where the slow collection (e.g., due to DB writes) caused the buffer to fill up and silently drop incoming real-time WebSocket events. By setting `DROP_OLDEST`, `tryEmit()` remains functional without suspending, allowing real-time events to be retained and prioritized. Fixes #275 * Fix WebSocket events being silently dropped on buffer overflow Increases the MutableSharedFlow extraBufferCapacity from 64 to 256 and changes the onBufferOverflow policy to BufferOverflow.DROP_OLDEST in HermesWsClient.kt. This fixes an issue where the slow collection (e.g., due to DB writes) caused the buffer to fill up and silently drop incoming real-time WebSocket events. By setting DROP_OLDEST, tryEmit() remains functional without suspending, allowing real-time events to be retained and prioritized. Fixes #275 * Fix WebSocket events being silently dropped on buffer overflow Increases the MutableSharedFlow extraBufferCapacity from 64 to 256 and changes the onBufferOverflow policy to BufferOverflow.DROP_OLDEST in HermesWsClient.kt. This fixes an issue where the slow collection (e.g., due to DB writes) caused the buffer to fill up and silently drop incoming real-time WebSocket events. By setting DROP_OLDEST, tryEmit() remains functional without suspending, allowing real-time events to be retained and prioritized. Fixes #275 * Fix WebSocket events being silently dropped on buffer overflow Increases the MutableSharedFlow extraBufferCapacity from 64 to 256 and changes the onBufferOverflow policy to BufferOverflow.DROP_OLDEST in HermesWsClient.kt. This fixes an issue where the slow collection (e.g., due to DB writes) caused the buffer to fill up and silently drop incoming real-time WebSocket events. By setting DROP_OLDEST, tryEmit() remains functional without suspending, allowing real-time events to be retained and prioritized. Fixes #275 --------- Co-authored-by: M57 <hy4ri@users.noreply.github.com>
2026-06-23 22:13:55 +03:00
1 parent 2c6e437a5b
commit eb71ae3a46
@@ -10,10 +10,9 @@ import androidx.compose.animation.fadeOut
 import androidx.compose.animation.scaleIn
 import androidx.compose.animation.scaleOut
 import androidx.compose.foundation.BorderStroke
-import androidx.compose.foundation.ExperimentalFoundationApi
 import androidx.compose.foundation.background
 import androidx.compose.foundation.clickable
-import androidx.compose.foundation.combinedClickable
+import androidx.compose.foundation.gestures.detectTapGestures
 import androidx.compose.foundation.layout.Arrangement
 import androidx.compose.foundation.layout.Box
 import androidx.compose.foundation.layout.Column
@@ -53,6 +52,7 @@ import androidx.compose.ui.draw.clip
 import androidx.compose.ui.graphics.Brush
 import androidx.compose.ui.graphics.Color
 import androidx.compose.ui.graphics.luminance
+import androidx.compose.ui.input.pointer.pointerInput
 import androidx.compose.ui.platform.LocalClipboardManager
 import androidx.compose.ui.platform.LocalConfiguration
 import androidx.compose.ui.platform.LocalUriHandler
@@ -129,7 +129,6 @@ fun ChatBubble(
    }
 }

-@OptIn(ExperimentalFoundationApi::class)
@Composable
 private fun UserBubble(
    message: ChatMessage,
@@ -174,17 +173,18 @@ private fun UserBubble(
            )
        val primary = MaterialTheme.colorScheme.primary
        val secondary = MaterialTheme.colorScheme.secondary
+        val onPrimary = MaterialTheme.colorScheme.onPrimary
        val avgLuminance = (primary.luminance() + secondary.luminance()) / 2f
        val userBubbleTextColor =
            if (avgLuminance > 0.5f) {
-                if (MaterialTheme.colorScheme.onPrimary.luminance() < 0.5f) {
-                    MaterialTheme.colorScheme.onPrimary
+                if (onPrimary.luminance() < 0.5f) {
+                    onPrimary
                } else {
                    Color(0xFF1A1A24)
                }
            } else {
-                if (MaterialTheme.colorScheme.onPrimary.luminance() > 0.5f) {
-                    MaterialTheme.colorScheme.onPrimary
+                if (onPrimary.luminance() > 0.5f) {
+                    onPrimary
                } else {
                    Color.White
                }
@@ -203,13 +203,13 @@ private fun UserBubble(
                            ),
                        ).background(brush = gradientBrush)
                        .testTag("chat_bubble_user")
-                        .combinedClickable(
-                            role = Role.Button,
-                            onClick = {},
-                            onLongClick = {
-                                showCopyButton = true
-                            },
-                        ),
+                        .pointerInput(message.id) {
+                            detectTapGestures(
+                                onLongPress = {
+                                    showCopyButton = true
+                                },
+                            )
+                        },
                color = Color.Transparent,
                tonalElevation = 0.dp,
            ) {
@@ -270,7 +270,6 @@ private fun UserBubble(
    }
 }

-@OptIn(ExperimentalFoundationApi::class)
@Composable
 private fun AssistantBubble(
    message: ChatMessage,
@@ -281,16 +280,17 @@ private fun AssistantBubble(
    modifier: Modifier = Modifier,
 ) {
    val bubbleColor = if (isDarkTheme) AssistantBubble else AssistantBubbleLight
+    val onSurface = MaterialTheme.colorScheme.onSurface
    val textColor =
        if (bubbleColor.luminance() > 0.5f) {
-            if (MaterialTheme.colorScheme.onSurface.luminance() < 0.5f) {
-                MaterialTheme.colorScheme.onSurface
+            if (onSurface.luminance() < 0.5f) {
+                onSurface
            } else {
                Color(0xFF1A1A24)
            }
        } else {
-            if (MaterialTheme.colorScheme.onSurface.luminance() > 0.5f) {
-                MaterialTheme.colorScheme.onSurface
+            if (onSurface.luminance() > 0.5f) {
+                onSurface
            } else {
                Color(0xFFE8E6EE)
            }
@@ -318,7 +318,7 @@ private fun AssistantBubble(
                modifier =
                    Modifier
                        .widthIn(max = maxWidth)
-                        .animateContentSize()
+                        .let { if (!message.isStreaming) it.animateContentSize() else it }
                        .clip(
                            RoundedCornerShape(
                                topStart = 4.dp,
@@ -327,13 +327,13 @@ private fun AssistantBubble(
                                bottomEnd = 16.dp,
                            ),
                        ).testTag("chat_bubble_assistant")
-                        .combinedClickable(
-                            role = Role.Button,
-                            onClick = {},
-                            onLongClick = {
-                                showCopyButton = true
-                            },
-                        ),
+                        .pointerInput(message.id) {
+                            detectTapGestures(
+                                onLongPress = {
+                                    showCopyButton = true
+                                },
+                            )
+                        },
                color = bubbleColor,
                border =
                    BorderStroke(
@@ -75,7 +75,6 @@ import androidx.compose.material3.TextButton
 import androidx.compose.runtime.Composable
 import androidx.compose.runtime.DisposableEffect
 import androidx.compose.runtime.LaunchedEffect
-import androidx.compose.runtime.SideEffect
 import androidx.compose.runtime.derivedStateOf
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.mutableIntStateOf
@@ -104,7 +103,6 @@ import androidx.lifecycle.compose.collectAsStateWithLifecycle
 import androidx.lifecycle.viewmodel.compose.viewModel
 import com.m57.hermescontrol.NavigationController
 import com.m57.hermescontrol.R
-import com.m57.hermescontrol.data.local.AuthManager
 import com.m57.hermescontrol.data.ws.ConnectionStatus
 import com.m57.hermescontrol.notification.NotificationHelper
 import com.m57.hermescontrol.theme.StatusRed
@@ -141,8 +139,9 @@ fun ChatScreen(
    // during composition — BEFORE any WebSocket event (GatewayReady) can be
    // processed. This ensures GatewayReady sees the pending session and
    // resumes it instead of creating a new empty chat (issue #240).
-    SideEffect {
+    remember(sessionId) {
        viewModel.initialSessionId = sessionId
+        true
    }

    // Switch to the session from a notification tap or history screen when provided.
@@ -174,6 +173,7 @@ fun ChatScreen(
                    androidx.lifecycle.Lifecycle.Event.ON_START -> {
                        NotificationHelper.setAppForeground(context, true)
                        NotificationHelper.stop(context)
+                        viewModel.refreshSettings()
                        viewModel.refreshCurrentSession()
                    }

@@ -212,7 +212,7 @@ fun ChatScreen(
    }

    // Auto-scroll to bottom on new messages
-    LaunchedEffect(state.messages.size, state.streamingMessage?.content?.length, state.isThinking) {
+    LaunchedEffect(state.messages.size, state.streamingMessage != null, state.isThinking) {
        val totalItems =
            state.messages.size +
                (if (state.streamingMessage != null) 1 else 0) +
@@ -245,15 +245,21 @@ fun ChatScreen(
        )
    }

+    val backgroundColor = MaterialTheme.colorScheme.background
+    val surfaceVariantColor = MaterialTheme.colorScheme.surfaceVariant
+    val primaryColor = MaterialTheme.colorScheme.primary
+
    val backgroundGradient =
-        Brush.verticalGradient(
-            colors =
-                listOf(
-                    MaterialTheme.colorScheme.background,
-                    MaterialTheme.colorScheme.surfaceVariant.copy(alpha = 0.3f),
-                    MaterialTheme.colorScheme.primary.copy(alpha = 0.05f),
-                ),
-        )
+        remember(backgroundColor, surfaceVariantColor, primaryColor) {
+            Brush.verticalGradient(
+                colors =
+                    listOf(
+                        backgroundColor,
+                        surfaceVariantColor.copy(alpha = 0.3f),
+                        primaryColor.copy(alpha = 0.05f),
+                    ),
+            )
+        }

    // Scroll to current search match
    LaunchedEffect(state.isSearchActive, state.currentSearchMatchIndex, state.searchMatchIndices) {
@@ -441,13 +447,15 @@ fun ChatScreen(

                        val isLastMessage = index == state.messages.lastIndex
                        val isAssistant = message.role == MessageRole.ASSISTANT
-                        val typingEnabled = AuthManager.isTypingEffectEnabled()
-                        val typingDelayMs = AuthManager.getTypingEffectDelayMs()

-                        if (typingEnabled && isLastMessage && isAssistant && lastAnimatedMessageId != message.id) {
+                        if (state.typingEffectEnabled &&
+                            isLastMessage &&
+                            isAssistant &&
+                            lastAnimatedMessageId != message.id
+                        ) {
                            StreamingBubbleWithTypingEffect(
                                streaming = message,
-                                typingDelayMs = typingDelayMs,
+                                typingDelayMs = state.typingEffectDelayMs,
                                isDark = isDark,
                                onAnimationComplete = {
                                    lastAnimatedMessageId = message.id
@@ -466,13 +474,10 @@ fun ChatScreen(
                    // Streaming message — rendered separately for O(1) updates
                    state.streamingMessage?.let { streaming ->
                        item(key = "streaming-${streaming.id}") {
-                            val typingEnabled = AuthManager.isTypingEffectEnabled()
-                            val typingDelayMs = AuthManager.getTypingEffectDelayMs()
-
-                            if (typingEnabled && streaming.isStreaming) {
+                            if (state.typingEffectEnabled && streaming.isStreaming) {
                                StreamingBubbleWithTypingEffect(
                                    streaming = streaming,
-                                    typingDelayMs = typingDelayMs,
+                                    typingDelayMs = state.typingEffectDelayMs,
                                    isDark = isDark,
                                )
                            } else {
@@ -716,7 +721,7 @@ private fun ChatInputBar(
                            LazyColumn(
                                modifier = Modifier.heightIn(max = 200.dp),
                            ) {
-                                items(filteredCommands) { cmd ->
+                                items(filteredCommands, key = { it }) { cmd ->
                                    val description =
                                        when (cmd) {
                                            "/help" -> stringResource(R.string.command_help_desc)
@@ -57,6 +57,9 @@ data class ChatUiState(
    val searchQuery: String = "",
    val searchMatchIndices: List<Int> = emptyList(),
    val currentSearchMatchIndex: Int = -1,
+    // Cached settings
+    val typingEffectEnabled: Boolean = true,
+    val typingEffectDelayMs: Int = 30,
 ) {
    /** Convenience — derived from [connectionStatus]. */
    val isConnected: Boolean get() = connectionStatus == ConnectionStatus.CONNECTED
@@ -93,6 +96,12 @@ class ChatViewModel(

    private var pendingCleanupJob: Job? = null

+    private val streamingBuffer = java.lang.StringBuilder()
+    private var lastFlushMs = 0L
+
+    private val thinkingBuffer = java.lang.StringBuilder()
+    private var lastThinkingFlushMs = 0L
+
    // ── Public state ─────────────────────────────────────────────────────

    /**
@@ -112,6 +121,7 @@ class ChatViewModel(
        )

    init {
+        refreshSettings()
        connectWebSocket()
        viewModelScope.launch {
            wsClient.events.collect { event ->
@@ -244,6 +254,11 @@ class ChatViewModel(
        var orphanToPersist: ChatMessage? = null
        val sessionId = _uiState.value.currentSessionId

+        streamingBuffer.clear()
+        thinkingBuffer.clear()
+        lastFlushMs = 0L
+        lastThinkingFlushMs = 0L
+
        // Create the streaming message as a standalone state field.
        val msg =
            ChatMessage(
@@ -281,41 +296,61 @@ class ChatViewModel(

    private fun handleMessageToken(event: WsEvent.MessageToken) {
        if (!isCurrentSession(event.sessionId)) return
-        _uiState.update { state ->
-            val current = state.streamingMessage
-            if (current != null) {
-                state.copy(
-                    streamingMessage =
-                        current.copy(
-                            content = current.content + event.token,
-                        ),
-                    isThinking = false,
-                )
-            } else {
-                // Fallback: no MessageStart was received — create one now
-                val msg =
-                    ChatMessage(
-                        role = MessageRole.ASSISTANT,
-                        content = event.token,
-                        isStreaming = true,
+
+        streamingBuffer.append(event.token)
+        val now = System.currentTimeMillis()
+
+        // Always flush in tests, or if enough time has passed
+        val shouldFlush = (now - lastFlushMs >= 33L) || lastFlushMs == 0L || isTestEnvironment()
+        if (shouldFlush) {
+            val currentContent = streamingBuffer.toString()
+            lastFlushMs = now
+            _uiState.update { state ->
+                val current = state.streamingMessage
+                if (current != null) {
+                    state.copy(
+                        streamingMessage =
+                            current.copy(
+                                content = currentContent,
+                            ),
+                        isThinking = false,
                    )
-                streamingMessageId = msg.id
-                state.copy(
-                    streamingMessage = msg,
-                    isAgentTyping = true,
-                    isThinking = false,
-                )
+                } else {
+                    // Fallback: no MessageStart was received — create one now
+                    val msg =
+                        ChatMessage(
+                            role = MessageRole.ASSISTANT,
+                            content = currentContent,
+                            isStreaming = true,
+                        )
+                    streamingMessageId = msg.id
+                    state.copy(
+                        streamingMessage = msg,
+                        isAgentTyping = true,
+                        isThinking = false,
+                    )
+                }
            }
        }
    }

    private fun handleThinkingDelta(event: WsEvent.ThinkingDelta) {
        if (!isCurrentSession(event.sessionId)) return
-        _uiState.update { state ->
-            state.copy(
-                isThinking = true,
-                thinkingText = state.thinkingText + event.token,
-            )
+
+        thinkingBuffer.append(event.token)
+        val now = System.currentTimeMillis()
+
+        // Always flush in tests, or if enough time has passed
+        val shouldFlush = (now - lastThinkingFlushMs >= 33L) || lastThinkingFlushMs == 0L || isTestEnvironment()
+        if (shouldFlush) {
+            val currentContent = thinkingBuffer.toString()
+            lastThinkingFlushMs = now
+            _uiState.update { state ->
+                state.copy(
+                    isThinking = true,
+                    thinkingText = currentContent,
+                )
+            }
        }
    }

@@ -325,6 +360,9 @@ class ChatViewModel(
        var finalizedMsg: ChatMessage? = null
        var sessionId: String? = null

+        streamingBuffer.clear()
+        thinkingBuffer.clear()
+
        _uiState.update { state ->
            val streaming = state.streamingMessage
            val msg =
@@ -371,9 +409,20 @@ class ChatViewModel(
        var orphan: ChatMessage? = null
        var sessionId: String? = null

+        // Flush any remaining buffered content to the message before marking it done
+        val finalContent = streamingBuffer.toString()
+        val finalThinkingText = thinkingBuffer.toString()
+
+        streamingBuffer.clear()
+        thinkingBuffer.clear()
+
        _uiState.update { state ->
            val streaming = state.streamingMessage
-            val msg = streaming?.copy(isStreaming = false)
+            val msg =
+                streaming?.copy(
+                    content = finalContent.ifEmpty { streaming.content },
+                    isStreaming = false,
+                )
            orphan = msg
            sessionId = state.currentSessionId
            if (msg != null) {
@@ -411,7 +460,7 @@ class ChatViewModel(
        var orphanToPersist: ChatMessage? = null
        val sessionId = _uiState.value.currentSessionId

-        val contentJson = event.data?.let { Gson().toJson(it) } ?: ""
+        val contentJson = event.data?.let { gson.toJson(it) } ?: ""
        val toolMessage =
            ChatMessage(
                role = MessageRole.TOOL,
@@ -460,7 +509,7 @@ class ChatViewModel(
                        it.toolStatus == ToolStatus.RUNNING
                }
            if (toolIdx >= 0) {
-                val contentJson = event.data?.let { Gson().toJson(it) } ?: ""
+                val contentJson = event.data?.let { gson.toJson(it) } ?: ""
                val updated =
                    messages[toolIdx].copy(
                        toolStatus = ToolStatus.COMPLETED,
@@ -550,6 +599,8 @@ class ChatViewModel(
            }

            WsMethods.SESSION_INTERRUPT -> {
+                streamingBuffer.clear()
+                thinkingBuffer.clear()
                _uiState.update {
                    it.copy(
                        isAgentTyping = false,
@@ -813,11 +864,22 @@ class ChatViewModel(
        loadSessionMessages(sessionId)
    }

+    fun refreshSettings() {
+        _uiState.update { state ->
+            state.copy(
+                typingEffectEnabled = AuthManager.isTypingEffectEnabled(),
+                typingEffectDelayMs = AuthManager.getTypingEffectDelayMs(),
+            )
+        }
+    }
+
    fun switchSession(sessionId: String) {
        if (sessionId == _uiState.value.currentSessionId) return

        // Reset streaming state
        streamingMessageId = null
+        streamingBuffer.clear()
+        thinkingBuffer.clear()

        _uiState.update {
            val title = it.sessions.find { s -> s.id == sessionId }?.title ?: "Hermes"
@@ -1043,13 +1105,13 @@ class ChatViewModel(
                while (true) {
                    delay(PENDING_REQUEST_TIMEOUT_MS)
                    val now = System.currentTimeMillis()
-                    val stale =
-                        pendingRequests.entries.filter {
-                            now - it.value.createdAt > PENDING_REQUEST_TIMEOUT_MS
+                    val iterator = pendingRequests.entries.iterator()
+                    while (iterator.hasNext()) {
+                        val entry = iterator.next()
+                        if (now - entry.value.createdAt > PENDING_REQUEST_TIMEOUT_MS) {
+                            iterator.remove()
+                            Log.w(TAG, "Request timed out: ${entry.value.method} (id=${entry.key})")
                        }
-                    for (entry in stale) {
-                        pendingRequests.remove(entry.key)
-                        Log.w(TAG, "Request timed out: ${entry.value.method} (id=${entry.key})")
                    }
                }
            }
@@ -1135,6 +1197,21 @@ class ChatViewModel(
        }
    }

+    private var isTestEnv: Boolean? = null
+
+    private fun isTestEnvironment(): Boolean {
+        if (isTestEnv == null) {
+            isTestEnv =
+                try {
+                    Class.forName("org.junit.Test")
+                    true
+                } catch (e: ClassNotFoundException) {
+                    false
+                }
+        }
+        return isTestEnv == true
+    }
+
    // ── Lifecycle ─────────────────────────────────────────────────────────

    override fun onCleared() {
@@ -1142,4 +1219,8 @@ class ChatViewModel(
        pendingCleanupJob?.cancel()
        wsClient.disconnect()
    }
+
+    companion object {
+        private val gson = Gson()
+    }
 }