From 4041480e609144f7c3ae469b7a7a3e2e5f2ee7b7 Mon Sep 17 00:00:00 2001 From: Ochenstarik Date: Tue, 25 Aug 2026 01:06:54 +0700 Subject: [PATCH] fix(chat): fix auto-scroll during streaming --- ...26-08-24-09-performance-and-attribution.md | Bin 0 -> 11748 bytes agents/antigravity/done/temp_measure.md | 8 + agents/antigravity/done/temp_report.md | 15 + .../mobile/core/model/MultiHostModels.kt | 4 +- .../repository/UnifiedSessionRepository.kt | 128 +++++- .../app/hermes/mobile/core/storage/Daos.kt | 30 ++ .../hermes/mobile/core/storage/Entities.kt | 11 + .../hermes/mobile/feature/chat/ChatScreen.kt | 28 +- .../mobile/feature/chat/ChatViewModel.kt | 5 + .../unified_sessions/UnifiedSessionsScreen.kt | 3 +- .../core/repository/CacheEvictionTest.kt | 182 +++++++++ .../repository/SessionListQueryCountTest.kt | 146 +++++++ .../core/repository/ToolAttributionTest.kt | 385 ++++++++++++++++++ .../hermes/mobile/core/storage/FakeDaos.kt | 22 + .../core/storage/MessageOrderingTest.kt | 3 +- 15 files changed, 946 insertions(+), 24 deletions(-) create mode 100644 agents/antigravity/done/TASK-2026-08-24-09-performance-and-attribution.md create mode 100644 agents/antigravity/done/temp_measure.md create mode 100644 agents/antigravity/done/temp_report.md create mode 100644 app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt create mode 100644 app/src/test/java/app/hermes/mobile/core/repository/SessionListQueryCountTest.kt create mode 100644 app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt diff --git a/agents/antigravity/done/TASK-2026-08-24-09-performance-and-attribution.md b/agents/antigravity/done/TASK-2026-08-24-09-performance-and-attribution.md new file mode 100644 index 0000000000000000000000000000000000000000..27b26175ae7d24b38cc80629d40212d537510fc7 GIT binary patch literal 11748 zcmd5>Yi}FZ5#0~`B%oh{04<85xrLCHlFgSa#j;{arYYGLqDO4XNr9lquE>?dnBp?K zOW6jD!m{Jqfs?o`TC^yFCP0Dqt1MfwEWiH3@*nk_xw~9Gq*!v`AP`9I-g|fM%$b=p zb65TSD)?9MJa`yvH?ON)UtfPeM!pTUgU#ScuoFDO+!n5%1<#sygU6~^a_V+pF0BR! z1x$V%Pk)hq873oR?X(-8I`wS+%n3RLUulx22 zW!9`pE%S-BZpnyVm|1g{AT3jJTs!0Wj%%&hnYPUa6fAxo+=pdbuyY&sZK~i!^Ew8% zgWcwzf}Q3)WnQdRmMeBSG^N;BTeIA&u`Q-b1~2f$6Rxe4&7;bMIaQk%7OY*j&pDny zBNHXpwtTys_c__9m*X4Q`HV4u%gj1ID zeg1n73uM`x@{_7C0Jw>|$rj%(X&K*2{n!P&g4JU~{pQ=c|$_o@n>BAjo)GY>Ii zuGoI3BfZd*XR6NSka-jZjtHT5nyk1^8;+^A{9Uo6GTUzOU&(&*==Tay4ov( zLG0S=inC>~Qv+Pt(2VzS6h2)67i$3k;G*Vr##{4tx4Yn19f|kh*UjKNe2|dJ?8YE? zq~oB)y1`DmuWu+WFGEpzIf05%uu7}8nqIF+B~iasbM_i?Pn`yJk}^AMODBFmu~e0W z>?*Nsw53efuG*GohuM=1{V{{gK`s*Uo#yAj|0{eWJD4hxnE1YjB;C=e6lKqjN=Dfp zRuZr1qXMxOg_5VQ*rkRK5W6Sxbrj)stJ?EKp=!C6<*Vl#cEj#@NUC4Bf;8)Kt5xLe zzWYc*H+=hwEo*(psV+L9wsTfJu6cj!EZiEkTOkQO3Z5#n?lx-nOt~7NCvoGE>6;r$ z+M1_;kM#kI=kmY*xR5WLo1Xf(I9)8xEX)_xxXNXTGbsw3DY3l8NbbNKIB!G70|{|p z4yUk@(2WuX#UV(O=9kSon0efyJTx01I%x32;A3*8_m6l` z!m*eVj)jzv%YqV$z6&-`i-;D5=1?uJR%(|J#VYS(W(g79-E#hkCBA24Za!oo9=OH_?7}WzCo>IJHv4bOZS|O6le)3D^ir1ZgM@==7Z?c`k-3^q}u;W11F<~TY9zhP%^a1+@K7d~&ALTO} zarLx$vsHRvSyW^H9q8)B?&Ag-6-c5Jj6!J2uKL#WI*=oGT>u+9H2}BIaDE-FjkQ#@ z^%JlPj%qt!uob{I{JMiV?s;33qap01z)RFw&K=jTsfwR=ovN+I$5G|vfN%k~={y*! zxR{?`6u2Q|1J8agym6x)JswM>joJe-)231B*wrf8{0!@#;Ew^etNTNuD{7GeNp;el zJ;%cP+z~^pG7Lnl66iUIP}vNLE_B;H#t#-*JoNNKNkIAz;>u)^F&4i426&&5)8sW(Z@!S*urVzvU;&)-5L<;|JZ-z8BAAuRB* z*aKUVI1n3hjzD1+ihy^#npO8!9iL5!=xsv&auv-GN|dx#2t+8qjiwc68sR85THx8P z<{s0?h%aSrNKmjH(aP{+6e5~f>35?P+(AMB6GZZ6L_z<>KHO^yPoh8IESdl)zrpxM zn!pNj+pa`;i`BveCBqSBq^G=`G1sgs3(LvuQRO?nRh{weH4im1*LH$s;<1z21}VZM zXh+G|pFF8OgG{9IEnOpPw2g5#W5ok`jOR+7Dxd=r&m&|c`)b&NZQ4; zCqcgGP*3JL`9eqlI{po(O?{Cp@L5STd0Y#o+YcM^YM)1UbfsaIoWp`wmE=!0bgJT-37(M z8Ff&Tj0q1)ts@I|z{3;tZPAV>-2+RcQYpTGQ^7wVmSEsf3A;?p#eIF(RPZ&0iI=z$ z2GA3$>=9T9*9wRu;;NN*b@aad}YX2j|<7%0BtoxDSQ=)V!EYt}8L;D2>NqXi7oT@8c7$ecSiYXrU^x{v6yKObHr+rwic;)@z>fS?>ir_NFMNFfM$5^JVihp87)~Mde;vGhiY~YC|&;X;s?vZl>DNkhWLxR`~rnH2prX)9;=kvZw!r%2HnZ z=#Ue7cL)i{BEWgLo3Sg5jC4ew1a!q!XqrM0&*W!j1*NB0aYqV}&V1}HzQE?_M%j9S zp>9eg>}~#NC^B0pL7AN`Yuu)SBKNQa)FBCQ9cKYhe0#&Eeeyme2>OyOIx<+Gq!6>R z%3JpcD-<35c0n4(9&skmBW(x5D3T+>1>=50`U$x(#17dnvBxw)S#I9#ODQv*S#d4A zS-6~DvV4_F)tpq>!ds}C;@j*gd#SO41Ew!KjFI1tWi$B5tf%Uh=hvaGH0YA+cZ4vCPtTyD(<*{Bb-;V zqGqgM#x_#Mrs73&E(0?mj_*(l<>UF#zBS;La-aikrPa&0bFrH;=JBD-GQR1xGW4*pJMn55OD&wJHRZetDn#Qru+T{6 z6Yl|)A$*F)v@wfM!B~XkF`#x5YevP}r{L>(cs!q22v<|eDDdZV=kXjh^r5Q{&DQX# z;p!+xFTgfOY#YLQKEt@mVWwRNt~?EYF{p@?#EiIu$eMPR;(WV)A z9X9FwX7pJ=Cql);H9hYGpQ_|lWG~s4tZ`0kh*-J+2FQkXjbetna$OdhjzD8xDM(^g z`dyJUi@oU!?xl=TAVUAmVlA0~zG5y+!-{2GQFD4yW5RmzF7P0)k!grbAwkwTD>$*n z4at?}kTKC+S4UP3YI+La*~Bp*tSb=hpticMP#@aBwc}XBCx)?t97%>HyN!ug_)P36 z*5C=uu%=CjU!%H3q*!lR&vXPU=&Rs0R>EIE7Nx{-!aWC#G0eV!9Zc{zW7Y?k(#N)- zmqw&&T)104)Q43xiZ)_Io$}(PccA|e_Be<|l*fo277UI-rZV#QW;D-WoHAHFlF+R) z>pa*O`FsJo&WHJvMBGkdjvT;xqN@r=$lOsZkWE>^#^F&$5Hr9;{M2AQIh<_j8*vSf zy`B@#?B#Ky!;H&8s>WtPyq!Rj+nJuZN4|(W6fYU zc33Xb7Mu_ZK7gc1i<0c4`@4vr$(Llum^F0YqB+4fSiefbI=Wt6hDJJ%Imc@)sf=}$ z!9D62)wr<2MZBWD^|Jm?>&{M%+4BbcbV_=UG5AUMtP0GrV0;M4#4AGC7Y-xVx(CzT zpEMLC920*tXBgM}GX}GmF~%ObGeak&BVm5#AusND$PbKK-E%S9hQqFmdXFZi@L`{R z2ESZ)98bTC4xP+0f=E+fgHHBgp6OU*Op>W)WwqXiofjGIJ;`lWe%*~}#wmocOZ4=q z;^$aTZsQaB{Ez{IMfuK*WpnNt(YKNtXg@v8m1J^uw^{r# zO4!lnF>?Z**L(jloaIT3da)CZf1`c-5dp+I;$QMiR+xtMlALo|n24vG*bgK7?%I#_ zBd8|Y14+~}b`q!Mlox6C8{G3$$c~@ULSH3O3mEsh0xS zZ;4&(XLtf(pF+gu5aUyrVSYySkyx^WpmaukQjpWA@)_={IhQ9`y>}#?RsbG zu$}c`9(Ub*U%(hW(3#_`qX+2;qH}b#7o}6%SUYAq<@Z)>Soo4%CgBTeq0hNdeJzA` N5ySSKbib#u{y#71cUJ%a literal 0 HcmV?d00001 diff --git a/agents/antigravity/done/temp_measure.md b/agents/antigravity/done/temp_measure.md new file mode 100644 index 0000000..0e02cb8 --- /dev/null +++ b/agents/antigravity/done/temp_measure.md @@ -0,0 +1,8 @@ + +## Замеры до/после (Coder 2) + +| Метрика | До (base) | После (Coder 1) | После фикса (Coder 2) | +| --- | --- | --- | --- | +| Число SQL-запросов (список 100 сессий) | ~100+ | 1 | 1 | +| Удерживаемая память (после 50 сессий) | Утечка (растёт) | Стабильна (ограничение + очистка) | Стабильна | +| Перезапуск эффекта прокрутки (10 сек стрима) | Перебивает ручную прокрутку | Регресс: автопрокрутка не работает при стриме | Исправлено: автопрокрутка плавно следует за стримом | diff --git a/agents/antigravity/done/temp_report.md b/agents/antigravity/done/temp_report.md new file mode 100644 index 0000000..4b6afaf --- /dev/null +++ b/agents/antigravity/done/temp_report.md @@ -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 (Регресс с автопрокруткой, исправлено). Все тесты и команды верификации зелёные. Изменения минимальны и закрывают задачу. diff --git a/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt b/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt index f72c137..04b29f1 100644 --- a/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt +++ b/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt @@ -90,7 +90,9 @@ data class UnifiedSession( val createdAt: Long = System.currentTimeMillis(), val updatedAt: Long = System.currentTimeMillis(), val bindings: Map = emptyMap(), - val timeline: List = emptyList() + val timeline: List = emptyList(), + val messageCount: Int = timeline.size, + val lastMessagePreview: String? = null ) @Serializable diff --git a/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt b/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt index 7c3a5b2..ee0e91e 100644 --- a/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt +++ b/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt @@ -46,15 +46,31 @@ class UnifiedSessionRepository( ) { private val json = Json { ignoreUnknownKeys = true } - val sessions: StateFlow> = sessionDao.getSessionsFlow() + companion object { + const val MAX_CACHED_SESSIONS = 10 + } + + val sessions: StateFlow> = 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() + // Per-session approval requests state private val sessionApprovalsState = ConcurrentHashMap>>() @@ -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> { + pruneIdleSessionCaches() return sessionMessagesState.computeIfAbsent(sessionId) { val flow = MutableStateFlow>(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 } diff --git a/app/src/main/java/app/hermes/mobile/core/storage/Daos.kt b/app/src/main/java/app/hermes/mobile/core/storage/Daos.kt index 6fb6bee..ba0f37b 100644 --- a/app/src/main/java/app/hermes/mobile/core/storage/Daos.kt +++ b/app/src/main/java/app/hermes/mobile/core/storage/Daos.kt @@ -35,6 +35,36 @@ interface UnifiedSessionDao { @Query("SELECT * FROM unified_sessions ORDER BY updatedAt DESC") suspend fun getSessions(): List + @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> + + @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 + @Query("SELECT * FROM unified_sessions WHERE id = :sessionId LIMIT 1") suspend fun getSession(sessionId: String): UnifiedSessionEntity? diff --git a/app/src/main/java/app/hermes/mobile/core/storage/Entities.kt b/app/src/main/java/app/hermes/mobile/core/storage/Entities.kt index 4ae507e..30c1d1d 100644 --- a/app/src/main/java/app/hermes/mobile/core/storage/Entities.kt +++ b/app/src/main/java/app/hermes/mobile/core/storage/Entities.kt @@ -90,3 +90,14 @@ data class UnifiedSessionWithDetails( val bindings: List = emptyList(), val messages: List = 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 +) diff --git a/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt b/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt index 8cb6409..32e6e99 100644 --- a/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt +++ b/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt @@ -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) + } + } } } diff --git a/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt b/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt index b9d1030..1a8e92b 100644 --- a/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt +++ b/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt @@ -143,4 +143,9 @@ class ChatViewModel( sessionRepo.dismissClarify(hostId, req.requestId, req.promptType, req.questionId) } } + + override fun onCleared() { + super.onCleared() + sessionRepo.releaseSession(sessionId) + } } diff --git a/app/src/main/java/app/hermes/mobile/feature/unified_sessions/UnifiedSessionsScreen.kt b/app/src/main/java/app/hermes/mobile/feature/unified_sessions/UnifiedSessionsScreen.kt index 3d98b56..ac87e94 100644 --- a/app/src/main/java/app/hermes/mobile/feature/unified_sessions/UnifiedSessionsScreen.kt +++ b/app/src/main/java/app/hermes/mobile/feature/unified_sessions/UnifiedSessionsScreen.kt @@ -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 ) diff --git a/app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt new file mode 100644 index 0000000..28c3eb2 --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt @@ -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) + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/repository/SessionListQueryCountTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/SessionListQueryCountTest.kt new file mode 100644 index 0000000..a6e08c6 --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/repository/SessionListQueryCountTest.kt @@ -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> { + queryCount.incrementAndGet() + return delegate.getSessionsFlow() + } + + override suspend fun getSessions(): List { + 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> { + queryCount.incrementAndGet() + return delegate.getUnifiedSessionsSummaryFlow() + } + + override suspend fun getUnifiedSessionsSummary(): List { + queryCount.incrementAndGet() + return delegate.getUnifiedSessionsSummary() + } + + override suspend fun getMessagesForSession(sessionId: String): List { + queryCount.incrementAndGet() + return delegate.getMessagesForSession(sessionId) + } + + override suspend fun getBindingsForSession(sessionId: String): List { + 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) + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt new file mode 100644 index 0000000..6e2621f --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt @@ -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) + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt b/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt index 6bc1073..68170b5 100644 --- a/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt +++ b/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt @@ -88,6 +88,28 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { sessions.values.sortedByDescending { it.updatedAt }.map { it.copy() } } + override fun getUnifiedSessionsSummaryFlow(): Flow> = + _sessionsFlow.map { getUnifiedSessionsSummary() } + + override suspend fun getUnifiedSessionsSummary(): List = 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() } diff --git a/app/src/test/java/app/hermes/mobile/core/storage/MessageOrderingTest.kt b/app/src/test/java/app/hermes/mobile/core/storage/MessageOrderingTest.kt index 12c0fe5..584ec50 100644 --- a/app/src/test/java/app/hermes/mobile/core/storage/MessageOrderingTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/storage/MessageOrderingTest.kt @@ -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) } }