refactor(core): execute Phase 3 state and delegate extractions

- Extract ChatClarifyDelegate from ChatViewModel for clarify prompt flow.
- Harden GroupChatViewModel maps with ConcurrentHashMap.
- Add failure circuit breaker to MCP OAuth polling loop.
This commit is contained in:
Hy4ri committed 2026-09-11 20:13:00 +03:00
1 parent 5f79bdee54
commit 02106c3781
4 files changed
+153 -107

No files matched your search

@@ -33,6 +33,7 @@ import kotlinx.serialization.json.JsonElement
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.decodeFromJsonElement
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
data class GroupChatUiState(
val groupName: String = "",
@@ -74,22 +75,22 @@ class GroupChatViewModel(
private var activeEpoch: String = ""
// Map of bot.name -> seen count of non-system messages
private val memberWatermarks = mutableMapOf<String, Int>()
private val memberWatermarks = ConcurrentHashMap<String, Int>()
// Map of bot.name -> active session info in this group
private val memberSessions = mutableMapOf<String, MemberSession>()
private val memberSessions = ConcurrentHashMap<String, MemberSession>()
// In-flight turn completion deferreds keyed by session_id
private val inFlightTurns = mutableMapOf<String, CompletableDeferred<String>>()
private val inFlightTurns = ConcurrentHashMap<String, CompletableDeferred<String>>()
// Map of session_id -> active streaming message ID
private val activeStreamMsgId = mutableMapOf<String, String>()
private val activeStreamMsgId = ConcurrentHashMap<String, String>()
// Map of session_id -> bot profile
private val sessionToBot = mutableMapOf<String, ProfileInfo>()
private val sessionToBot = ConcurrentHashMap<String, ProfileInfo>()
// Map of session_id -> last activity epoch ms (streaming token, tool call, thinking)
private val sessionLastActivity = mutableMapOf<String, Long>()
private val sessionLastActivity = ConcurrentHashMap<String, Long>()
init {
if (groupName.isNotBlank()) {
@@ -0,0 +1,123 @@
package com.m57.hermescontrol.ui.chat
import com.m57.hermescontrol.data.ws.HermesWsClient
import com.m57.hermescontrol.data.ws.WsMethods
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch
class ChatClarifyDelegate(
private val uiState: MutableStateFlow<ChatUiState>,
private val scope: CoroutineScope,
private val ioDispatcher: CoroutineDispatcher,
private val persistMessage: suspend (ChatMessage, String) -> Unit,
private val wsClient: HermesWsClient = HermesWsClient,
private val trackRequest: (String, String) -> Unit = { _, _ -> },
) {
fun respondToClarify(option: String) {
val clarify = uiState.value.clarifyRequest
// Only use synthesized qid when this is a true batch or legacy with explicit qid.
val qid =
if (clarify != null &&
(clarify.questionId != null || clarify.questions.isNotEmpty())
) {
clarify.resolvedQuestions.firstOrNull()?.qid ?: clarify.questionId
} else {
null
}
if (qid != null && clarify?.resolvedQuestions?.size == 1) {
respondToClarifyBatch(mapOf(qid to option))
} else {
respondToClarifyBatch(emptyMap(), singleFallbackAnswer = option)
}
}
fun respondToClarifyBatch(
answers: Map<String, String>,
singleFallbackAnswer: String? = null,
) {
val sessionId = uiState.value.currentSessionId ?: return
val clarify = uiState.value.clarifyRequest
val clarifyId = clarify?.clarifyId
val isBatch = !clarify?.questions.isNullOrEmpty()
val questions = clarify?.resolvedQuestions.orEmpty()
uiState.update { it.copy(clarifyRequest = null) }
val displayContent =
if (questions.size > 1) {
questions
.mapIndexed { index, q ->
val ans = answers[q.qid]?.trim().orEmpty()
"${index + 1}. ${ans.ifEmpty { "(Skipped)" }}"
}.joinToString("\n")
} else {
val loneAns = answers.values.firstOrNull()?.trim() ?: singleFallbackAnswer?.trim().orEmpty()
loneAns
}
val userMessage =
ChatMessage(
role = MessageRole.USER,
content = displayContent,
)
uiState.update { state ->
state.copy(
messages = state.messages + userMessage,
isAgentTyping = true,
)
}
scope.launch(ioDispatcher) {
persistMessage(userMessage, sessionId)
}
scope.launch(ioDispatcher) {
if (isBatch) {
for (q in questions) {
val ans = answers[q.qid]?.trim().orEmpty()
val params =
mutableMapOf<String, Any>(
"session_id" to sessionId,
"response" to ans,
"answer" to ans,
"question_id" to q.qid,
)
if (clarifyId != null) {
params["clarify_id"] = clarifyId
params["request_id"] = clarifyId
}
wsClient.send(
method = WsMethods.CLARIFY_RESPOND,
params = params,
onSent = { id -> trackRequest(id, WsMethods.CLARIFY_RESPOND) },
)
}
} else {
// Legacy single: only include question_id when explicitly present.
val qid = clarify?.questionId
val ans = answers.values.firstOrNull() ?: singleFallbackAnswer.orEmpty()
val params =
mutableMapOf<String, Any>(
"session_id" to sessionId,
"response" to ans,
"answer" to ans,
)
if (clarifyId != null) {
params["clarify_id"] = clarifyId
params["request_id"] = clarifyId
}
if (qid != null) {
params["question_id"] = qid
}
wsClient.send(
method = WsMethods.CLARIFY_RESPOND,
params = params,
onSent = { id -> trackRequest(id, WsMethods.CLARIFY_RESPOND) },
)
}
}
}
}
@@ -435,6 +435,16 @@ class ChatViewModel(
addSystemMessage = { text -> addSystemMessage(text) },
)
private val clarifyDelegate =
ChatClarifyDelegate(
uiState = _uiState,
scope = viewModelScope,
ioDispatcher = ioDispatcher,
persistMessage = { msg, sid -> repo.persistMessage(msg, sid) },
wsClient = wsClient,
trackRequest = { id, method -> trackRequest(id, method) },
)
private val streamingController =
ChatStreamingController(
scope = viewModelScope,
@@ -3172,110 +3182,12 @@ class ChatViewModel(
}
}
fun respondToClarify(option: String) {
val clarify = _uiState.value.clarifyRequest
// Only use synthesized qid when this is a true batch or legacy with explicit qid.
val qid =
if (clarify != null &&
(clarify.questionId != null || clarify.questions.isNotEmpty())
) {
clarify.resolvedQuestions.firstOrNull()?.qid ?: clarify.questionId
} else {
null
}
if (qid != null && clarify?.resolvedQuestions?.size == 1) {
respondToClarifyBatch(mapOf(qid to option))
} else {
respondToClarifyBatch(emptyMap(), singleFallbackAnswer = option)
}
}
fun respondToClarify(option: String) = clarifyDelegate.respondToClarify(option)
fun respondToClarifyBatch(
answers: Map<String, String>,
singleFallbackAnswer: String? = null,
) {
val sessionId = _uiState.value.currentSessionId ?: return
val clarify = _uiState.value.clarifyRequest
val clarifyId = clarify?.clarifyId
val isBatch = !clarify?.questions.isNullOrEmpty()
val questions = clarify?.resolvedQuestions.orEmpty()
_uiState.update { it.copy(clarifyRequest = null) }
val displayContent =
if (questions.size > 1) {
questions
.mapIndexed { index, q ->
val ans = answers[q.qid]?.trim().orEmpty()
"${index + 1}. ${ans.ifEmpty { "(Skipped)" }}"
}.joinToString("\n")
} else {
val loneAns = answers.values.firstOrNull()?.trim() ?: singleFallbackAnswer?.trim().orEmpty()
loneAns
}
val userMessage =
ChatMessage(
role = MessageRole.USER,
content = displayContent,
)
_uiState.update { state ->
state.copy(
messages = state.messages + userMessage,
isAgentTyping = true,
)
}
viewModelScope.launch(ioDispatcher) {
repo.persistMessage(userMessage, sessionId)
}
viewModelScope.launch(ioDispatcher) {
if (isBatch) {
for (q in questions) {
val ans = answers[q.qid]?.trim().orEmpty()
val params =
mutableMapOf<String, Any>(
"session_id" to sessionId,
"response" to ans,
"answer" to ans,
"question_id" to q.qid,
)
if (clarifyId != null) {
params["clarify_id"] = clarifyId
params["request_id"] = clarifyId
}
wsClient.send(
method = WsMethods.CLARIFY_RESPOND,
params = params,
onSent = { id -> trackRequest(id, WsMethods.CLARIFY_RESPOND) },
)
}
} else {
// Legacy single: only include question_id when explicitly present.
val qid = clarify?.questionId
val ans = answers.values.firstOrNull() ?: singleFallbackAnswer.orEmpty()
val params =
mutableMapOf<String, Any>(
"session_id" to sessionId,
"response" to ans,
"answer" to ans,
)
if (clarifyId != null) {
params["clarify_id"] = clarifyId
params["request_id"] = clarifyId
}
if (qid != null) {
params["question_id"] = qid
}
wsClient.send(
method = WsMethods.CLARIFY_RESPOND,
params = params,
onSent = { id -> trackRequest(id, WsMethods.CLARIFY_RESPOND) },
)
}
}
}
) = clarifyDelegate.respondToClarifyBatch(answers, singleFallbackAnswer)
fun clearError() {
_uiState.update { it.copy(errorMessage = null) }
@@ -585,6 +585,7 @@ class McpServersViewModel :
oauthPollJob =
viewModelScope.launch {
var polling = true
var consecutiveFailures = 0
while (polling) {
kotlinx.coroutines.delay(2000)
val result =
@@ -629,7 +630,16 @@ class McpServersViewModel :
}
is NetworkResult.Failure -> {
// Keep polling or stop after too many failures? Let's just log/toast n retry a few times
consecutiveFailures++
if (consecutiveFailures >= 10) {
polling = false
_uiState.update {
it.copy(
activeOAuthFlow = null,
toastMessage = "OAuth polling timed out after repeated network failures",
)
}
}
}
}
}