fix(chat): fix auto-scroll during streaming

This commit is contained in:
Ochenstarik 2026-08-25 01:06:54 +07:00
parent ab1e5cba68
commit 4041480e60
15 changed files with 946 additions and 24 deletions

View file

@ -0,0 +1,8 @@
## Замеры до/после (Coder 2)
| Метрика | До (base) | После (Coder 1) | После фикса (Coder 2) |
| --- | --- | --- | --- |
| Число SQL-запросов (список 100 сессий) | ~100+ | 1 | 1 |
| Удерживаемая память (после 50 сессий) | Утечка (растёт) | Стабильна (ограничение + очистка) | Стабильна |
| Перезапуск эффекта прокрутки (10 сек стрима) | Перебивает ручную прокрутку | Регресс: автопрокрутка не работает при стриме | Исправлено: автопрокрутка плавно следует за стримом |

View file

@ -0,0 +1,15 @@
## Кодер 2 (review + доработка)
**Отчёт по анти-чеклисту:**
1. `sessions` flow uses single-query projection — **проверено — чисто** (переход на `getUnifiedSessionsSummaryFlow`).
2. Single-query projection does not perform N individual subqueries — **проверено — чисто** (подзапросы в SELECT компилируются в один SQL statement, Room видит 1 запрос, тест `SessionListQueryCountTest` проходит).
3. Memory release clears maps — **проверено — чисто**`releaseSession`).
4. Session host mutexes are cleaned up — **проверено — чисто** (очищаются в `releaseSession`).
5. `toolToMessageMap` is cleaned up upon message completion — **проверено — чисто**.
6. Strict `messageId` matching handles fallback gracefully — **проверено — чисто** (добавлен безопасный fallback при пустом id).
7. Auto-scroll follows active streaming smoothly — **нарушено — удаление параметра lastMessageLength из `LaunchedEffect` привело к тому, что автопрокрутка не реагировала на изменение длины текста при потоковой передаче**.
- *Исправление*: В `ChatScreen.kt` восстановлено отслеживание длины текста последнего сообщения (`messages.lastOrNull()?.content?.length`) внутри `snapshotFlow`, чтобы прокрутка возобновлялась во время стриминга.
8. Before/after measurements are verified on the same dataset/scenario — **проверено — чисто**.
9. Verification commands actually executed with exit codes captured — **проверено — чисто**.
**Вердикт:** LOW severity finding (Регресс с автопрокруткой, исправлено). Все тесты и команды верификации зелёные. Изменения минимальны и закрывают задачу.

View file

@ -90,7 +90,9 @@ data class UnifiedSession(
val createdAt: Long = System.currentTimeMillis(),
val updatedAt: Long = System.currentTimeMillis(),
val bindings: Map<HermesHostId, HostSessionBinding> = emptyMap(),
val timeline: List<UnifiedMessage> = emptyList()
val timeline: List<UnifiedMessage> = emptyList(),
val messageCount: Int = timeline.size,
val lastMessagePreview: String? = null
)
@Serializable

View file

@ -46,15 +46,31 @@ class UnifiedSessionRepository(
) {
private val json = Json { ignoreUnknownKeys = true }
val sessions: StateFlow<List<UnifiedSession>> = sessionDao.getSessionsFlow()
companion object {
const val MAX_CACHED_SESSIONS = 10
}
val sessions: StateFlow<List<UnifiedSession>> = sessionDao.getUnifiedSessionsSummaryFlow()
.map { list ->
list.map { entity ->
val details = sessionDao.getSessionWithDetails(entity.id)
details?.toDomain() ?: entity.toDomainPlaceholder()
list.map { summary ->
UnifiedSession(
id = UnifiedSessionId(summary.id),
title = summary.title,
activeHostId = HermesHostId(summary.activeHostId),
createdAt = summary.createdAt,
updatedAt = summary.updatedAt,
bindings = emptyMap(),
timeline = emptyList(),
messageCount = summary.messageCount,
lastMessagePreview = summary.lastMessagePreview
)
}
}
.stateIn(scope, SharingStarted.Eagerly, emptyList())
// Scoped tool to message attribution mapping (toolId -> messageId)
private val toolToMessageMap = ConcurrentHashMap<String, String>()
// Per-session approval requests state
private val sessionApprovalsState = ConcurrentHashMap<UnifiedSessionId, MutableStateFlow<List<HostAttributedApproval>>>()
@ -126,7 +142,40 @@ class UnifiedSessionRepository(
return sessionHostMutexes.computeIfAbsent(Pair(sessionId, hostId)) { Mutex() }
}
private fun pruneIdleSessionCaches() {
if (sessionMessagesState.size > MAX_CACHED_SESSIONS) {
val idleSessionIds = sessionMessagesState.keys.filter { sid ->
val isExec = sessionExecutingState[sid]?.value ?: false
!isExec
}
for (sid in idleSessionIds) {
if (sessionMessagesState.size <= MAX_CACHED_SESSIONS) break
releaseSession(sid)
}
}
}
fun releaseSession(sessionId: UnifiedSessionId) {
val executing = sessionExecutingState[sessionId]?.value ?: false
if (executing) return
val messages = sessionMessagesState.remove(sessionId)?.value ?: emptyList()
for (m in messages) {
for (t in m.tools) {
toolToMessageMap.remove(t.id)
}
}
sessionExecutingState.remove(sessionId)
sessionApprovalsState.remove(sessionId)
sessionClarifyQueueState.remove(sessionId)
sessionActiveClarifyFlows.remove(sessionId)
hostExecutingState.entries.removeIf { it.key.first == sessionId }
sessionHostMutexes.entries.removeIf { it.key.first == sessionId }
_hasActiveTasks.update { hostExecutingState.values.any { it.value } }
}
fun getSessionMessages(sessionId: UnifiedSessionId): StateFlow<List<UnifiedMessage>> {
pruneIdleSessionCaches()
return sessionMessagesState.computeIfAbsent(sessionId) {
val flow = MutableStateFlow<List<UnifiedMessage>>(emptyList())
scope.launch {
@ -174,6 +223,7 @@ class UnifiedSessionRepository(
title: String = "New Session",
initialHostId: HermesHostId? = null
): UnifiedSession {
pruneIdleSessionCaches()
val hostId = initialHostId ?: connectionManager.activeHostId.value
?: connectionManager.hosts.value.firstOrNull()?.id
?: throw IllegalStateException("No Hermes hosts configured. Please add a host before creating a session.")
@ -211,7 +261,12 @@ class UnifiedSessionRepository(
suspend fun deleteUnifiedSession(sessionId: UnifiedSessionId) {
sessionDao.deleteSession(sessionId.value)
sessionMessagesState.remove(sessionId)
val msgs = sessionMessagesState.remove(sessionId)?.value ?: emptyList()
for (m in msgs) {
for (t in m.tools) {
toolToMessageMap.remove(t.id)
}
}
sessionExecutingState.remove(sessionId)
sessionApprovalsState.remove(sessionId)
sessionClarifyQueueState.remove(sessionId)
@ -771,6 +826,8 @@ class UnifiedSessionRepository(
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
setHostExecuting(sessionId, hostId, false)
// Clean up tool mappings associated with this message upon completion
toolToMessageMap.entries.removeIf { it.value == event.messageId }
scope.launch {
sessionDao.updateBindingState(sessionId.value, hostId.value, BindingState.READY.name)
}
@ -800,7 +857,12 @@ class UnifiedSessionRepository(
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId }
val targetAssistant = if (event.messageId.isNotBlank()) {
flow.value.find { it.id == event.messageId && (it.hostId == hostId || it.hostId == null) }
} else {
// Fallback for events without messageId: bind to last assistant for this host
flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId }
}
if (targetAssistant != null) {
updateMessageInSession(sessionId, targetAssistant.id, immediate = false) {
it.copy(thinking = (it.thinking ?: "") + event.delta)
@ -812,7 +874,12 @@ class UnifiedSessionRepository(
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId }
val targetAssistant = if (event.messageId.isNotBlank()) {
flow.value.find { it.id == event.messageId && (it.hostId == hostId || it.hostId == null) }
} else {
// Fallback for events without messageId: bind to last assistant for this host
flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId }
}
if (targetAssistant != null) {
updateMessageInSession(sessionId, targetAssistant.id, immediate = false) {
it.copy(thinking = (it.thinking ?: "") + event.delta)
@ -824,7 +891,12 @@ class UnifiedSessionRepository(
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId }
val targetAssistant = if (event.messageId.isNotBlank()) {
flow.value.find { it.id == event.messageId && (it.hostId == hostId || it.hostId == null) }
} else {
// Fallback for events without messageId: bind to last assistant for this host
flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId }
}
if (targetAssistant != null) {
updateMessageInSession(sessionId, targetAssistant.id, immediate = true) {
it.copy(thinking = event.reasoning)
@ -834,9 +906,12 @@ class UnifiedSessionRepository(
is GatewayEvent.ToolStartEvent -> {
if (event.toolId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId) ?: return
val explicitMessageId = (event.rawPayload["payload"] as? kotlinx.serialization.json.JsonObject)?.get("message_id")?.let {
if (it is kotlinx.serialization.json.JsonPrimitive) it.content else null
}
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = explicitMessageId) ?: return
val tool = ToolActivity(id = event.toolId, name = event.name, status = "running")
attachToolToSessionMessage(sessionId, hostId, tool)
attachToolToSessionMessage(sessionId, hostId, tool, explicitMessageId = explicitMessageId)
}
is GatewayEvent.ToolProgressEvent -> {
@ -980,17 +1055,32 @@ class UnifiedSessionRepository(
}
}
private fun attachToolToSessionMessage(sessionId: UnifiedSessionId, hostId: HermesHostId, tool: ToolActivity) {
private fun attachToolToSessionMessage(
sessionId: UnifiedSessionId,
hostId: HermesHostId,
tool: ToolActivity,
explicitMessageId: String? = null
) {
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val lastAssistant = flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId }
if (lastAssistant != null) {
updateMessageInSession(sessionId, lastAssistant.id, immediate = true) {
val targetMsg = if (!explicitMessageId.isNullOrBlank()) {
flow.value.find { it.id == explicitMessageId && (it.hostId == hostId || it.hostId == null) }
} else {
// Strict attribution to the currently streaming assistant message for this host, or the last assistant message for this host
flow.value.lastOrNull { it.isStreaming && it.role == MessageRole.ASSISTANT && it.hostId == hostId }
?: flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId }
}
if (targetMsg != null) {
toolToMessageMap[tool.id] = targetMsg.id
updateMessageInSession(sessionId, targetMsg.id, immediate = true) {
val updatedTools = it.tools.filterNot { t -> t.id == tool.id } + tool
it.copy(tools = updatedTools)
}
} else {
val newId = explicitMessageId?.ifBlank { null } ?: UUID.randomUUID().toString()
toolToMessageMap[tool.id] = newId
val newMsg = UnifiedMessage(
id = UUID.randomUUID().toString(),
id = newId,
role = MessageRole.ASSISTANT,
content = "",
hostId = hostId,
@ -1007,7 +1097,13 @@ class UnifiedSessionRepository(
transform: (ToolActivity) -> ToolActivity
) {
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val targetMsg = flow.value.lastOrNull { msg -> msg.tools.any { it.id == toolId } }
val boundMessageId = toolToMessageMap[toolId]
val targetMsg = if (boundMessageId != null) {
flow.value.find { it.id == boundMessageId }
} else {
flow.value.lastOrNull { msg -> msg.tools.any { it.id == toolId } }
}
if (targetMsg != null) {
updateMessageInSession(sessionId, targetMsg.id, immediate = true) { msg ->
val updatedTools = msg.tools.map { if (it.id == toolId) transform(it) else it }

View file

@ -35,6 +35,36 @@ interface UnifiedSessionDao {
@Query("SELECT * FROM unified_sessions ORDER BY updatedAt DESC")
suspend fun getSessions(): List<UnifiedSessionEntity>
@Query("""
SELECT
s.id AS id,
s.title AS title,
s.activeHostId AS activeHostId,
s.createdAt AS createdAt,
s.updatedAt AS updatedAt,
(SELECT COUNT(*) FROM unified_messages WHERE sessionId = s.id) AS messageCount,
(SELECT COUNT(*) FROM host_bindings WHERE sessionId = s.id) AS bindingCount,
(SELECT content FROM unified_messages WHERE sessionId = s.id ORDER BY createdAt DESC, id DESC LIMIT 1) AS lastMessagePreview
FROM unified_sessions s
ORDER BY s.updatedAt DESC
""")
fun getUnifiedSessionsSummaryFlow(): Flow<List<UnifiedSessionSummaryProjection>>
@Query("""
SELECT
s.id AS id,
s.title AS title,
s.activeHostId AS activeHostId,
s.createdAt AS createdAt,
s.updatedAt AS updatedAt,
(SELECT COUNT(*) FROM unified_messages WHERE sessionId = s.id) AS messageCount,
(SELECT COUNT(*) FROM host_bindings WHERE sessionId = s.id) AS bindingCount,
(SELECT content FROM unified_messages WHERE sessionId = s.id ORDER BY createdAt DESC, id DESC LIMIT 1) AS lastMessagePreview
FROM unified_sessions s
ORDER BY s.updatedAt DESC
""")
suspend fun getUnifiedSessionsSummary(): List<UnifiedSessionSummaryProjection>
@Query("SELECT * FROM unified_sessions WHERE id = :sessionId LIMIT 1")
suspend fun getSession(sessionId: String): UnifiedSessionEntity?

View file

@ -90,3 +90,14 @@ data class UnifiedSessionWithDetails(
val bindings: List<HostBindingEntity> = emptyList(),
val messages: List<UnifiedMessageEntity> = emptyList()
)
data class UnifiedSessionSummaryProjection(
val id: String,
val title: String,
val activeHostId: String,
val createdAt: Long = 0L,
val updatedAt: Long = 0L,
val messageCount: Int = 0,
val bindingCount: Int = 0,
val lastMessagePreview: String? = null
)

View file

@ -55,6 +55,8 @@ import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
import androidx.compose.runtime.setValue
import androidx.compose.runtime.snapshotFlow
import kotlinx.coroutines.flow.conflate
import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier
import androidx.compose.ui.draw.clip
@ -111,10 +113,28 @@ fun ChatScreen(
}
}
LaunchedEffect(messages.size, messages.lastOrNull()?.content?.length, approvals.size) {
if (messages.isNotEmpty() || approvals.isNotEmpty()) {
val totalCount = messages.size + approvals.size
listState.animateScrollToItem(totalCount)
LaunchedEffect(listState) {
snapshotFlow {
val totalItems = messages.size + approvals.size
val isStreaming = isExecuting || messages.any { it.isStreaming }
val isAtBottom = !listState.canScrollForward || (listState.layoutInfo.visibleItemsInfo.lastOrNull()?.index ?: -1) >= (listState.layoutInfo.totalItemsCount - 2)
val lastMessageLength = messages.lastOrNull()?.content?.length ?: 0
listOf(totalItems, isStreaming, isAtBottom, lastMessageLength)
}
.conflate()
.collect { stateList ->
val totalItems = stateList[0] as Int
val isStreaming = stateList[1] as Boolean
val isAtBottom = stateList[2] as Boolean
if (totalItems > 0 && isAtBottom) {
val targetIndex = maxOf(0, totalItems - 1)
if (isStreaming) {
listState.scrollToItem(targetIndex)
} else {
listState.animateScrollToItem(targetIndex)
}
}
}
}

View file

@ -143,4 +143,9 @@ class ChatViewModel(
sessionRepo.dismissClarify(hostId, req.requestId, req.promptType, req.questionId)
}
}
override fun onCleared() {
super.onCleared()
sessionRepo.releaseSession(sessionId)
}
}

View file

@ -285,8 +285,9 @@ fun UnifiedSessionCard(
// Attached hosts counter
val attachedCount = session.bindings.size.coerceAtLeast(1)
val msgCount = if (session.messageCount > 0) session.messageCount else session.timeline.size
Text(
text = "$attachedCount attached • ${session.timeline.size} msgs",
text = "$attachedCount attached • $msgCount msgs",
style = MaterialTheme.typography.labelSmall,
color = MaterialTheme.colorScheme.onSurfaceVariant
)

View file

@ -0,0 +1,182 @@
package app.hermes.mobile.core.repository
import app.hermes.mobile.core.model.*
import app.hermes.mobile.core.network.JsonRpcGatewayClient
import app.hermes.mobile.core.runtime.HermesConnectionManager
import app.hermes.mobile.core.runtime.HermesHostRuntime
import app.hermes.mobile.core.security.InMemoryTokenVault
import app.hermes.mobile.core.storage.FakeHostDao
import app.hermes.mobile.core.storage.FakeUnifiedSessionDao
import app.hermes.mobile.core.storage.HostBindingEntity
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Test
import java.util.concurrent.ConcurrentHashMap
/**
* Verifies that opening, streaming in, and closing/unsubscribing from 50 sessions
* does not leak memory in repository caches (sessionMessagesState, hostExecutingState,
* sessionExecutingState, sessionHostMutexes, toolIdToMessageId).
*/
class CacheEvictionTest {
@Suppress("UNCHECKED_CAST")
private fun getInternalMapSize(repository: UnifiedSessionRepository, fieldName: String): Int {
return try {
val field = UnifiedSessionRepository::class.java.getDeclaredField(fieldName)
field.isAccessible = true
val map = field.get(repository) as? Map<*, *>
map?.size ?: 0
} catch (_: NoSuchFieldException) {
0
}
}
@Test
fun test50SessionsCycleEvictsAndBoundsMemoryCaches() = runBlocking {
val hostId = HermesHostId("host-cache-1")
val host = HermesHost(id = hostId, displayName = "Cache Host", baseUrl = "http://cache-host:9119")
val hostDao = FakeHostDao()
val sessionDao = FakeUnifiedSessionDao()
val tokenVault = InMemoryTokenVault()
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
val connectionManager = HermesConnectionManager(
hostDao = hostDao,
tokenVault = tokenVault,
scope = scope,
runtimeFactory = { parentScope, h ->
val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default)
HermesHostRuntime(
initialHost = h,
gatewayClient = JsonRpcGatewayClient(scope = childScope),
tokenVault = tokenVault,
scope = childScope
)
}
)
val repository = UnifiedSessionRepository(
connectionManager = connectionManager,
sessionDao = sessionDao,
scope = scope
)
connectionManager.addHost(host)
delay(50)
val runtime = connectionManager.getRuntime(hostId)!!
val sessionCount = 50
// Cycle through 50 sessions
for (i in 1..sessionCount) {
val session = repository.createUnifiedSession(title = "Session $i", initialHostId = hostId)
val rtSessionId = "rt_session_$i"
val msgId = "msg_$i"
repository.registerRuntimeBinding(session.id, hostId, RuntimeSessionId(rtSessionId))
sessionDao.insertOrUpdateBinding(
HostBindingEntity(
sessionId = session.id.value,
hostId = hostId.value,
durableSessionId = "dur_$i",
runtimeSessionId = rtSessionId,
state = BindingState.RUNNING.name
)
)
// Simulate subscription to session flows
val messagesFlow = repository.getSessionMessages(session.id)
val execFlow = repository.getSessionExecuting(session.id)
val hostExecFlow = repository.getHostExecuting(session.id, hostId)
val job = launch {
messagesFlow.collect {}
}
// Stream a message with tool
val startJson = buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.start")
put("session_id", rtSessionId)
put("payload", buildJsonObject {
put("message_id", msgId)
put("role", "assistant")
})
})
}.toString()
runtime.gatewayClient.handleIncomingMessage(startJson)
val toolStartJson = buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "tool.start")
put("session_id", rtSessionId)
put("payload", buildJsonObject {
put("tool_id", "tool_$i")
put("name", "test_tool")
})
})
}.toString()
runtime.gatewayClient.handleIncomingMessage(toolStartJson)
val completeJson = buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.complete")
put("session_id", rtSessionId)
put("payload", buildJsonObject {
put("message_id", msgId)
put("content", "Done $i")
})
})
}.toString()
runtime.gatewayClient.handleIncomingMessage(completeJson)
delay(20)
// Unsubscribe
job.cancel()
// Release session explicitly if method exists
try {
val releaseMethod = repository.javaClass.getMethod("releaseSession", UnifiedSessionId::class.java)
releaseMethod.invoke(repository, session.id)
} catch (_: NoSuchMethodException) {
// Base SHA does not have releaseSession
}
}
delay(100)
val messagesCacheSize = getInternalMapSize(repository, "sessionMessagesState")
val sessionExecSize = getInternalMapSize(repository, "sessionExecutingState")
val hostExecSize = getInternalMapSize(repository, "hostExecutingState")
val mutexesSize = getInternalMapSize(repository, "sessionHostMutexes")
val toolMapSize = getInternalMapSize(repository, "toolToMessageMap").coerceAtLeast(
getInternalMapSize(repository, "toolIdToMessageId")
)
println("BASELINE MEASUREMENT [Cache sizes after 50 sessions]: messagesCache=$messagesCacheSize, sessionExec=$sessionExecSize, hostExec=$hostExecSize, mutexes=$mutexesSize, toolMap=$toolMapSize")
// Assert caches are pruned / bounded (bounded to at most small LRU size e.g. <= 5 or 0 when released)
assertTrue("sessionMessagesState ($messagesCacheSize) must be bounded <= 5", messagesCacheSize <= 5)
assertTrue("sessionExecutingState ($sessionExecSize) must be bounded <= 5", sessionExecSize <= 5)
assertTrue("hostExecutingState ($hostExecSize) must be bounded <= 5", hostExecSize <= 5)
assertTrue("sessionHostMutexes ($mutexesSize) must be bounded <= 5", mutexesSize <= 5)
assertTrue("toolToMessageMap ($toolMapSize) must be pruned upon completion", toolMapSize <= 5)
}
}

View file

@ -0,0 +1,146 @@
package app.hermes.mobile.core.repository
import app.hermes.mobile.core.model.HermesHost
import app.hermes.mobile.core.model.HermesHostId
import app.hermes.mobile.core.runtime.HermesConnectionManager
import app.hermes.mobile.core.security.InMemoryTokenVault
import app.hermes.mobile.core.storage.FakeHostDao
import app.hermes.mobile.core.storage.FakeUnifiedSessionDao
import app.hermes.mobile.core.storage.HostBindingEntity
import app.hermes.mobile.core.storage.UnifiedMessageEntity
import app.hermes.mobile.core.storage.UnifiedSessionDao
import app.hermes.mobile.core.storage.UnifiedSessionEntity
import app.hermes.mobile.core.storage.UnifiedSessionWithDetails
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.runBlocking
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Test
import java.util.concurrent.atomic.AtomicInteger
/**
* Verifies that loading the list of 100 sessions does not perform N+1 database queries.
* On base SHA, collecting sessions flow causes 100+ queries (1 getSessionsFlow + 100 getSessionWithDetails).
*/
class SessionListQueryCountTest {
private class TrackingUnifiedSessionDao(
private val delegate: FakeUnifiedSessionDao
) : UnifiedSessionDao by delegate {
val queryCount = AtomicInteger(0)
val detailsQueryCount = AtomicInteger(0)
override fun getSessionsFlow(): Flow<List<UnifiedSessionEntity>> {
queryCount.incrementAndGet()
return delegate.getSessionsFlow()
}
override suspend fun getSessions(): List<UnifiedSessionEntity> {
queryCount.incrementAndGet()
return delegate.getSessions()
}
override suspend fun getSession(sessionId: String): UnifiedSessionEntity? {
queryCount.incrementAndGet()
return delegate.getSession(sessionId)
}
override suspend fun getSessionWithDetails(sessionId: String): UnifiedSessionWithDetails? {
queryCount.incrementAndGet()
detailsQueryCount.incrementAndGet()
return delegate.getSessionWithDetails(sessionId)
}
override fun getUnifiedSessionsSummaryFlow(): Flow<List<app.hermes.mobile.core.storage.UnifiedSessionSummaryProjection>> {
queryCount.incrementAndGet()
return delegate.getUnifiedSessionsSummaryFlow()
}
override suspend fun getUnifiedSessionsSummary(): List<app.hermes.mobile.core.storage.UnifiedSessionSummaryProjection> {
queryCount.incrementAndGet()
return delegate.getUnifiedSessionsSummary()
}
override suspend fun getMessagesForSession(sessionId: String): List<UnifiedMessageEntity> {
queryCount.incrementAndGet()
return delegate.getMessagesForSession(sessionId)
}
override suspend fun getBindingsForSession(sessionId: String): List<HostBindingEntity> {
queryCount.incrementAndGet()
return delegate.getBindingsForSession(sessionId)
}
}
@Test
fun test100SessionsListQueryCountIsFixedAndNotNPlusOne() = runBlocking {
val hostDao = FakeHostDao()
val rawSessionDao = FakeUnifiedSessionDao()
val trackingDao = TrackingUnifiedSessionDao(rawSessionDao)
val tokenVault = InMemoryTokenVault()
val scope = CoroutineScope(Dispatchers.Default)
val hostId = HermesHostId("host-perf-1")
val host = HermesHost(id = hostId, displayName = "Host Perf", baseUrl = "http://host-perf:9119")
val connectionManager = HermesConnectionManager(
hostDao = hostDao,
tokenVault = tokenVault,
scope = scope
)
connectionManager.addHost(host)
val sessionCount = 100
val messagesPerSession = 20
// Populate 100 sessions with 20 messages each
for (i in 1..sessionCount) {
val sid = "session-$i"
rawSessionDao.insertSession(
UnifiedSessionEntity(
id = sid,
title = "Session $i",
activeHostId = hostId.value,
createdAt = 1000L + i,
updatedAt = 1000L + i
)
)
val msgs = (1..messagesPerSession).map { mIdx ->
UnifiedMessageEntity(
id = "msg-$i-$mIdx",
sessionId = sid,
role = if (mIdx % 2 == 0) "ASSISTANT" else "USER",
content = "Content $i - $mIdx",
hostId = hostId.value,
createdAt = 1000L + i * 100 + mIdx
)
}
rawSessionDao.insertMessages(msgs)
}
// Reset tracking counters before initializing repository & collecting sessions
trackingDao.queryCount.set(0)
trackingDao.detailsQueryCount.set(0)
val repository = UnifiedSessionRepository(
connectionManager = connectionManager,
sessionDao = trackingDao,
scope = scope
)
// Read the sessions flow
val sessionsList = repository.sessions.first { it.size == sessionCount }
assertEquals(sessionCount, sessionsList.size)
val totalQueries = trackingDao.queryCount.get()
val detailsQueries = trackingDao.detailsQueryCount.get()
println("BASELINE MEASUREMENT [100 Sessions List]: totalQueries=$totalQueries, detailsQueries=$detailsQueries")
// In a lightweight projection, detailsQueries must be 0 and totalQueries must be <= 2 (fixed count, O(1))
assertEquals("Should not invoke getSessionWithDetails in N+1 loop", 0, detailsQueries)
assertTrue("Total queries ($totalQueries) must be <= 2, not O(N)", totalQueries <= 2)
}
}

View file

@ -0,0 +1,385 @@
package app.hermes.mobile.core.repository
import app.hermes.mobile.core.model.*
import app.hermes.mobile.core.network.JsonRpcGatewayClient
import app.hermes.mobile.core.runtime.HermesConnectionManager
import app.hermes.mobile.core.runtime.HermesHostRuntime
import app.hermes.mobile.core.security.InMemoryTokenVault
import app.hermes.mobile.core.storage.FakeHostDao
import app.hermes.mobile.core.storage.FakeUnifiedSessionDao
import app.hermes.mobile.core.storage.HostBindingEntity
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.runBlocking
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Test
/**
* Verifies that when 2 hosts stream concurrently into the same session with tool events
* and thinking deltas, tools and thinking are strictly bound to their respective host and
* explicit messageId, avoiding false attribution.
*/
class ToolAttributionTest {
@Test
fun testTwoConcurrentHostsAttributionAndThinkingIsolation() = runBlocking {
val hostAId = HermesHostId("host-a")
val hostBId = HermesHostId("host-b")
val hostA = HermesHost(id = hostAId, displayName = "Host A", baseUrl = "http://host-a:9119")
val hostB = HermesHost(id = hostBId, displayName = "Host B", baseUrl = "http://host-b:9119")
val hostDao = FakeHostDao()
val sessionDao = FakeUnifiedSessionDao()
val tokenVault = InMemoryTokenVault()
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
val connectionManager = HermesConnectionManager(
hostDao = hostDao,
tokenVault = tokenVault,
scope = scope,
runtimeFactory = { parentScope, h ->
val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default)
HermesHostRuntime(
initialHost = h,
gatewayClient = JsonRpcGatewayClient(scope = childScope),
tokenVault = tokenVault,
scope = childScope
)
}
)
val repository = UnifiedSessionRepository(
connectionManager = connectionManager,
sessionDao = sessionDao,
scope = scope
)
connectionManager.addHost(hostA)
connectionManager.addHost(hostB)
delay(50)
val session = repository.createUnifiedSession(title = "Dual Host Attribution Test", initialHostId = hostAId)
val rtSessionA = "rt_session_host_a"
val rtSessionB = "rt_session_host_b"
repository.registerRuntimeBinding(session.id, hostAId, RuntimeSessionId(rtSessionA))
repository.registerRuntimeBinding(session.id, hostBId, RuntimeSessionId(rtSessionB))
sessionDao.insertOrUpdateBinding(
HostBindingEntity(
sessionId = session.id.value,
hostId = hostAId.value,
durableSessionId = "dur_a",
runtimeSessionId = rtSessionA,
state = BindingState.RUNNING.name
)
)
sessionDao.insertOrUpdateBinding(
HostBindingEntity(
sessionId = session.id.value,
hostId = hostBId.value,
durableSessionId = "dur_b",
runtimeSessionId = rtSessionB,
state = BindingState.RUNNING.name
)
)
val runtimeA = connectionManager.getRuntime(hostAId)!!
val runtimeB = connectionManager.getRuntime(hostBId)!!
val msgAId = "msg_host_a_1"
val msgBId = "msg_host_b_1"
// 1. Host A starts message
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.start")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("message_id", msgAId)
put("role", "assistant")
})
})
}.toString())
// 2. Host B starts message
runtimeB.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.start")
put("session_id", rtSessionB)
put("payload", buildJsonObject {
put("message_id", msgBId)
put("role", "assistant")
})
})
}.toString())
delay(30)
// 3. Host A streams thinking delta for msgA
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "thinking.delta")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("message_id", msgAId)
put("delta", "Plan on Host A")
})
})
}.toString())
// 4. Host B streams thinking delta for msgB
runtimeB.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "thinking.delta")
put("session_id", rtSessionB)
put("payload", buildJsonObject {
put("message_id", msgBId)
put("delta", "Plan on Host B")
})
})
}.toString())
// 5. Host A starts tool
val toolAId = "tool_host_a_1"
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "tool.start")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("tool_id", toolAId)
put("name", "bash_executor")
})
})
}.toString())
// 6. Host B starts tool
val toolBId = "tool_host_b_1"
runtimeB.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "tool.start")
put("session_id", rtSessionB)
put("payload", buildJsonObject {
put("tool_id", toolBId)
put("name", "file_editor")
})
})
}.toString())
delay(30)
// 7. Update tool progress and completion for host A
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "tool.complete")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("tool_id", toolAId)
put("result", "Output from Host A")
put("is_error", false)
})
})
}.toString())
// 8. Update tool progress and completion for host B
runtimeB.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "tool.complete")
put("session_id", rtSessionB)
put("payload", buildJsonObject {
put("tool_id", toolBId)
put("result", "Output from Host B")
put("is_error", false)
})
})
}.toString())
// 9. Complete messages
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.complete")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("message_id", msgAId)
put("content", "Final content A")
})
})
}.toString())
runtimeB.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.complete")
put("session_id", rtSessionB)
put("payload", buildJsonObject {
put("message_id", msgBId)
put("content", "Final content B")
})
})
}.toString())
delay(50)
val messages = repository.getSessionMessages(session.id).value
val msgA = messages.find { it.id == msgAId }
val msgB = messages.find { it.id == msgBId }
assertNotNull("Message A must exist", msgA)
assertNotNull("Message B must exist", msgB)
assertEquals("Host A attribution", hostAId, msgA?.hostId)
assertEquals("Host B attribution", hostBId, msgB?.hostId)
assertEquals("Thinking for Message A", "Plan on Host A", msgA?.thinking)
assertEquals("Thinking for Message B", "Plan on Host B", msgB?.thinking)
assertEquals("Tools count for Message A", 1, msgA?.tools?.size)
assertEquals("Tool ID for Message A", toolAId, msgA?.tools?.firstOrNull()?.id)
assertEquals("Tool result for Message A", "Output from Host A", msgA?.tools?.firstOrNull()?.result)
assertEquals("Tools count for Message B", 1, msgB?.tools?.size)
assertEquals("Tool ID for Message B", toolBId, msgB?.tools?.firstOrNull()?.id)
assertEquals("Tool result for Message B", "Output from Host B", msgB?.tools?.firstOrNull()?.result)
}
@Test
fun testThinkingDeltaWithExplicitMessageIdDoesNotFallBackToLastAssistant() = runBlocking {
val hostAId = HermesHostId("host-a")
val hostA = HermesHost(id = hostAId, displayName = "Host A", baseUrl = "http://host-a:9119")
val hostDao = FakeHostDao()
val sessionDao = FakeUnifiedSessionDao()
val tokenVault = InMemoryTokenVault()
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
val connectionManager = HermesConnectionManager(
hostDao = hostDao,
tokenVault = tokenVault,
scope = scope,
runtimeFactory = { parentScope, h ->
val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default)
HermesHostRuntime(
initialHost = h,
gatewayClient = JsonRpcGatewayClient(scope = childScope),
tokenVault = tokenVault,
scope = childScope
)
}
)
val repository = UnifiedSessionRepository(
connectionManager = connectionManager,
sessionDao = sessionDao,
scope = scope
)
connectionManager.addHost(hostA)
delay(50)
val session = repository.createUnifiedSession(title = "Message Targeting Test", initialHostId = hostAId)
val rtSessionA = "rt_session_host_a"
repository.registerRuntimeBinding(session.id, hostAId, RuntimeSessionId(rtSessionA))
sessionDao.insertOrUpdateBinding(
HostBindingEntity(
sessionId = session.id.value,
hostId = hostAId.value,
durableSessionId = "dur_a",
runtimeSessionId = rtSessionA,
state = BindingState.RUNNING.name
)
)
val runtimeA = connectionManager.getRuntime(hostAId)!!
val msg1Id = "msg_first_1"
val msg2Id = "msg_second_2"
// Host A starts first assistant message
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.start")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("message_id", msg1Id)
put("role", "assistant")
})
})
}.toString())
// Host A starts second assistant message
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.start")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("message_id", msg2Id)
put("role", "assistant")
})
})
}.toString())
delay(30)
// Send thinking delta explicitly targeted at msg1Id
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "thinking.delta")
put("session_id", rtSessionA)
put("payload", buildJsonObject {
put("message_id", msg1Id)
put("delta", "Thinking targeted exclusively at msg1")
})
})
}.toString())
delay(30)
val messages = repository.getSessionMessages(session.id).value
val msg1 = messages.find { it.id == msg1Id }
val msg2 = messages.find { it.id == msg2Id }
assertNotNull("msg1 must exist", msg1)
assertNotNull("msg2 must exist", msg2)
// On base SHA: targetAssistant is lastOrNull { (it.id == msg1Id || it.role == ASSISTANT) }
// Because msg2 has role == ASSISTANT and is last, it matches msg2!
// So msg1 thinking will be null and msg2 thinking will have the text on base SHA!
assertEquals("msg1 must receive thinking targeted at it", "Thinking targeted exclusively at msg1", msg1?.thinking)
assertEquals("msg2 must NOT receive thinking intended for msg1", null, msg2?.thinking)
}
}

View file

@ -88,6 +88,28 @@ class FakeUnifiedSessionDao : UnifiedSessionDao {
sessions.values.sortedByDescending { it.updatedAt }.map { it.copy() }
}
override fun getUnifiedSessionsSummaryFlow(): Flow<List<UnifiedSessionSummaryProjection>> =
_sessionsFlow.map { getUnifiedSessionsSummary() }
override suspend fun getUnifiedSessionsSummary(): List<UnifiedSessionSummaryProjection> = synchronized(lock) {
sessions.values.sortedByDescending { it.updatedAt }.map { s ->
val msgList = messages[s.id]
val lastMsg = msgList?.sortedWith(messageComparator)?.lastOrNull()?.content
val msgCount = msgList?.size ?: 0
val bindCount = bindings[s.id]?.size ?: 0
UnifiedSessionSummaryProjection(
id = s.id,
title = s.title,
activeHostId = s.activeHostId,
createdAt = s.createdAt,
updatedAt = s.updatedAt,
messageCount = msgCount,
bindingCount = bindCount,
lastMessagePreview = lastMsg
)
}
}
override suspend fun getSession(sessionId: String): UnifiedSessionEntity? = synchronized(lock) {
sessions[sessionId]?.copy()
}

View file

@ -86,7 +86,6 @@ class MessageOrderingTest {
assertEquals("Repository getUnifiedSession must order messages by createdAt ASC, id ASC", expected, timelineIds)
val sessionFromList = repository.sessions.value.find { it.id == session.id }
val sessionListTimelineIds = sessionFromList?.timeline?.map { it.id }
assertEquals("Repository sessions flow must order messages by createdAt ASC, id ASC", expected, sessionListTimelineIds)
assertEquals("Repository sessions flow projection must report message count", 3, sessionFromList?.messageCount)
}
}