feat(bots): reactive multi-agent group chat orchestration with CAS leasing (#1007)

Co-authored-by: M57 <hy4ri@users.noreply.github.com>
This commit is contained in:
M57andM57 authored and GitHub committed 2026-08-29 18:44:08 +03:00
1 parent 7e24f5eba3
commit 0d09f43469
9 files changed
+1059 -147

No files matched your search

@@ -157,6 +157,17 @@ data class GroupChatSyncSnapshot(
room.revision?.let { roomMap["revision"] = it }
room.syncRevision?.let { roomMap["syncRevision"] = it }
room.tombstone?.let { roomMap["tombstone"] = it }
room.maxBotMessages?.let { roomMap["maxBotMessages"] = it }
room.maxContinuationPasses?.let { roomMap["maxContinuationPasses"] = it }
room.systemPrompt?.let { roomMap["systemPrompt"] = it }
room.lease?.let { lease ->
roomMap["lease"] =
buildMap<String, Any?> {
lease.driverId?.let { put("driverId", it) }
lease.epoch?.let { put("epoch", it) }
lease.expiresAt?.let { put("expiresAt", it) }
}
}
room.members?.let { members ->
roomMap["members"] =
members.mapNotNull {
@@ -200,6 +211,13 @@ data class GroupChatSyncSnapshot(
}
}
@Serializable
data class GroupChatRoomLease(
val driverId: String? = null,
val epoch: String? = null,
val expiresAt: Long? = null,
)
@Serializable
data class GroupChatRoomMeta(
val id: String? = null,
@@ -214,6 +232,10 @@ data class GroupChatRoomMeta(
val updatedAt: Long? = null,
val createdAt: Long? = null,
val tombstone: Boolean? = null,
val lease: GroupChatRoomLease? = null,
val maxBotMessages: Int? = null,
val maxContinuationPasses: Int? = null,
val systemPrompt: String? = null,
) {
val memberNames: List<String>
get() =
@@ -3,6 +3,8 @@ package com.m57.hermescontrol.data.model
import kotlinx.serialization.Serializable
import kotlinx.serialization.json.JsonElement
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.doubleOrNull
import kotlinx.serialization.json.longOrNull
@Serializable
data class SessionMessagesResponse(
@@ -52,6 +54,20 @@ data class SessionMessage(
val timestampText: String?
get() = (timestamp as? JsonPrimitive)?.content
val timestampEpochMs: Long?
get() {
val prim = timestamp as? JsonPrimitive ?: return null
val asLong = prim.longOrNull
if (asLong != null) {
return if (asLong < 10_000_000_000L) asLong * 1000L else asLong
}
val asDouble = prim.doubleOrNull
if (asDouble != null) {
return (asDouble * 1000L).toLong()
}
return null
}
val contentText: String
get() =
when (content) {
@@ -16,12 +16,15 @@ data class GroupChatMessage(
val timestamp: Long = System.currentTimeMillis(),
val isStreaming: Boolean = false,
val thread: String? = null,
val isSystem: Boolean = false,
)
/**
* Pure helper for mention extraction and responder resolution matching Desktop parity.
*/
object GroupChatMentions {
private val MENTION_REGEX = Regex("""@([a-zA-Z0-9._-]+)""")
data class MentionResult(
val isEveryone: Boolean,
val mentionedBots: Set<String>,
@@ -33,16 +36,42 @@ object GroupChatMentions {
fun parseMentions(
text: String,
members: List<ProfileInfo>,
excludedSpeaker: String? = null,
): MentionResult {
val lowerText = text.lowercase()
val isEveryone = lowerText.contains("@all") || lowerText.contains("@everyone")
var isEveryone = false
val mentioned = mutableSetOf<String>()
val handles = mutableMapOf<String, String>()
for (member in members) {
val handle = member.name.lowercase()
val titleHandle = member.effectiveTitle.lowercase().replace(" ", "")
if (lowerText.contains("@$handle") || lowerText.contains("@$titleHandle")) {
mentioned.add(member.name)
val name = member.name.lowercase()
handles[name] = member.name
handles[name.replace(Regex("""[\s_-]+"""), "")] = member.name
val title = member.effectiveTitle.lowercase().trim()
if (title.isNotEmpty()) {
handles[title.replace(" ", "")] = member.name
val firstWord = title.split(Regex("""\s+""")).firstOrNull().orEmpty()
if (firstWord.isNotEmpty()) {
handles[firstWord] = member.name
}
}
}
for (match in MENTION_REGEX.findAll(text)) {
val rawHandle = match.groupValues[1].lowercase()
if (rawHandle == "all" || rawHandle == "everyone") {
isEveryone = true
continue
}
if (rawHandle == "user" || rawHandle == "you") {
continue
}
val resolved = handles[rawHandle] ?: handles[rawHandle.replace(Regex("""[._-]+"""), "")]
if (resolved != null) {
if (excludedSpeaker == null || !resolved.equals(excludedSpeaker, ignoreCase = true)) {
mentioned.add(resolved)
}
}
}
@@ -58,10 +87,15 @@ object GroupChatMentions {
fun resolveResponders(
text: String,
members: List<ProfileInfo>,
excludedSpeaker: String? = null,
): List<ProfileInfo> {
val parsed = parseMentions(text, members)
val parsed = parseMentions(text, members, excludedSpeaker)
return if (parsed.isEveryone || parsed.mentionedBots.isEmpty()) {
members
if (excludedSpeaker != null) {
members.filter { !it.name.equals(excludedSpeaker, ignoreCase = true) }
} else {
members
}
} else {
members.filter { parsed.mentionedBots.contains(it.name) }
}
@@ -75,11 +109,25 @@ object GroupChatMentions {
viewer: ProfileInfo,
peers: List<ProfileInfo>,
recentLog: List<GroupChatMessage>,
historyLimit: Int = 10,
customRoomPrompt: String? = null,
): String {
val viewerHandle = viewer.name
val peerNames = peers.joinToString(", ") { "@${it.name}" }
val viewerLabel =
if (viewer.effectiveTitle.isNotBlank() && !viewer.effectiveTitle.equals(viewer.name, ignoreCase = true)) {
"@${viewer.name} (${viewer.effectiveTitle})"
} else {
"@${viewer.name}"
}
val peerNames =
peers.joinToString(", ") {
if (it.effectiveTitle.isNotBlank() && !it.effectiveTitle.equals(it.name, ignoreCase = true)) {
"@${it.name} (${it.effectiveTitle})"
} else {
"@${it.name}"
}
}
val lines =
recentLog.takeLast(10).joinToString("\n") { msg ->
recentLog.filter { !it.isSystem }.takeLast(historyLimit).joinToString("\n") { msg ->
val speaker = if (msg.isUser) "User" else "@${msg.senderName}"
" $speaker: ${msg.text}"
}
@@ -90,20 +138,33 @@ object GroupChatMentions {
val actionGuidance =
if (isDirectOrAll) {
"- You were explicitly addressed (via @$viewerHandle or @all). Do NOT pass — reply conversationally with your greeting, insight, answer, or status."
"- You were explicitly addressed (via @${viewer.name} or @all). Do NOT pass — reply conversationally with your greeting, insight, answer, or status."
} else {
"- If you have nothing new to add, reply with exactly \"(pass)\". Passing is good and lets the conversation settle."
}
return """
[Group chat: "$groupName"] You are @$viewerHandle, one participant in a group chat with $peerNames and the user.
val customInstructions =
if (!customRoomPrompt.isNullOrBlank()) {
"\nRoom Instructions:\n${customRoomPrompt.trim()}\n"
} else {
""
}
val defaultSentenceRule =
if (customRoomPrompt.isNullOrBlank()) {
"- Reply with ONE conversational message (1-3 sentences).\n"
} else {
""
}
return """
[Group chat: "$groupName"] You are $viewerLabel, one participant in a group chat with $peerNames and the user.
$customInstructions
Recent messages in the room:
$lines
Rules for this room:
- Reply with ONE conversational message (1-3 sentences).
$actionGuidance
$defaultSentenceRule$actionGuidance
- Mention a teammate as @name if you want to pull them in.
- Your reply goes to the room verbatim — no preamble.
""".trimIndent()
@@ -24,13 +24,19 @@ import androidx.compose.foundation.text.KeyboardActions
import androidx.compose.foundation.text.KeyboardOptions
import androidx.compose.material.icons.Icons
import androidx.compose.material.icons.automirrored.filled.Send
import androidx.compose.material.icons.filled.Settings
import androidx.compose.material3.AlertDialog
import androidx.compose.material3.Button
import androidx.compose.material3.CircularProgressIndicator
import androidx.compose.material3.Icon
import androidx.compose.material3.IconButton
import androidx.compose.material3.IconButtonDefaults
import androidx.compose.material3.MaterialTheme
import androidx.compose.material3.OutlinedTextField
import androidx.compose.material3.Slider
import androidx.compose.material3.Surface
import androidx.compose.material3.Text
import androidx.compose.material3.TextButton
import androidx.compose.runtime.Composable
import androidx.compose.runtime.CompositionLocalProvider
import androidx.compose.runtime.LaunchedEffect
@@ -60,6 +66,7 @@ import com.m57.hermescontrol.ui.common.BotAvatar
import com.m57.hermescontrol.ui.common.EmptyState
import com.m57.hermescontrol.ui.common.HermesScaffold
import com.m57.hermescontrol.ui.common.NavIcon
import kotlin.math.roundToInt
@Composable
fun GroupChatScreen(
@@ -74,6 +81,7 @@ fun GroupChatScreen(
mutableStateOf(TextFieldValue(""))
}
var isFocused by remember { mutableStateOf(false) }
var showSettingsDialog by remember { mutableStateOf(false) }
val handleSend = {
val text = inputFieldValue.text
@@ -111,6 +119,17 @@ fun GroupChatScreen(
}
},
navigationIcon = NavIcon.Back(onBack),
actions = {
IconButton(
onClick = { showSettingsDialog = true },
modifier = Modifier.testTag("group_chat_settings_button"),
) {
Icon(
imageVector = Icons.Filled.Settings,
contentDescription = stringResource(R.string.group_chat_action_settings),
)
}
},
modifier = modifier,
) {
Column(
@@ -276,6 +295,185 @@ fun GroupChatScreen(
}
}
}
if (showSettingsDialog) {
GroupChatSettingsDialog(
currentMaxMessages = state.maxBotMessages,
currentMaxPasses = state.maxContinuationPasses,
currentSystemPrompt = state.systemPrompt,
onDismiss = { showSettingsDialog = false },
onSave = { maxMsgs, maxPasses, prompt ->
viewModel.updateGroupLimits(maxMsgs, maxPasses, prompt)
showSettingsDialog = false
},
)
}
}
@Composable
private fun GroupChatSettingsDialog(
currentMaxMessages: Int,
currentMaxPasses: Int,
currentSystemPrompt: String?,
onDismiss: () -> Unit,
onSave: (Int, Int, String?) -> Unit,
) {
var maxMessages by remember(currentMaxMessages) { mutableStateOf(currentMaxMessages.toFloat()) }
var maxPasses by remember(currentMaxPasses) { mutableStateOf(currentMaxPasses.toFloat()) }
var systemPrompt by remember(currentSystemPrompt) { mutableStateOf(currentSystemPrompt.orEmpty()) }
AlertDialog(
onDismissRequest = onDismiss,
title = {
Text(
text = stringResource(R.string.group_chat_settings_title),
style = MaterialTheme.typography.titleMedium,
fontWeight = FontWeight.Bold,
)
},
text = {
Column(
modifier = Modifier.fillMaxWidth(),
verticalArrangement = Arrangement.spacedBy(16.dp),
) {
// Max Bot Messages Slider
Column {
Row(
modifier = Modifier.fillMaxWidth(),
horizontalArrangement = Arrangement.SpaceBetween,
verticalAlignment = Alignment.CenterVertically,
) {
Text(
text = stringResource(R.string.group_chat_settings_max_messages),
style = MaterialTheme.typography.bodyMedium,
fontWeight = FontWeight.SemiBold,
)
Text(
text = "${maxMessages.roundToInt()}",
style = MaterialTheme.typography.labelLarge,
color = MaterialTheme.colorScheme.primary,
fontWeight = FontWeight.Bold,
)
}
Text(
text =
stringResource(
R.string.group_chat_settings_max_messages_desc,
maxMessages.roundToInt(),
),
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
Slider(
value = maxMessages,
onValueChange = { maxMessages = it },
valueRange = 1f..20f,
steps = 18,
modifier = Modifier.testTag("max_messages_slider"),
)
}
// Max Continuation Handoffs Slider
Column {
Row(
modifier = Modifier.fillMaxWidth(),
horizontalArrangement = Arrangement.SpaceBetween,
verticalAlignment = Alignment.CenterVertically,
) {
Text(
text = stringResource(R.string.group_chat_settings_max_handoffs),
style = MaterialTheme.typography.bodyMedium,
fontWeight = FontWeight.SemiBold,
)
Text(
text = "${maxPasses.roundToInt()}",
style = MaterialTheme.typography.labelLarge,
color = MaterialTheme.colorScheme.primary,
fontWeight = FontWeight.Bold,
)
}
Text(
text =
stringResource(
R.string.group_chat_settings_max_handoffs_desc,
maxPasses.roundToInt(),
),
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
Slider(
value = maxPasses,
onValueChange = { maxPasses = it },
valueRange = 0f..6f,
steps = 5,
modifier = Modifier.testTag("max_handoffs_slider"),
)
}
// Room Instructions / System Prompt
Column {
Text(
text = stringResource(R.string.group_chat_settings_system_prompt),
style = MaterialTheme.typography.bodyMedium,
fontWeight = FontWeight.SemiBold,
)
Text(
text = stringResource(R.string.group_chat_settings_system_prompt_desc),
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
Spacer(modifier = Modifier.height(6.dp))
OutlinedTextField(
value = systemPrompt,
onValueChange = { systemPrompt = it },
placeholder = {
Text(
text = stringResource(R.string.group_chat_settings_system_prompt_hint),
style = MaterialTheme.typography.bodySmall,
)
},
modifier =
Modifier
.fillMaxWidth()
.heightIn(min = 80.dp, max = 160.dp)
.testTag("room_system_prompt_input"),
textStyle = MaterialTheme.typography.bodyMedium,
maxLines = 4,
)
}
}
},
confirmButton = {
Button(
onClick = {
onSave(
maxMessages.roundToInt(),
maxPasses.roundToInt(),
systemPrompt.trim().ifBlank { null },
)
},
modifier = Modifier.testTag("save_settings_button"),
) {
Text(stringResource(R.string.group_chat_settings_save))
}
},
dismissButton = {
Row {
TextButton(
onClick = {
maxMessages = DEFAULT_MAX_BOT_MESSAGES.toFloat()
maxPasses = DEFAULT_MAX_CONTINUATION_PASSES.toFloat()
systemPrompt = ""
},
) {
Text(stringResource(R.string.group_chat_settings_reset_default))
}
TextButton(onClick = onDismiss) {
Text(stringResource(R.string.common_cancel))
}
}
},
)
}
@Composable
@@ -283,7 +481,25 @@ private fun GroupMessageCard(
message: GroupChatMessage,
modifier: Modifier = Modifier,
) {
if (message.isUser) {
if (message.isSystem) {
Box(
modifier = modifier.fillMaxWidth().padding(vertical = 4.dp),
contentAlignment = Alignment.Center,
) {
Surface(
shape = RoundedCornerShape(12.dp),
color = MaterialTheme.colorScheme.surfaceVariant.copy(alpha = 0.6f),
border = BorderStroke(1.dp, MaterialTheme.colorScheme.outlineVariant.copy(alpha = 0.5f)),
) {
Text(
text = message.text,
style = MaterialTheme.typography.labelSmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
modifier = Modifier.padding(horizontal = 12.dp, vertical = 6.dp),
)
}
}
} else if (message.isUser) {
Row(
modifier = modifier.fillMaxWidth(),
horizontalArrangement = Arrangement.End,
@@ -3,6 +3,7 @@ package com.m57.hermescontrol.ui.bots.group
import android.util.Log
import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import com.m57.hermescontrol.data.model.GroupChatRoomLease
import com.m57.hermescontrol.data.model.GroupChatRoomMeta
import com.m57.hermescontrol.data.model.GroupChatSyncFrom
import com.m57.hermescontrol.data.model.GroupChatSyncLogEntry
@@ -39,6 +40,9 @@ data class GroupChatUiState(
val activeSpeaker: String? = null,
val isLoading: Boolean = true,
val errorMessage: String? = null,
val maxBotMessages: Int = DEFAULT_MAX_BOT_MESSAGES,
val maxContinuationPasses: Int = DEFAULT_MAX_CONTINUATION_PASSES,
val systemPrompt: String? = null,
)
data class MemberSession(
@@ -46,6 +50,13 @@ data class MemberSession(
val storedSessionId: String? = null,
)
const val DEFAULT_MAX_BOT_MESSAGES = 6
const val DEFAULT_MAX_CONTINUATION_PASSES = 2
private const val LEASE_STEP_TTL_MS = 45_000L
private const val TURN_TIMEOUT_MS = 45_000L
private const val HISTORY_LIMIT = 10
private const val CAPPED_SYSTEM_TEXT = "Discussion paused (turn limit reached) — reply to continue"
class GroupChatViewModel(
private var groupName: String = "",
private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
@@ -54,6 +65,12 @@ class GroupChatViewModel(
private val _uiState = MutableStateFlow(GroupChatUiState(groupName = groupName))
val uiState: StateFlow<GroupChatUiState> = _uiState.asStateFlow()
private val localDriverId = "android_" + UUID.randomUUID().toString().take(8)
private var activeEpoch: String = ""
// Map of bot.name -> seen count of non-system messages
private val memberWatermarks = mutableMapOf<String, Int>()
// Map of bot.name -> active session info in this group
private val memberSessions = mutableMapOf<String, MemberSession>()
@@ -80,6 +97,8 @@ class GroupChatViewModel(
fun setGroup(name: String) {
if (name == groupName && _uiState.value.groupName.isNotBlank()) return
groupName = name
activeEpoch = UUID.randomUUID().toString()
memberWatermarks.clear()
memberSessions.clear()
inFlightTurns.clear()
activeStreamMsgId.clear()
@@ -252,10 +271,22 @@ class GroupChatViewModel(
}
}
val nonSystemCount = initialMessages.filter { !it.isSystem }.size
for (member in groupMembers) {
memberWatermarks[member.name] = nonSystemCount
}
val maxMessages = matchingRoom?.maxBotMessages ?: DEFAULT_MAX_BOT_MESSAGES
val maxPasses = matchingRoom?.maxContinuationPasses ?: DEFAULT_MAX_CONTINUATION_PASSES
val roomSystemPrompt = matchingRoom?.systemPrompt
_uiState.update {
it.copy(
members = groupMembers,
messages = initialMessages,
maxBotMessages = maxMessages,
maxContinuationPasses = maxPasses,
systemPrompt = roomSystemPrompt,
isLoading = false,
)
}
@@ -270,10 +301,93 @@ class GroupChatViewModel(
}
}
private suspend fun ensureMemberSession(bot: ProfileInfo): MemberSession? {
val cached = memberSessions[bot.name]
if (cached != null) {
return cached
private fun findRoomEntry(rooms: Map<String, GroupChatRoomMeta>?): Pair<String, GroupChatRoomMeta?> {
if (rooms == null) return ("name:$groupName" to null)
val exact =
rooms.entries.find {
it.key == groupName || it.key == "name:$groupName" || it.key == "id:$groupName"
}
if (exact != null) return (exact.key to exact.value)
val byName = rooms.entries.find { it.value.name.equals(groupName, ignoreCase = true) }
if (byName != null) return (byName.key to byName.value)
return ("name:$groupName" to null)
}
fun updateGroupLimits(
maxMessages: Int,
maxPasses: Int,
systemPrompt: String? = null,
) {
val clampedMessages = maxMessages.coerceIn(1, 20)
val clampedPasses = maxPasses.coerceIn(0, 10)
val cleanPrompt = systemPrompt?.trim()?.ifBlank { null }
_uiState.update {
it.copy(
maxBotMessages = clampedMessages,
maxContinuationPasses = clampedPasses,
systemPrompt = cleanPrompt,
)
}
viewModelScope.launch(ioDispatcher) {
try {
val allProfiles = fetchProfiles()
val defaultProfile =
allProfiles.find { it.is_default == true || it.name == "default" } ?: return@launch
val existingSnapshot =
defaultProfile.groupChatSyncSnapshot()
?: GroupChatSyncSnapshot(version = 3, rooms = emptyMap())
val now = System.currentTimeMillis()
val (targetKey, existingRoom) = findRoomEntry(existingSnapshot.rooms)
val updatedRoom =
(existingRoom ?: GroupChatRoomMeta(name = groupName, createdAt = now)).copy(
name = groupName,
members = _uiState.value.members.map { JsonPrimitive(it.name) },
maxBotMessages = clampedMessages,
maxContinuationPasses = clampedPasses,
systemPrompt = cleanPrompt,
updatedAt = now,
)
val updatedRooms = existingSnapshot.rooms.orEmpty().toMutableMap()
updatedRooms[targetKey] = updatedRoom
val newSnapshot =
existingSnapshot.copy(
version = 3,
updatedAt = now,
rooms = updatedRooms,
)
val uiMetaPayload = mapOf("hermes-bots-groups" to newSnapshot.toMap())
HermesWsClient.request(
WsMethods.PROFILES_CONFIGURE,
mapOf(
"name" to defaultProfile.name,
"ui_meta" to uiMetaPayload,
),
).await()
} catch (e: Exception) {
Log.w("GroupChatViewModel", "updateGroupLimits failed: ${e.message}")
}
}
}
private suspend fun ensureMemberSession(
bot: ProfileInfo,
forceRefresh: Boolean = false,
): MemberSession? {
if (!forceRefresh) {
val cached = memberSessions[bot.name]
if (cached != null) {
return cached
}
} else {
memberSessions.remove(bot.name)
}
val title = "Group: $groupName"
@@ -298,6 +412,258 @@ class GroupChatViewModel(
return null
}
private suspend fun acquireOrVerifyLease(
defaultProfile: ProfileInfo,
epoch: String,
): Boolean {
try {
val syncSnapshot = defaultProfile.groupChatSyncSnapshot()
val (targetKey, existingRoom) = findRoomEntry(syncSnapshot?.rooms)
val currentLease = existingRoom?.lease
val now = System.currentTimeMillis()
if (currentLease?.driverId != null &&
currentLease.driverId != localDriverId &&
(currentLease.expiresAt ?: 0L) > now
) {
val remainingSec = ((currentLease.expiresAt ?: 0L) - now) / 1000
Log.i(
"GroupChatViewModel",
"Active lease held by ${currentLease.driverId} (expires in ${remainingSec}s). Passive mode.",
)
return false
}
val myLease =
GroupChatRoomLease(
driverId = localDriverId,
epoch = epoch,
expiresAt = now + LEASE_STEP_TTL_MS,
)
val existingSnapshot = syncSnapshot ?: GroupChatSyncSnapshot(version = 3, rooms = emptyMap())
val updatedRoom =
(existingRoom ?: GroupChatRoomMeta(name = groupName, createdAt = now)).copy(
name = groupName,
members = _uiState.value.members.map { JsonPrimitive(it.name) },
lease = myLease,
maxBotMessages = _uiState.value.maxBotMessages,
maxContinuationPasses = _uiState.value.maxContinuationPasses,
systemPrompt = _uiState.value.systemPrompt,
updatedAt = now,
)
val updatedRooms = existingSnapshot.rooms.orEmpty().toMutableMap()
updatedRooms[targetKey] = updatedRoom
val newSnapshot = existingSnapshot.copy(version = 3, updatedAt = now, rooms = updatedRooms)
val uiMetaPayload = mapOf("hermes-bots-groups" to newSnapshot.toMap())
HermesWsClient.request(
WsMethods.PROFILES_CONFIGURE,
mapOf(
"name" to defaultProfile.name,
"ui_meta" to uiMetaPayload,
),
).await()
return true
} catch (e: Exception) {
Log.w("GroupChatViewModel", "acquireOrVerifyLease failed: ${e.message}")
return true
}
}
private suspend fun executeBotTurn(
bot: ProfileInfo,
epoch: String,
members: List<ProfileInfo>,
): String? {
if (activeEpoch != epoch) return null
val currentNonSystemMessages = _uiState.value.messages.filter { !it.isSystem }
val seen = memberWatermarks[bot.name] ?: 0
val delta = currentNonSystemMessages.drop(seen)
if (delta.isEmpty() && currentNonSystemMessages.isNotEmpty()) {
return null
}
_uiState.update { it.copy(activeSpeaker = bot.effectiveTitle) }
val prompt =
GroupChatMentions.buildTurnPrompt(
groupName = groupName,
viewer = bot,
peers = members.filter { it.name != bot.name },
recentLog = currentNonSystemMessages,
historyLimit = HISTORY_LIMIT,
customRoomPrompt = _uiState.value.systemPrompt,
)
val turnStartTime = System.currentTimeMillis()
val streamMsgId = UUID.randomUUID().toString()
var activeRuntimeId: String? = null
var activeStoredId: String? = null
try {
var sessionInfo = ensureMemberSession(bot)
if (sessionInfo == null) {
Log.w("GroupChatViewModel", "Could not obtain session for ${bot.name}")
return null
}
var runtimeId = sessionInfo.runtimeSessionId
var storedId = sessionInfo.storedSessionId
activeRuntimeId = runtimeId
activeStoredId = storedId
activeStreamMsgId[runtimeId] = streamMsgId
storedId?.let { activeStreamMsgId[it] = streamMsgId }
sessionToBot[runtimeId] = bot
storedId?.let { sessionToBot[it] = bot }
ActiveSessionHolder.set(runtimeId, runtimeId)
val turnDeferred = CompletableDeferred<String>()
inFlightTurns[runtimeId] = turnDeferred
storedId?.let { inFlightTurns[it] = turnDeferred }
try {
HermesWsClient.request(
WsMethods.PROMPT_SUBMIT,
mapOf("session_id" to runtimeId, "text" to prompt),
).await()
} catch (submitErr: Exception) {
Log.w(
"GroupChatViewModel",
"prompt.submit failed for ${bot.name} on $runtimeId: ${submitErr.message}, recreating session",
)
activeStreamMsgId.remove(runtimeId)
storedId?.let { activeStreamMsgId.remove(it) }
sessionToBot.remove(runtimeId)
storedId?.let { sessionToBot.remove(it) }
inFlightTurns.remove(runtimeId)
storedId?.let { inFlightTurns.remove(it) }
sessionInfo = ensureMemberSession(bot, forceRefresh = true)
if (sessionInfo == null) return null
runtimeId = sessionInfo.runtimeSessionId
storedId = sessionInfo.storedSessionId
activeRuntimeId = runtimeId
activeStoredId = storedId
activeStreamMsgId[runtimeId] = streamMsgId
storedId?.let { activeStreamMsgId[it] = streamMsgId }
sessionToBot[runtimeId] = bot
storedId?.let { sessionToBot[it] = bot }
ActiveSessionHolder.set(runtimeId, runtimeId)
inFlightTurns[runtimeId] = turnDeferred
storedId?.let { inFlightTurns[it] = turnDeferred }
try {
HermesWsClient.request(
WsMethods.PROMPT_SUBMIT,
mapOf("session_id" to runtimeId, "text" to prompt),
).await()
} catch (retryErr: Exception) {
Log.e("GroupChatViewModel", "Retry prompt.submit failed for ${bot.name}: ${retryErr.message}")
return null
}
}
var replyText =
withTimeoutOrNull(TURN_TIMEOUT_MS) {
turnDeferred.await()
}
inFlightTurns.remove(runtimeId)
storedId?.let { inFlightTurns.remove(it) }
if (replyText == null) {
val targetIds = listOfNotNull(storedId, runtimeId).distinct()
for (tid in targetIds) {
try {
val res =
safeApiCall {
ApiClient.hermesApi.getSessionMessages(
sessionId = tid,
limit = 10,
order = "latest",
)
}
if (res is NetworkResult.Success) {
val lastAssistant =
res.data.messages.reversed().firstOrNull { msg ->
val ts = msg.timestampEpochMs
msg.role == "assistant" &&
(ts == null || ts >= (turnStartTime - 5000L))
}
val candidate = lastAssistant?.contentText?.trim()
if (!candidate.isNullOrBlank()) {
replyText = candidate
break
}
}
} catch (e: Exception) {
Log.w("GroupChatViewModel", "REST fallback failed for $tid: ${e.message}")
}
}
}
if (activeEpoch != epoch) {
_uiState.update { state ->
state.copy(messages = state.messages.filter { it.id != streamMsgId })
}
return null
}
if (replyText != null && !isPass(replyText) && replyText.isNotBlank()) {
val finalMsg =
GroupChatMessage(
id = streamMsgId,
senderName = bot.name,
senderDisplayName = bot.effectiveTitle,
isUser = false,
avatarMeta = bot.botMeta()?.avatar,
text = replyText.trim(),
isStreaming = false,
)
_uiState.update { state ->
val existing = state.messages.find { it.id == streamMsgId }
val updatedList =
if (existing != null) {
state.messages.map { if (it.id == streamMsgId) finalMsg else it }
} else {
state.messages + finalMsg
}
state.copy(messages = updatedList)
}
return replyText.trim()
} else {
_uiState.update { state ->
state.copy(messages = state.messages.filter { it.id != streamMsgId })
}
return null
}
} catch (e: Exception) {
Log.w("GroupChatViewModel", "Turn failed for ${bot.name}: ${e.message}")
_uiState.update { state ->
state.copy(messages = state.messages.filter { it.id != streamMsgId })
}
return null
} finally {
activeRuntimeId?.let {
activeStreamMsgId.remove(it)
sessionToBot.remove(it)
inFlightTurns.remove(it)
}
activeStoredId?.let {
activeStreamMsgId.remove(it)
sessionToBot.remove(it)
inFlightTurns.remove(it)
}
}
}
fun sendMessage(rawText: String) {
val text = rawText.trim()
Log.d("GroupChatViewModel", "sendMessage called with: '$text'")
@@ -309,6 +675,9 @@ class GroupChatViewModel(
return
}
val epoch = UUID.randomUUID().toString()
activeEpoch = epoch
val userMessage =
GroupChatMessage(
id = UUID.randomUUID().toString(),
@@ -321,142 +690,152 @@ class GroupChatViewModel(
_uiState.update {
it.copy(messages = it.messages + userMessage)
}
persistSyncSnapshot(_uiState.value.messages)
persistSyncSnapshot(_uiState.value.messages, extendLease = true)
viewModelScope.launch(ioDispatcher) {
val responders = GroupChatMentions.resolveResponders(text, members)
Log.d("GroupChatViewModel", "Responders: ${responders.map { it.name }}")
val allProfiles = fetchProfiles()
val defaultProfile =
allProfiles.find { it.is_default == true || it.name == "default" }
?: allProfiles.firstOrNull()
for (bot in responders) {
_uiState.update { it.copy(activeSpeaker = bot.effectiveTitle) }
val prompt =
GroupChatMentions.buildTurnPrompt(
groupName = groupName,
viewer = bot,
peers = members.filter { it.name != bot.name },
recentLog = _uiState.value.messages,
)
if (defaultProfile != null) {
val hasLease = acquireOrVerifyLease(defaultProfile, epoch)
if (!hasLease) {
Log.i("GroupChatViewModel", "Passive mode active — skipping local turn drive")
return@launch
}
}
val streamMsgId = UUID.randomUUID().toString()
var activeRuntimeId: String? = null
var activeStoredId: String? = null
try {
val sessionInfo = ensureMemberSession(bot)
if (sessionInfo == null) {
Log.w("GroupChatViewModel", "Could not obtain session for ${bot.name}")
continue
}
val runtimeId = sessionInfo.runtimeSessionId
val storedId = sessionInfo.storedSessionId
activeRuntimeId = runtimeId
activeStoredId = storedId
var postedBotMessages = 0
var continuationsCount = 0
activeStreamMsgId[runtimeId] = streamMsgId
storedId?.let { activeStreamMsgId[it] = streamMsgId }
val maxBotLimit = _uiState.value.maxBotMessages
val maxPassLimit = _uiState.value.maxContinuationPasses
sessionToBot[runtimeId] = bot
storedId?.let { sessionToBot[it] = bot }
val initialResponders = GroupChatMentions.resolveResponders(text, members).toMutableList()
val remainingInitial = initialResponders.toMutableList()
// Set ActiveSessionHolder so the active WebSocket context matches this turn
ActiveSessionHolder.set(runtimeId, runtimeId)
// ── Round 1: Initial Turn ──
while (remainingInitial.isNotEmpty()) {
if (activeEpoch != epoch) return@launch
if (postedBotMessages >= maxBotLimit) break
// Register in-flight listener
val turnDeferred = CompletableDeferred<String>()
inFlightTurns[runtimeId] = turnDeferred
storedId?.let { inFlightTurns[it] = turnDeferred }
val bot = remainingInitial.removeAt(0)
val reply = executeBotTurn(bot, epoch, members)
val nonSystemSize = _uiState.value.messages.filter { !it.isSystem }.size
memberWatermarks[bot.name] = nonSystemSize
// Submit prompt via sendMessage
HermesWsClient.sendMessage(
sessionId = runtimeId,
text = prompt,
)
if (reply != null) {
postedBotMessages++
persistSyncSnapshot(_uiState.value.messages, extendLease = true)
}
}
// Await event or polling fallback
var replyText =
withTimeoutOrNull(45_000L) {
turnDeferred.await()
}
inFlightTurns.remove(runtimeId)
storedId?.let { inFlightTurns.remove(it) }
// ── Reactive Continuation Passes ──
while (continuationsCount < maxPassLimit && postedBotMessages < maxBotLimit) {
if (activeEpoch != epoch) return@launch
// Polling fallback if event dropped or session was reaped
if (replyText == null) {
val targetIds = listOfNotNull(storedId, runtimeId).distinct()
for (tid in targetIds) {
try {
val res =
safeApiCall {
ApiClient.hermesApi.getSessionMessages(
sessionId = tid,
limit = 10,
order = "latest",
)
}
if (res is NetworkResult.Success) {
val lastAssistant =
res.data.messages.reversed().firstOrNull { it.role == "assistant" }
val candidate = lastAssistant?.contentText?.trim()
if (!candidate.isNullOrBlank()) {
replyText = candidate
break
}
}
} catch (e: Exception) {
Log.w("GroupChatViewModel", "REST fallback failed for $tid: ${e.message}")
val currentLog = _uiState.value.messages.filter { !it.isSystem }
val userIndex = currentLog.indexOfLast { it.isUser }
val recentAssistantMsgs =
if (userIndex >= 0) currentLog.drop(userIndex + 1) else currentLog
val pendingCitedBots = mutableListOf<ProfileInfo>()
for (msg in recentAssistantMsgs) {
val parsed =
GroupChatMentions.parseMentions(
text = msg.text,
members = members,
excludedSpeaker = msg.senderName,
)
for (citedName in parsed.mentionedBots) {
val bot = members.find { it.name.equals(citedName, ignoreCase = true) }
if (bot != null && !pendingCitedBots.any { it.name == bot.name }) {
val seenCount = memberWatermarks[bot.name] ?: 0
if (seenCount < currentLog.size) {
pendingCitedBots.add(bot)
}
}
}
}
if (replyText != null && !isPass(replyText) && replyText.isNotBlank()) {
_uiState.update { state ->
val existing = state.messages.find { it.id == streamMsgId }
val finalMsg =
GroupChatMessage(
id = streamMsgId,
senderName = bot.name,
senderDisplayName = bot.effectiveTitle,
isUser = false,
avatarMeta = bot.botMeta()?.avatar,
text = replyText.trim(),
isStreaming = false,
)
val updatedList =
if (existing != null) {
state.messages.map { if (it.id == streamMsgId) finalMsg else it }
} else {
state.messages + finalMsg
}
state.copy(messages = updatedList)
}
persistSyncSnapshot(_uiState.value.messages)
} else {
_uiState.update { state ->
state.copy(messages = state.messages.filter { it.id != streamMsgId })
}
}
} catch (e: Exception) {
Log.w("GroupChatViewModel", "Turn failed for ${bot.name}: ${e.message}")
_uiState.update { state ->
state.copy(messages = state.messages.filter { it.id != streamMsgId })
}
} finally {
activeRuntimeId?.let {
activeStreamMsgId.remove(it)
sessionToBot.remove(it)
inFlightTurns.remove(it)
}
activeStoredId?.let {
activeStreamMsgId.remove(it)
sessionToBot.remove(it)
inFlightTurns.remove(it)
if (pendingCitedBots.isEmpty()) {
// Quiescence / Natural Settle reached
break
}
continuationsCount++
val continuationResponders = pendingCitedBots.toMutableList()
while (continuationResponders.isNotEmpty()) {
if (activeEpoch != epoch) return@launch
if (postedBotMessages >= maxBotLimit) break
val bot = continuationResponders.removeAt(0)
val reply = executeBotTurn(bot, epoch, members)
val nonSystemSize = _uiState.value.messages.filter { !it.isSystem }.size
memberWatermarks[bot.name] = nonSystemSize
if (reply != null) {
postedBotMessages++
persistSyncSnapshot(_uiState.value.messages, extendLease = true)
}
}
}
// ── Capped Exit Banner Check ──
val isBudgetExceeded =
postedBotMessages >= maxBotLimit || continuationsCount >= maxPassLimit
val hasUnfinishedWork = remainingInitial.isNotEmpty() || hasPendingMentions(members)
if (isBudgetExceeded && hasUnfinishedWork && activeEpoch == epoch) {
val cappedMessage =
GroupChatMessage(
id = UUID.randomUUID().toString(),
senderName = "system",
senderDisplayName = "System",
isUser = false,
isSystem = true,
text = CAPPED_SYSTEM_TEXT,
)
_uiState.update { it.copy(messages = it.messages + cappedMessage) }
persistSyncSnapshot(_uiState.value.messages, extendLease = false)
}
_uiState.update { it.copy(activeSpeaker = null) }
}
}
private fun persistSyncSnapshot(currentMessages: List<GroupChatMessage>) {
private fun hasPendingMentions(members: List<ProfileInfo>): Boolean {
val currentLog = _uiState.value.messages.filter { !it.isSystem }
val userIndex = currentLog.indexOfLast { it.isUser }
val recentAssistantMsgs =
if (userIndex >= 0) currentLog.drop(userIndex + 1) else currentLog
for (msg in recentAssistantMsgs) {
val parsed =
GroupChatMentions.parseMentions(
text = msg.text,
members = members,
excludedSpeaker = msg.senderName,
)
for (citedName in parsed.mentionedBots) {
val bot = members.find { it.name.equals(citedName, ignoreCase = true) }
if (bot != null) {
val seenCount = memberWatermarks[bot.name] ?: 0
if (seenCount < currentLog.size) {
return true
}
}
}
}
return false
}
private fun persistSyncSnapshot(
currentMessages: List<GroupChatMessage>,
extendLease: Boolean = true,
) {
viewModelScope.launch(ioDispatcher) {
try {
val allProfiles = fetchProfiles()
@@ -467,7 +846,7 @@ class GroupChatViewModel(
?: GroupChatSyncSnapshot(version = 3, rooms = emptyMap())
val logEntries =
currentMessages.takeLast(32).map { msg ->
currentMessages.filter { !it.isSystem }.takeLast(32).map { msg ->
GroupChatSyncLogEntry(
id = msg.id,
from =
@@ -481,25 +860,38 @@ class GroupChatViewModel(
)
}
val roomKey = "name:$groupName"
val existingRoom =
existingSnapshot.rooms?.get(roomKey)
?: existingSnapshot.rooms?.get(groupName)
?: existingSnapshot.rooms?.values?.find { it.name.equals(groupName, ignoreCase = true) }
val now = System.currentTimeMillis()
val (targetKey, existingRoom) = findRoomEntry(existingSnapshot.rooms)
val updatedLease =
if (extendLease) {
GroupChatRoomLease(
driverId = localDriverId,
epoch = activeEpoch,
expiresAt = now + LEASE_STEP_TTL_MS,
)
} else {
existingRoom?.lease
}
val updatedRoom =
(existingRoom ?: GroupChatRoomMeta(name = groupName, createdAt = System.currentTimeMillis())).copy(
(existingRoom ?: GroupChatRoomMeta(name = groupName, createdAt = now)).copy(
name = groupName,
members = _uiState.value.members.map { JsonPrimitive(it.name) },
log = logEntries,
updatedAt = System.currentTimeMillis(),
lease = updatedLease,
maxBotMessages = _uiState.value.maxBotMessages,
maxContinuationPasses = _uiState.value.maxContinuationPasses,
systemPrompt = _uiState.value.systemPrompt,
updatedAt = now,
)
val updatedRooms = (existingSnapshot.rooms.orEmpty() + (roomKey to updatedRoom))
val updatedRooms = existingSnapshot.rooms.orEmpty().toMutableMap()
updatedRooms[targetKey] = updatedRoom
val newSnapshot =
existingSnapshot.copy(
version = 3,
updatedAt = System.currentTimeMillis(),
updatedAt = now,
rooms = updatedRooms,
)
@@ -520,7 +912,7 @@ class GroupChatViewModel(
private fun isPass(text: String): Boolean {
val clean = text.trim().lowercase()
return clean == "(pass)" || clean == "pass"
return clean == "(pass)" || clean == "pass" || clean == "(pass)." || clean == "pass."
}
private fun extractSessionInfo(response: Any?): MemberSession? {