From be0bbcb2c9d574ae88f46aa6af6c3cbeebb1ed71 Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:03:53 +0300 Subject: [PATCH] fix(chat): keep session hydration consistent across retries (#804 rebase) (#813) * fix(chat): keep session state consistent across resume retries * fix(chat): preserve hydration retry after resume ack * fix(chat): publish branched runtime session id --------- Co-authored-by: AlexDev_ <56083016+alexdev03@users.noreply.github.com> --- .../hermescontrol/ui/chat/ChatViewModel.kt | 444 ++++++++++++------ .../ui/chat/ChatToolDedupeTest.kt | 31 ++ .../ui/chat/ChatViewModelTest.kt | 407 ++++++++++++++++ 3 files changed, 740 insertions(+), 142 deletions(-) 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 66e32ecb..091c0e1f 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 @@ -154,15 +154,47 @@ internal fun mergeTranscriptWithLive( restMessages: List, currentMessages: List, ): List { + // User rows can carry live/cache-only metadata (for example whether the + // bubble continues the active turn). Keep that richer copy when REST + // returns the same logical row. + val currentUsers = currentMessages.filter { it.role == MessageRole.USER }.toMutableList() + val mergedRest = + restMessages.map { rest -> + if (rest.role != MessageRole.USER) return@map rest + val matchIndex = + currentUsers.indexOfFirst { it.id == rest.id }.takeIf { it >= 0 } + ?: currentUsers.indexOfFirst { sameLogicalMessage(rest, it) } + if (matchIndex >= 0) currentUsers.removeAt(matchIndex) else rest + } val restIds = restMessages.map { it.id }.toSet() val liveTail = currentMessages.filter { old -> old.id !in restIds && restMessages.none { sameLogicalMessage(it, old) } } - if (liveTail.isEmpty()) return restMessages - return (restMessages + liveTail).sortedBy { it.timestamp } + if (liveTail.isEmpty()) return mergedRest + return (mergedRest + liveTail).sortedBy { it.timestamp } } +/** Merge a REST page without matching it against already-settled transcript rows. */ +internal fun mergeIncrementalTranscriptPage( + restMessages: List, + currentMessages: List, + sessionId: String, + offset: Int, +): List { + val settledEnd = + currentMessages.indexOfLast { message -> + serverMessageIndex(message.id, sessionId)?.let { it < offset } == true + } + return currentMessages.take(settledEnd + 1) + + mergeTranscriptWithLive(restMessages, currentMessages.drop(settledEnd + 1)) +} + +private fun serverMessageIndex( + id: String, + sessionId: String, +): Int? = id.removePrefix("rest-$sessionId-").takeIf { it != id }?.toIntOrNull() + data class ChatUiState( val messages: List = emptyList(), val currentSessionId: String? = null, @@ -318,6 +350,21 @@ class ChatViewModel( /** Maps an in-flight RPC id to its method for UI error labeling. */ private val idToMethod = ConcurrentHashMap() + private data class SessionRequest( + val generation: Long, + val resumeSequence: Long = 0L, + val sessionId: String? = null, + ) + + private val sessionRequestById = ConcurrentHashMap() + private var sessionGeneration = 0L + private var resumeRequestSequence = 0L + private var activeResumeRequestSequence = 0L + private var hydrationRequestSequence = 0L + private var activeHydrationRequestSequence = 0L + private var resumedGeneration = -1L + private var hydratedGeneration = -1L + /** Runtime TUI session returned by session.resume; Desktop storage keeps the original ID. */ private var runtimeSessionId: String? = null @@ -520,14 +567,7 @@ class ChatViewModel( val currentId = _uiState.value.currentSessionId if (currentId != null) { if (sessionHasServerPresence) { - viewModelScope.launch(Dispatchers.IO) { - wsClient.send( - WsMethods.SESSION_RESUME, - mapOf("session_id" to currentId), - onSent = { id -> trackRequest(id, WsMethods.SESSION_RESUME) }, - ) - } - loadSessionMessages(currentId) + resumeSession(currentId, sessionGeneration) } // No server-side row yet (session created but never prompted): // resuming would 4007 and the REST transcript would 404. Keep the @@ -545,6 +585,14 @@ class ChatViewModel( } private fun handleWsEvent(event: WsEvent) { + // RpcError is reduced before ViewModel request handling. Drop stale + // session errors here so the shared reducer cannot clear loading or + // surface an error for a newly selected session. + if (event is WsEvent.RpcError && isStaleSessionRequest(event.id)) { + forgetRequest(event.id) + return + } + // Flush any throttled reasoning before a state transition so the // finalized/orphan message carries the latest reasoning text. when (event) { @@ -764,6 +812,8 @@ class ChatViewModel( result: Any?, ) { val method = idToMethod.remove(id) ?: return + val request = sessionRequestById.remove(id) + if (request != null && isStaleSessionRequest(request)) return when (method) { WsMethods.SESSION_CREATE -> { val resultMap = result as? Map ?: return @@ -811,25 +861,18 @@ class ChatViewModel( // misses) and the REST transcript 404. val runtimeId = resultMap["session_id"] as? String ?: return val storageId = resultMap["stored_session_id"] as? String ?: runtimeId + val generation = + resetSessionState( + sessionId = storageId, + title = (resultMap["title"] as? String)?.takeIf { it.isNotBlank() } ?: "Hermes", + isLoading = false, + ) runtimeSessionId = runtimeId + ActiveSessionHolder.set(runtimeId) sessionHasServerPresence = false sessionGoneRecoveryInFlight = false - _uiState.update { - it.copy( - currentSessionId = storageId, - isLoading = false, - messages = emptyList(), - chatTitle = (resultMap["title"] as? String)?.takeIf { t -> t.isNotBlank() } ?: "Hermes", - usedContextTokens = null, - fullContextTokens = null, - contextBreakdown = null, - compressionCount = null, - ) - } - ActiveSessionHolder.set(runtimeId) - _streamingState.update { StreamingState() } addSystemMessage("Session branched", persist = true) - loadSessionMessages(storageId) + loadSessionMessages(storageId, generation) loadSessions() fetchContextUsage() } @@ -860,7 +903,8 @@ class ChatViewModel( // Resume succeeded — the gateway confirmed the DB row. sessionHasServerPresence = true val sessionId = - (resultMap?.get("resumed") as? String) + request?.sessionId + ?: (resultMap?.get("resumed") as? String) ?: _uiState.value.currentSessionId // Parse session info from backend — model, provider, reasoning_effort @@ -874,16 +918,10 @@ class ChatViewModel( // the WS round-trip. Calling loadCachedMessages() here would // overwrite any message the user sent between switchSession() and // the server ack, making the chat appear to go blank. - // A successful resume also retires any pending retry job and - // clears the retry indicators (a reconnect re-resume landing - // while a backoff timer was armed must not double-fire). - resumeRetryJob?.cancel() - resumeRetryJob = null _uiState.update { it.copy( isLoading = false, - isResumeRetrying = false, - resumeError = null, + errorMessage = null, currentSessionId = sessionId, currentSessionModel = if (model != null && provider != null) { @@ -903,6 +941,9 @@ class ChatViewModel( ActiveSessionHolder.set(runtimeSessionId ?: sessionId) addSystemMessage("Session resumed") fetchContextUsage() + val generation = request?.generation ?: sessionGeneration + resumedGeneration = generation + finishResumeWhenHydrated(generation) } WsMethods.SESSION_INTERRUPT -> { @@ -980,6 +1021,8 @@ class ChatViewModel( error: Any?, ) { val method = idToMethod.remove(id) ?: return + val request = sessionRequestById.remove(id) + if (request != null && isStaleSessionRequest(request)) return val errorMsg = when (error) { is Map<*, *> -> error["message"] as? String ?: error.toString() @@ -992,9 +1035,11 @@ class ChatViewModel( // brief backoff usually clears it. Persistent failure ends in the // explicit error + Retry state. if (method == WsMethods.SESSION_RESUME) { - val sessionId = _uiState.value.currentSessionId + val sessionId = request?.sessionId ?: _uiState.value.currentSessionId if (sessionId != null) { - handleResumeFailure(sessionId, errorMsg) + val generation = request?.generation ?: sessionGeneration + if (resumedGeneration == generation) resumedGeneration = -1L + handleResumeFailure(sessionId, generation, errorMsg) } return } @@ -1275,11 +1320,12 @@ class ChatViewModel( val arg = command.split(" ", limit = 2).getOrElse(1) { "" }.trim() val params = mutableMapOf("session_id" to sessionId) if (arg.isNotBlank()) params["name"] = arg + val generation = sessionGeneration viewModelScope.launch(Dispatchers.IO) { wsClient.send( WsMethods.SESSION_BRANCH, params, - onSent = { id -> trackRequest(id, WsMethods.SESSION_BRANCH) }, + onSent = { id -> trackSessionRequest(id, WsMethods.SESSION_BRANCH, generation) }, ) } } @@ -1429,35 +1475,24 @@ class ChatViewModel( } } - private var sessionCreateCounter = 0L - fun createNewSession(setLoading: Boolean = true) { // A fresh create has no persisted row until the first prompt. sessionHasServerPresence = false - _uiState.update { - it.copy( - isLoading = setLoading, - messages = emptyList(), - chatTitle = "Hermes", - ) - } - _streamingState.update { StreamingState() } - streamingController.resetStreaming() + val generation = resetSessionState(sessionId = null, title = "Hermes", isLoading = setLoading) viewModelScope.launch(Dispatchers.IO) { wsClient.send( WsMethods.SESSION_CREATE, params = mapOf("source" to "desktop"), - onSent = { id -> trackRequest(id, WsMethods.SESSION_CREATE) }, + onSent = { id -> trackSessionRequest(id, WsMethods.SESSION_CREATE, generation) }, ) } // B7 safety timeout: clear loading state if RPC response never arrives if (setLoading && !isTestEnvironment()) { - val generation = ++sessionCreateCounter viewModelScope.launch { delay(10_000L) // Only clear if no newer session creation has started — prevents a // stale timeout from wiping the loading flag of a subsequent request. - if (generation == sessionCreateCounter && _uiState.value.isLoading) { + if (generation == sessionGeneration && _uiState.value.isLoading) { _uiState.update { it.copy(isLoading = false) } } } @@ -1487,7 +1522,7 @@ class ChatViewModel( // No server-side copy yet (created but never prompted): the REST // transcript 404s and would burn the resume retry budget for nothing. if (!sessionHasServerPresence) return - loadSessionMessages(sessionId) + loadSessionMessages(sessionId, sessionGeneration) } fun refreshSettings() { @@ -1695,10 +1730,6 @@ class ChatViewModel( fun switchSession(sessionId: String) { if (sessionId == _uiState.value.currentSessionId) return - // Reset streaming, pagination, and any pending resume retry before - // resuming the Desktop session. - cancelResumeRetry() - runtimeSessionId = null // The id came from the gateway's own session list / picker — its row // is expected to exist, so resume it optimistically on reconnect even // before the REST page confirms (a transient 500 must not strand the @@ -1707,57 +1738,34 @@ class ChatViewModel( sessionHasServerPresence = true sessionGoneRecoveryInFlight = false pendingGoneSessionNotice = false - loadedMessageOffset = 0 - streamingController.resetStreaming() - _uiState.update { - val title = it.sessions.find { s -> s.id == sessionId }?.title ?: "Hermes" - it.copy( - isLoading = true, - isLoadingOlder = false, - hasOlderMessages = false, - currentSessionId = sessionId, - messages = emptyList(), - chatTitle = title, - showSessionPicker = false, - isAgentTyping = false, - usedContextTokens = null, - fullContextTokens = null, - contextBreakdown = null, - compressionCount = null, - resumeError = null, - isResumeRetrying = false, - ) - } - // Mirror the active session id app-wide (issue #532). - ActiveSessionHolder.set(sessionId) - _streamingState.update { StreamingState() } + val title = _uiState.value.sessions.find { it.id == sessionId }?.title ?: "Hermes" + val generation = resetSessionState(sessionId, title, isLoading = true) viewModelScope.launch { // Warm-cache fast-path (desktop parity): paint the cached Room // transcript immediately so the screen never sits blank, then load // the fresh server transcript in parallel. If the cache is empty // the spinner stays up until the server page lands. - loadCachedMessages(sessionId) - // Resume the selected desktop session, then load its complete transcript. - launch(Dispatchers.IO) { - wsClient.send( - WsMethods.SESSION_RESUME, - mapOf("session_id" to sessionId), - onSent = { id -> trackRequest(id, WsMethods.SESSION_RESUME) }, - ) - } - loadSessionMessages(sessionId) + loadCachedMessages(sessionId, generation) + // Resume the selected desktop session and hydrate its transcript. + resumeSession(sessionId, generation) loadSessions() } } - private fun loadCachedMessages(sessionId: String): Job = + private fun loadCachedMessages( + sessionId: String, + generation: Long, + ): Job = viewModelScope.launch(Dispatchers.IO) { val cachedMessages = dedupeCachedMessages(repo.loadMessages(sessionId)) _uiState.update { state -> // Only paint if still showing this session AND no fresher server // page has landed yet (a fast REST fetch must not be clobbered // by a stale cache read that loses the race). - if (state.currentSessionId == sessionId && state.messages.isEmpty()) { + if (isCurrentSessionRequest(sessionId, generation) && + state.messages.isEmpty() && + cachedMessages.isNotEmpty() + ) { state.copy(messages = cachedMessages, isLoading = false) } else { state @@ -1765,11 +1773,18 @@ class ChatViewModel( } } - private fun loadSessionMessages(sessionId: String) { + private fun loadSessionMessages( + sessionId: String, + generation: Long, + ) { + val requestSequence = ++hydrationRequestSequence + activeHydrationRequestSequence = requestSequence viewModelScope.launch { - val messageCount = fetchServerMessageCount(sessionId) + val messageCount = fetchServerMessageCount(sessionId, generation, requestSequence) + if (!isCurrentHydration(sessionId, generation, requestSequence)) return@launch val offset = (messageCount - MESSAGE_PAGE_SIZE).coerceAtLeast(0) val result = fetchMessagePage(sessionId, offset, MESSAGE_PAGE_SIZE) + if (!isCurrentHydration(sessionId, generation, requestSequence)) return@launch when (result) { is NetworkResult.Success -> { // REST 200 — the gateway has the row for this session. @@ -1780,8 +1795,9 @@ class ChatViewModel( withContext(Dispatchers.IO) { repo.persistMessages(chatMessages, sessionId) } + if (!isCurrentHydration(sessionId, generation, requestSequence)) return@launch _uiState.update { state -> - if (state.currentSessionId != sessionId) return@update state + if (!isCurrentHydration(sessionId, generation, requestSequence)) return@update state // Merge, don't replace: a reload mid-turn must not // drop live WS bubbles (running tool call, streaming // answer) the server hasn't persisted yet. Issue #771. @@ -1793,11 +1809,14 @@ class ChatViewModel( isLoadingOlder = false, ) } + hydratedGeneration = generation + finishResumeWhenHydrated(generation) } is NetworkResult.Failure -> { + if (hydratedGeneration == generation) hydratedGeneration = -1L _uiState.update { - if (it.currentSessionId != sessionId) return@update it + if (!isCurrentHydration(sessionId, generation, requestSequence)) return@update it it.copy( isLoading = false, isLoadingOlder = false, @@ -1807,7 +1826,11 @@ class ChatViewModel( // retry (desktop parity) instead of a one-shot snackbar — // a transient backend/network blip recovers on its own, // and a persistent failure ends in an explicit Retry. - handleResumeFailure(sessionId, "Failed to load messages: ${result.error.message}") + handleResumeFailure( + sessionId, + generation, + "Failed to load messages: ${result.error.message}", + ) } } } @@ -1815,6 +1838,86 @@ class ChatViewModel( // ── Session resume recovery (desktop parity) ───────────────────────── + private fun resetSessionState( + sessionId: String?, + title: String, + isLoading: Boolean, + ): Long { + val generation = ++sessionGeneration + cancelResumeRetry() + resumedGeneration = -1L + hydratedGeneration = -1L + runtimeSessionId = null + ActiveSessionHolder.set(sessionId) + loadedMessageOffset = 0 + isSyncingMessages = false + streamingController.resetStreaming() + _streamingState.value = StreamingState() + _uiState.update { + it.copy( + messages = emptyList(), + currentSessionId = sessionId, + chatTitle = title, + isAgentTyping = false, + isThinking = false, + thinkingText = "", + isLoading = isLoading, + isLoadingOlder = false, + hasOlderMessages = false, + streamingMessage = null, + errorMessage = null, + openError = null, + clarifyRequest = null, + sudoPrompt = null, + secretPrompt = null, + showSessionPicker = false, + isSearchActive = false, + searchQuery = "", + searchMatchIndices = emptyList(), + currentSearchMatchIndex = -1, + showModelPicker = false, + modelPickerLoading = false, + currentSessionModel = null, + reasoningLevel = null, + usedContextTokens = null, + fullContextTokens = null, + contextBreakdown = null, + compressionCount = null, + pendingAttachments = emptyList(), + reactionKind = null, + subagentIndicators = emptyList(), + todos = emptyList(), + resumeError = null, + isResumeRetrying = false, + ) + } + return generation + } + + private fun resumeSession( + sessionId: String, + generation: Long, + ) { + val requestSequence = ++resumeRequestSequence + activeResumeRequestSequence = requestSequence + viewModelScope.launch(Dispatchers.IO) { + wsClient.send( + WsMethods.SESSION_RESUME, + mapOf("session_id" to sessionId), + onSent = { id -> + trackSessionRequest( + id = id, + method = WsMethods.SESSION_RESUME, + generation = generation, + resumeSequence = requestSequence, + sessionId = sessionId, + ) + }, + ) + } + loadSessionMessages(sessionId, generation) + } + private fun cancelResumeRetry() { resumeRetryJob?.cancel() resumeRetryJob = null @@ -1822,6 +1925,24 @@ class ChatViewModel( resumeRetryAttempt = 0 } + private fun finishResumeWhenHydrated(generation: Long) { + if (generation != sessionGeneration || + resumedGeneration != generation || + hydratedGeneration != generation + ) { + return + } + cancelResumeRetry() + _uiState.update { + it.copy( + isLoading = false, + isResumeRetrying = false, + resumeError = null, + errorMessage = null, + ) + } + } + private fun resumeRetryDelayMs(attempt: Int): Long = minOf(RESUME_RETRY_MAX_MS, RESUME_RETRY_BASE_MS * (1L shl attempt)) @@ -1873,10 +1994,12 @@ class ChatViewModel( private fun handleResumeFailure( sessionId: String, + generation: Long, errorMessage: String, ) { // Only handle if still on this session. - if (_uiState.value.currentSessionId != sessionId) return + if (!isCurrentSessionRequest(sessionId, generation)) return + _uiState.update { it.copy(errorMessage = null) } // New session → reset the counter for a fresh backoff cycle. if (resumeRetrySessionId != sessionId) { @@ -1898,6 +2021,7 @@ class ChatViewModel( isLoading = false, isResumeRetrying = false, resumeError = errorMessage, + errorMessage = null, ) } return @@ -1918,6 +2042,7 @@ class ChatViewModel( isLoading = false, isResumeRetrying = true, resumeError = null, + errorMessage = null, ) } @@ -1927,7 +2052,7 @@ class ChatViewModel( delay(delayMs) // Re-check liveness at fire time: the user may have switched // sessions or the gateway may have reconnected meanwhile. - if (_uiState.value.currentSessionId != sessionId) return@launch + if (!isCurrentSessionRequest(sessionId, generation)) return@launch _uiState.update { it.copy(isResumeRetrying = false) } if (_uiState.value.messages.isEmpty()) { @@ -1935,14 +2060,7 @@ class ChatViewModel( } // Retry the full resume: rebind the runtime via WS + refresh // the transcript via REST. Both are idempotent. - launch(Dispatchers.IO) { - wsClient.send( - WsMethods.SESSION_RESUME, - mapOf("session_id" to sessionId), - onSent = { id -> trackRequest(id, WsMethods.SESSION_RESUME) }, - ) - } - loadSessionMessages(sessionId) + resumeSession(sessionId, generation) } } @@ -1953,6 +2071,7 @@ class ChatViewModel( */ fun retryResumeSession() { val sessionId = _uiState.value.currentSessionId ?: return + val generation = sessionGeneration cancelResumeRetry() _uiState.update { it.copy( @@ -1961,21 +2080,13 @@ class ChatViewModel( isLoading = true, ) } - viewModelScope.launch { - launch(Dispatchers.IO) { - wsClient.send( - WsMethods.SESSION_RESUME, - mapOf("session_id" to sessionId), - onSent = { id -> trackRequest(id, WsMethods.SESSION_RESUME) }, - ) - } - loadSessionMessages(sessionId) - } + resumeSession(sessionId, generation) } fun loadOlderMessages() { val state = _uiState.value val sessionId = state.currentSessionId ?: return + val generation = sessionGeneration if (!state.hasOlderMessages || state.isLoadingOlder || loadedMessageOffset <= 0) return val oldOffset = loadedMessageOffset val newOffset = (oldOffset - MESSAGE_PAGE_SIZE).coerceAtLeast(0) @@ -1984,12 +2095,13 @@ class ChatViewModel( viewModelScope.launch { when (val result = fetchMessagePage(sessionId, newOffset, limit)) { is NetworkResult.Success -> { + if (!isCurrentSessionRequest(sessionId, generation)) return@launch val returnedOffset = result.data.offset ?: newOffset val older = mapServerMessages(sessionId, result.data.messages.orEmpty(), returnedOffset) loadedMessageOffset = returnedOffset withContext(Dispatchers.IO) { repo.persistMessages(older, sessionId) } _uiState.update { current -> - if (current.currentSessionId != sessionId) return@update current + if (!isCurrentSessionRequest(sessionId, generation)) return@update current current.copy( messages = (older + current.messages).distinctBy { it.id }, isLoadingOlder = false, @@ -1999,7 +2111,13 @@ class ChatViewModel( } is NetworkResult.Failure -> { - _uiState.update { it.copy(isLoadingOlder = false) } + _uiState.update { + if (isCurrentSessionRequest(sessionId, generation)) { + it.copy(isLoadingOlder = false) + } else { + it + } + } } } } @@ -2008,6 +2126,7 @@ class ChatViewModel( fun syncCurrentSession() { val state = _uiState.value val sessionId = state.currentSessionId ?: return + val generation = sessionGeneration if (isSyncingMessages || state.isLoading || state.isLoadingOlder || state.isAgentTyping || _streamingState.value.streamingMessage != null ) { @@ -2024,11 +2143,12 @@ class ChatViewModel( try { when (val result = fetchMessagePage(sessionId, nextOffset, MESSAGE_PAGE_SIZE)) { is NetworkResult.Success -> { + if (!isCurrentSessionRequest(sessionId, generation)) return@launch val incoming = mapServerMessages(sessionId, result.data.messages.orEmpty(), nextOffset) if (incoming.isEmpty()) return@launch withContext(Dispatchers.IO) { repo.persistMessages(incoming, sessionId) } _uiState.update { current -> - if (current.currentSessionId != sessionId) return@update current + if (!isCurrentSessionRequest(sessionId, generation)) return@update current // Issue #771: the sync merge was dropping the // newest tool bubble — the incoming REST page // didn't include it yet (server persists tool rows @@ -2094,7 +2214,7 @@ class ChatViewModel( is NetworkResult.Failure -> {} } } finally { - isSyncingMessages = false + if (generation == sessionGeneration) isSyncingMessages = false } } } @@ -2215,7 +2335,11 @@ class ChatViewModel( } } - private suspend fun fetchServerMessageCount(sessionId: String): Int { + private suspend fun fetchServerMessageCount( + sessionId: String, + generation: Long, + requestSequence: Long, + ): Int { val known = _uiState.value.sessions .find { it.id == sessionId } @@ -2230,23 +2354,30 @@ class ChatViewModel( val count = sessions.find { it.id == sessionId }?.message_count if (count != null) { _uiState.update { current -> - current.copy( - sessions = - current.sessions.map { - if (it.id == sessionId) { - it.copy( - messageCount = count, - ) - } else { - it - } - }, - ) + if (isCurrentHydration(sessionId, generation, requestSequence)) { + current.copy( + sessions = + current.sessions.map { + if (it.id == sessionId) { + it.copy(messageCount = count) + } else { + it + } + }, + ) + } else { + current + } } return count } } - return known ?: _uiState.value.messages.size + return known + ?: if (isCurrentHydration(sessionId, generation, requestSequence)) { + _uiState.value.messages.size + } else { + 0 + } } private suspend fun fetchMessagePage( @@ -2292,9 +2423,6 @@ class ChatViewModel( } } - val baseUrl = AuthManager.getBaseUrl() - val token = AuthManager.getToken().orEmpty() - val mapped = mutableListOf() // The gateway stores a reasoning-model's thinking as its OWN assistant // row (content = "", reasoning = trace) directly before the answer row. @@ -2340,6 +2468,8 @@ class ChatViewModel( if (role == MessageRole.ASSISTANT && rawContent.contains("MEDIA:")) { val items = HostMediaExtractor.extract(rawContent) if (items.isNotEmpty()) { + val baseUrl = AuthManager.getBaseUrl() + val token = AuthManager.getToken().orEmpty() finalContent = HostMediaExtractor.strip(rawContent) attachments = items @@ -2608,11 +2738,6 @@ class ChatViewModel( _uiState.update { it.copy(openError = null) } } - private fun serverMessageIndex( - id: String, - sessionId: String, - ): Int? = id.removePrefix("rest-$sessionId-").takeIf { it != id }?.toIntOrNull() - private fun sameMessages( left: List, right: List, @@ -3004,6 +3129,41 @@ class ChatViewModel( idToMethod[id] = method } + private fun trackSessionRequest( + id: String, + method: String, + generation: Long, + resumeSequence: Long = 0L, + sessionId: String? = null, + ) { + sessionRequestById[id] = SessionRequest(generation, resumeSequence, sessionId) + trackRequest(id, method) + } + + private fun isCurrentSessionRequest( + sessionId: String, + generation: Long, + ): Boolean = generation == sessionGeneration && sessionId == _uiState.value.currentSessionId + + private fun isCurrentHydration( + sessionId: String, + generation: Long, + requestSequence: Long, + ): Boolean = requestSequence == activeHydrationRequestSequence && isCurrentSessionRequest(sessionId, generation) + + private fun isStaleSessionRequest(id: String): Boolean = + sessionRequestById[id]?.let(::isStaleSessionRequest) == true + + private fun isStaleSessionRequest(request: SessionRequest): Boolean = + request.generation != sessionGeneration || + (request.sessionId != null && request.sessionId != _uiState.value.currentSessionId) || + (request.resumeSequence != 0L && request.resumeSequence != activeResumeRequestSequence) + + private fun forgetRequest(id: String) { + idToMethod.remove(id) + sessionRequestById.remove(id) + } + // ── Search ──────────────────────────────────────────────────────────── // Compatibility façade: stable public API around ChatSearchDelegate. // These thin delegates keep ChatViewModel's public surface intact while diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatToolDedupeTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatToolDedupeTest.kt index 07e2b4e3..035c9038 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatToolDedupeTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatToolDedupeTest.kt @@ -4,6 +4,7 @@ import com.m57.hermescontrol.data.ws.WsEvent import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNull +import org.junit.Assert.assertSame import org.junit.Assert.assertTrue import org.junit.Test @@ -154,6 +155,17 @@ class ChatToolDedupeTest { assertEquals("terminal", merged[1].toolName) } + @Test + fun merge_matchingUser_keepsLiveOnlyMetadata() { + val live = ChatMessage(role = MessageRole.USER, content = "redirect", isStreaming = true, timestamp = 1) + val rest = ChatMessage(role = MessageRole.USER, content = "redirect", id = "rest-s-0", timestamp = 1) + + val merged = mergeTranscriptWithLive(listOf(rest), listOf(live)) + + // Preserves cache/live-only user flags such as continuesActiveTurn. + assertSame(live, merged.single()) + } + @Test fun merge_reusedLiveToolId_noDuplicate() { // Post-completion reload: mapServerMessages reuses the live WS tool @@ -405,4 +417,23 @@ class ChatToolDedupeTest { // → it's kept. Verify sameLogicalMessage returns false for all incoming. assertTrue(incoming.none { sameLogicalMessage(it, existingTool) }) } + + @Test + fun incrementalMerge_keepsEarlierDuplicateContent() { + val live = ChatMessage(role = MessageRole.USER, content = "retry", isStreaming = true, timestamp = 2) + val current = + listOf( + ChatMessage(id = "rest-s-0", role = MessageRole.USER, content = "retry", timestamp = 1), + live, + ) + val incoming = + listOf( + ChatMessage(id = "rest-s-1", role = MessageRole.USER, content = "retry", timestamp = 2), + ) + + val merged = mergeIncrementalTranscriptPage(incoming, current, "s", 1) + + assertEquals(listOf("retry", "retry"), merged.map { it.content }) + assertSame(live, merged.last()) + } } diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt index 21c5398b..40101bb2 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt @@ -14,6 +14,7 @@ import com.m57.hermescontrol.data.remote.ApiClient import com.m57.hermescontrol.data.remote.GatewayFile import com.m57.hermescontrol.data.remote.GatewayFileClient import com.m57.hermescontrol.data.remote.GatewayFileResult +import com.m57.hermescontrol.data.session.ActiveSessionHolder import com.m57.hermescontrol.data.ws.ConnectionStatus import com.m57.hermescontrol.data.ws.HermesWsClient import com.m57.hermescontrol.data.ws.JsonRpcError @@ -35,6 +36,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.first import kotlinx.coroutines.test.StandardTestDispatcher import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.advanceUntilIdle @@ -42,6 +44,7 @@ import kotlinx.coroutines.test.resetMain import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.setMain +import kotlinx.coroutines.withTimeout import kotlinx.serialization.json.JsonPrimitive import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.put @@ -89,6 +92,7 @@ class ChatViewModelTest { app = mockk(relaxed = true) fakeRepo = FakeChatPersistenceRepository() + ActiveSessionHolder.set(null) mockConnectionStatus.value = ConnectionStatus.DISCONNECTED @@ -249,6 +253,30 @@ class ChatViewModelTest { assertEquals("my-fork", branchSent.second["name"]) } + @Test + fun testSessionBranch_publishesRuntimeSessionId() = + runTest { + val (viewModel, _) = createViewModelWithSession() + every { HermesWsClient.send(any(), any(), any()) } answers { + arg<((String) -> Unit)?>(2)?.invoke("branch-request") + "branch-request" + } + + viewModel.sendMessage("/fork") + mockEventsFlow.emit( + WsEvent.RpcResult( + "branch-request", + mapOf( + "session_id" to "runtime-branch", + "stored_session_id" to "stored-branch", + ), + ), + ) + advanceUntilIdle() + + assertEquals("runtime-branch", ActiveSessionHolder.activeSessionId.value) + } + @Test fun testSlashCommand_fork_withoutName_omitsNameParam() = runTest { @@ -1165,6 +1193,317 @@ class ChatViewModelTest { verify { HermesWsClient.send(WsMethods.SESSION_RESUME, mapOf("session_id" to "session-456"), any()) } } + @Test + fun testSwitchSession_ignoresLateResumeResultFromPreviousSelection() = + runTest { + stubEmptySessionRests("session-a", "session-b") + val (viewModel, _) = createViewModelWithSession() + val resumeRequests = mutableMapOf() + every { HermesWsClient.send(WsMethods.SESSION_RESUME, any(), any()) } answers { + val sessionId = arg>(1).getValue("session_id") + val requestId = "resume-$sessionId" + resumeRequests[sessionId] = requestId + arg<((String) -> Unit)?>(2)?.invoke(requestId) + requestId + } + + viewModel.switchSession("session-a") + runCurrent() + viewModel.switchSession("session-b") + runCurrent() + mockEventsFlow.emit( + WsEvent.RpcResult( + resumeRequests.getValue("session-b"), + mapOf("session_id" to "runtime-b", "resumed" to "session-b"), + ), + ) + mockEventsFlow.emit( + WsEvent.RpcResult( + resumeRequests.getValue("session-a"), + mapOf("session_id" to "runtime-a", "resumed" to "session-a"), + ), + ) + advanceUntilIdle() + + assertEquals("session-b", viewModel.uiState.value.currentSessionId) + assertEquals("runtime-b", ActiveSessionHolder.activeSessionId.value) + } + + @Test + fun testReconnect_ignoresSupersededResumeForSameSession() = + runTest { + stubEmptySessionRests("session-a") + val (viewModel, _) = createViewModelWithSession() + val resumeRequests = mutableListOf() + every { HermesWsClient.send(WsMethods.SESSION_RESUME, any(), any()) } answers { + val requestId = "resume-${resumeRequests.size + 1}" + resumeRequests += requestId + arg<((String) -> Unit)?>(2)?.invoke(requestId) + requestId + } + + viewModel.switchSession("session-a") + runCurrent() + mockEventsFlow.emit(WsEvent.GatewayReady(null)) + runCurrent() + mockEventsFlow.emit( + WsEvent.RpcResult(resumeRequests[1], mapOf("session_id" to "runtime-new")), + ) + mockEventsFlow.emit( + WsEvent.RpcResult(resumeRequests[0], mapOf("session_id" to "runtime-old")), + ) + advanceUntilIdle() + + assertEquals("session-a", viewModel.uiState.value.currentSessionId) + assertEquals("runtime-new", ActiveSessionHolder.activeSessionId.value) + } + + @Test + fun testSwitchSession_ignoresLateResumeErrorBeforeReducerMutation() = + runTest { + val mockApi = ApiClient.hermesApi + val messagesA = + CompletableDeferred>() + val messagesB = + CompletableDeferred>() + every { AuthManager.getBaseUrl() } returns "http://test.local/" + coEvery { mockApi.getSessions(any(), any(), any()) } returns + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionListResponse(sessions = emptyList(), total = 0), + ) + coEvery { mockApi.getSessionMessages("session-a", any(), any(), any()) } coAnswers { messagesA.await() } + coEvery { mockApi.getSessionMessages("session-b", any(), any(), any()) } coAnswers { messagesB.await() } + + val (viewModel, _) = createViewModelWithSession() + val resumeRequests = mutableMapOf() + every { HermesWsClient.send(WsMethods.SESSION_RESUME, any(), any()) } answers { + val sessionId = arg>(1).getValue("session_id") + val requestId = "resume-$sessionId" + resumeRequests[sessionId] = requestId + arg<((String) -> Unit)?>(2)?.invoke(requestId) + requestId + } + + viewModel.switchSession("session-a") + runCurrent() + viewModel.switchSession("session-b") + runCurrent() + mockEventsFlow.emit( + WsEvent.RpcError( + resumeRequests.getValue("session-a"), + JsonRpcError(code = -32000, message = "stale resume failure"), + ), + ) + runCurrent() + + assertEquals("session-b", viewModel.uiState.value.currentSessionId) + assertTrue(viewModel.uiState.value.isLoading) + assertNull(viewModel.uiState.value.errorMessage) + assertNull(viewModel.uiState.value.resumeError) + + val empty = + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionMessagesResponse(messages = emptyList()), + ) + messagesA.complete(empty) + messagesB.complete(empty) + advanceUntilIdle() + } + + @Test + fun testSwitchSession_ignoresLateRestHydrationFromPreviousGeneration() = + runTest { + val mockApi = ApiClient.hermesApi + val messagesA = + CompletableDeferred>() + val messagesB = + CompletableDeferred>() + every { AuthManager.getBaseUrl() } returns "http://test.local/" + coEvery { mockApi.getSessions(any(), any(), any()) } returns + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionListResponse(sessions = emptyList(), total = 0), + ) + coEvery { mockApi.getSessionMessages("session-a", any(), any(), any()) } coAnswers { messagesA.await() } + coEvery { mockApi.getSessionMessages("session-b", any(), any(), any()) } coAnswers { messagesB.await() } + + val (viewModel, _) = createViewModelWithSession() + viewModel.switchSession("session-a") + runCurrent() + viewModel.switchSession("session-b") + runCurrent() + + messagesB.complete( + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionMessagesResponse( + messages = + listOf( + com.m57.hermescontrol.data.model.SessionMessage( + role = "assistant", + content = JsonPrimitive("message-b"), + ), + ), + ), + ), + ) + runCurrent() + messagesA.complete( + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionMessagesResponse( + messages = + listOf( + com.m57.hermescontrol.data.model.SessionMessage( + role = "assistant", + content = JsonPrimitive("stale-message-a"), + ), + ), + ), + ), + ) + advanceUntilIdle() + + assertEquals("session-b", viewModel.uiState.value.currentSessionId) + assertEquals(listOf("message-b"), viewModel.uiState.value.messages.map { it.content }) + } + + @Test + fun testSwitchSession_clearsSessionBoundUiState() = + runTest { + stubEmptySessionRests("session-b") + val (viewModel, sessionId) = createViewModelWithSession() + mockEventsFlow.emit( + WsEvent.ClarifyRequest("Choose", listOf("A"), "clarify-1", sessionId), + ) + mockEventsFlow.emit(WsEvent.SudoRequest("sudo-1", sessionId)) + mockEventsFlow.emit(WsEvent.SecretRequest("secret-1", sessionId)) + mockEventsFlow.emit( + WsEvent.SessionInfo( + mapOf("model" to "model-a", "provider" to "provider-a", "reasoning_effort" to "high"), + ), + ) + mockEventsFlow.emit( + WsEvent.SubagentEvent( + type = "subagent.start", + payload = mapOf("subagent_id" to "sub-1", "goal" to "work"), + sessionId = sessionId, + ), + ) + mockEventsFlow.emit( + WsEvent.ToolStart( + name = "todo", + data = mapOf("todos" to listOf(mapOf("id" to "todo-1", "content" to "work"))), + sessionId = sessionId, + ), + ) + mockEventsFlow.emit(WsEvent.GatewayError("old error")) + viewModel.openModelPicker() + runCurrent() + + val oldState = viewModel.uiState.value + assertNotNull(oldState.clarifyRequest) + assertNotNull(oldState.sudoPrompt) + assertNotNull(oldState.secretPrompt) + assertNotNull(oldState.currentSessionModel) + assertTrue(oldState.subagentIndicators.isNotEmpty()) + assertTrue(oldState.todos.isNotEmpty()) + assertTrue(oldState.showModelPicker) + assertNotNull(oldState.errorMessage) + + viewModel.switchSession("session-b") + runCurrent() + val state = viewModel.uiState.value + + assertEquals("session-b", state.currentSessionId) + assertTrue(state.messages.isEmpty()) + assertNull(state.clarifyRequest) + assertNull(state.sudoPrompt) + assertNull(state.secretPrompt) + assertNull(state.currentSessionModel) + assertNull(state.reasoningLevel) + assertTrue(state.subagentIndicators.isEmpty()) + assertTrue(state.todos.isEmpty()) + assertFalse(state.showModelPicker) + assertNull(state.errorMessage) + assertNull(state.resumeError) + } + + @Test + fun testCreateNewSession_invalidatesResumeAndClearsSessionStateBeforeSend() = + runTest { + stubEmptySessionRests("session-a") + val (viewModel, _) = createViewModelWithSession() + var resumeRequestId = "" + every { HermesWsClient.send(WsMethods.SESSION_RESUME, any(), any()) } answers { + resumeRequestId = "resume-a" + arg<((String) -> Unit)?>(2)?.invoke(resumeRequestId) + resumeRequestId + } + viewModel.switchSession("session-a") + advanceUntilIdle() + mockEventsFlow.emit(WsEvent.SudoRequest("sudo-1", "session-a")) + mockEventsFlow.emit(WsEvent.GatewayError("old error")) + runCurrent() + + viewModel.createNewSession() + mockEventsFlow.emit( + WsEvent.RpcError( + resumeRequestId, + JsonRpcError(code = -32000, message = "stale resume failure"), + ), + ) + runCurrent() + + val state = viewModel.uiState.value + assertNull(state.currentSessionId) + assertNull(state.sudoPrompt) + assertNull(state.errorMessage) + assertNull(state.resumeError) + assertTrue(state.isLoading) + } + + @Test + fun testSessionBranchResult_clearsSessionBoundUiState() = + runTest { + stubEmptySessionRests("branch-1") + val (viewModel, sessionId) = createViewModelWithSession() + val captured = captureSends() + mockEventsFlow.emit(WsEvent.SecretRequest("secret-1", sessionId)) + mockEventsFlow.emit( + WsEvent.SubagentEvent( + type = "subagent.start", + payload = mapOf("subagent_id" to "sub-1", "goal" to "work"), + sessionId = sessionId, + ), + ) + mockEventsFlow.emit( + WsEvent.ToolStart( + name = "todo", + data = mapOf("todos" to listOf(mapOf("id" to "todo-1", "content" to "work"))), + sessionId = sessionId, + ), + ) + mockEventsFlow.emit(WsEvent.SessionInfo(mapOf("model" to "model-a", "provider" to "provider-a"))) + runCurrent() + + viewModel.sendMessage("/fork") + runCurrent() + val branchRequest = captured.last { it.first == WsMethods.SESSION_BRANCH }.second + mockEventsFlow.emit( + WsEvent.RpcResult( + branchRequest, + mapOf("session_id" to "branch-1", "title" to "Branch"), + ), + ) + runCurrent() + + val state = viewModel.uiState.value + assertEquals("branch-1", state.currentSessionId) + assertNull(state.secretPrompt) + assertNull(state.currentSessionModel) + assertTrue(state.subagentIndicators.isEmpty()) + assertTrue(state.todos.isEmpty()) + assertNull(state.errorMessage) + } + // ── Session resume recovery (desktop parity: warm cache + bounded retry) ── /** Override the send stub to capture (method → id) pairs. */ @@ -1225,6 +1564,31 @@ class ChatViewModelTest { } } + private fun stubEmptySessionRests(vararg sessionIds: String) { + val mockApi = ApiClient.hermesApi + every { AuthManager.getBaseUrl() } returns "http://test.local/" + coEvery { mockApi.getSessions(any(), any(), any()) } returns + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionListResponse( + sessions = + sessionIds.map { + com.m57.hermescontrol.data.model.SessionInfo( + id = it, + title = it, + message_count = 0, + ) + }, + total = sessionIds.size, + ), + ) + sessionIds.forEach { sessionId -> + coEvery { mockApi.getSessionMessages(sessionId, any(), any(), any()) } returns + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionMessagesResponse(messages = emptyList()), + ) + } + } + @Test fun testSwitchSession_paintsWarmCache_andKeepsItWhenResumeExhausted() = runTest { @@ -1561,6 +1925,49 @@ class ChatViewModelTest { ) } + @Test + fun testSessionResume_wsSuccessWaitsForRestRetry() = + runTest { + stubSession456Rests(success = true) + val mockApi = ApiClient.hermesApi + var restCalls = 0 + coEvery { + mockApi.getSessionMessages("session-456", any(), any(), any()) + } coAnswers { + val call = restCalls++ + if (call < 3) { + retrofit2.Response.error( + 500, + okhttp3.ResponseBody.create(null, "{\"detail\":\"boom\"}"), + ) + } else { + retrofit2.Response.success( + com.m57.hermescontrol.data.model.SessionMessagesResponse(messages = emptyList()), + ) + } + } + val (viewModel, _) = createViewModelWithSession() + val captured = captureSends() + + viewModel.switchSession("session-456") + withTimeout(2_000) { viewModel.uiState.first { it.isResumeRetrying } } + + val resumeId = captured.last { it.first == WsMethods.SESSION_RESUME }.second + mockEventsFlow.emit( + WsEvent.RpcResult(resumeId, mapOf("session_id" to "runtime-456")), + ) + runCurrent() + + assertTrue("WS success must leave the REST retry armed", viewModel.uiState.value.isResumeRetrying) + assertEquals(1, captured.count { it.first == WsMethods.SESSION_RESUME }) + + advanceUntilIdle() + + assertFalse(viewModel.uiState.value.isResumeRetrying) + assertNull(viewModel.uiState.value.resumeError) + assertEquals(2, captured.count { it.first == WsMethods.SESSION_RESUME }) + } + @Test fun testRetryResumeSession_clearsErrorAndResends() = runTest {