* 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>
This commit is contained in:
1 parent
b28440d659
commit
be0bbcb2c9
3 files changed
+740
-142
No files matched your search
@@ -154,15 +154,47 @@ internal fun mergeTranscriptWithLive(
|
||||
restMessages: List<ChatMessage>,
|
||||
currentMessages: List<ChatMessage>,
|
||||
): List<ChatMessage> {
|
||||
// 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<ChatMessage>,
|
||||
currentMessages: List<ChatMessage>,
|
||||
sessionId: String,
|
||||
offset: Int,
|
||||
): List<ChatMessage> {
|
||||
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<ChatMessage> = 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<String, String>()
|
||||
|
||||
private data class SessionRequest(
|
||||
val generation: Long,
|
||||
val resumeSequence: Long = 0L,
|
||||
val sessionId: String? = null,
|
||||
)
|
||||
|
||||
private val sessionRequestById = ConcurrentHashMap<String, SessionRequest>()
|
||||
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<String, Any?> ?: 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<String, Any>("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<ChatMessage>()
|
||||
// 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<ChatMessage>,
|
||||
right: List<ChatMessage>,
|
||||
@@ -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
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
@@ -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<String, String>()
|
||||
every { HermesWsClient.send(WsMethods.SESSION_RESUME, any(), any()) } answers {
|
||||
val sessionId = arg<Map<String, String>>(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<String>()
|
||||
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<retrofit2.Response<com.m57.hermescontrol.data.model.SessionMessagesResponse>>()
|
||||
val messagesB =
|
||||
CompletableDeferred<retrofit2.Response<com.m57.hermescontrol.data.model.SessionMessagesResponse>>()
|
||||
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<String, String>()
|
||||
every { HermesWsClient.send(WsMethods.SESSION_RESUME, any(), any()) } answers {
|
||||
val sessionId = arg<Map<String, String>>(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<retrofit2.Response<com.m57.hermescontrol.data.model.SessionMessagesResponse>>()
|
||||
val messagesB =
|
||||
CompletableDeferred<retrofit2.Response<com.m57.hermescontrol.data.model.SessionMessagesResponse>>()
|
||||
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 {
|
||||
|
||||
Reference in new issue
Block a user