diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt index 1a979cd6..78fbc763 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt @@ -68,12 +68,14 @@ object EventParser { } "reasoning.available" -> { - WsEvent.ReasoningAvailable(sessionId) + val text = payload?.get("text") as? String + WsEvent.ReasoningAvailable(sessionId, text) } "message.complete" -> { val text = payload?.get("text") as? String ?: "" - WsEvent.MessageComplete(text, sessionId) + val reasoning = payload?.get("reasoning") as? String + WsEvent.MessageComplete(text, sessionId, reasoning) } "message.done" -> { diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt index 19e86973..96d0368f 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt @@ -37,11 +37,26 @@ sealed class WsEvent { data class ReasoningAvailable( val sessionId: String?, + /** + * Full reasoning text, emitted once reasoning finishes streaming + * (`reasoning.available` payload `text`). The gateway sends the + * complete trace here even when per-token `reasoning.delta` events + * were throttled, dropped, or wiped by a mid-turn tool.start — the + * client must use it as the authoritative fill for the card. + */ + val text: String? = null, ) : WsEvent() data class MessageComplete( val text: String, val sessionId: String?, + /** + * Full reasoning trace for the completed message, carried in the + * `message.complete` payload (`reasoning` key). Authoritative + * fallback when per-token `reasoning.delta` events never arrived + * (e.g. a tool call wiped the streaming buffer mid-turn). + */ + val reasoning: String? = null, ) : WsEvent() data class MessageDone( diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt index 6b816da5..f1770373 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatStreamingController.kt @@ -55,6 +55,25 @@ class ChatStreamingController( streamingState.update { it.copy(isReasoning = false, reasoningText = "") } } + /** + * Clears ONLY the throttled token buffers (and their flush timers) — + * never touches `streamingMessage`, `isReasoning` or `reasoningText`. + * + * Used at tool.start: the reducer keeps the streaming message + its + * reasoning alive across the tool call (issue #771), and the VM must + * not undo that by resetting the shared streaming state. The buffers + * themselves are drained (flushPendingReasoning) before reduce, so a + * buffer-only clear is safe and prevents stale deltas from re-flushing. + */ + fun clearStreamingBuffers() { + streamingBuffer.clear() + thinkingBuffer.clear() + reasoningBuffer.clear() + lastFlushMs = 0L + lastThinkingFlushMs = 0L + lastReasoningFlushMs = 0L + } + /** Resets buffers and starts a fresh streaming message (called on MessageStart). */ fun beginStreamingMessage() { resetStreaming() diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt index 5eb96b38..9e10621a 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt @@ -58,6 +58,111 @@ import java.util.concurrent.ConcurrentHashMap private const val TAG = "ChatViewModel" private const val MESSAGE_PAGE_SIZE = 150 +/** + * Canonical comparison key for a tool message's result payload. + * + * WS tool messages store the full tool.complete payload + * (`{"tool_id":..., "name":..., "args":..., "result": {...}}`) while REST + * transcript rows store just the result object (`{"output":..., "exit_code":...}`). + * This key normalizes both sides — preferring the `result` field when present, + * and treating int/float JSON numbers as equal — so the two representations of + * the SAME tool call can be matched regardless of position or pagination. + * Returns null for unparseable content (no match possible). + */ +internal fun canonicalToolResultKey(content: String): String? { + val element = + try { + OkHttpProvider.json.parseToJsonElement(content) + } catch (_: Exception) { + return null + } + + fun canon(e: kotlinx.serialization.json.JsonElement): String = + when (e) { + is kotlinx.serialization.json.JsonObject -> + e.entries.sortedBy { it.key }.joinToString("|") { "${it.key}=${canon(it.value)}" } + is kotlinx.serialization.json.JsonArray -> + e.joinToString(",") { canon(it) } + is kotlinx.serialization.json.JsonPrimitive -> { + // Canonicalize ALL numbers through double, collapsing int/float + // spellings of the same value (0 vs 0.0 → "i0", 0.5 → "d0.5"). + val s = e.content + val d = s.toDoubleOrNull() + if (d != null) { + if (d == d.toLong().toDouble()) "i${d.toLong()}" else "d$d" + } else { + "s$s" + } + } + } + return when (element) { + is kotlinx.serialization.json.JsonObject -> + element["result"]?.let { canon(it) } ?: canon(element) + else -> canon(element) + } +} + +/** + * True when [a] and [b] are the same logical message (the WS-persisted and + * REST-persisted copies of one row — they carry different ids, see #771). + * Tool messages match on their normalized result payload; other roles on + * exact content. + */ +internal fun sameLogicalMessage( + a: ChatMessage, + b: ChatMessage, +): Boolean { + if (a.role != b.role) return false + if (a.role == MessageRole.TOOL) { + val ka = canonicalToolResultKey(a.content) + val kb = canonicalToolResultKey(b.content) + return ka != null && ka == kb + } + return a.content == b.content +} + +/** + * Room accumulates BOTH the WS-persisted copy (UUID id, rich tool payload, + * tool name) and the REST-persisted copy (`rest-` id, result-only payload, + * no tool name) of every message. Painting the cache verbatim renders the + * same call twice. Drop the `rest-` copy whenever a WS copy of the same + * logical message exists (issue #771). + */ +internal fun dedupeCachedMessages(messages: List): List { + val rest = messages.filter { it.id.startsWith("rest-") } + if (rest.isEmpty()) return messages + val nonRest = messages.filterNot { it.id.startsWith("rest-") } + if (nonRest.isEmpty()) return messages + val keepRest = rest.filter { restMsg -> nonRest.none { sameLogicalMessage(it, restMsg) } } + return (nonRest + keepRest).sortedBy { it.timestamp } +} + +/** + * A transcript reload must NOT yank live WS bubbles the server has not + * persisted yet. The gateway stores a tool row only once the tool + * COMPLETES server-side, so a reload that lands while a tool is running + * (app background/foreground mid-turn, reconnect re-resume, pull-refresh) + * returns a page without the tool row — replacing the list outright made + * the in-flight tool bubble vanish and left tool.complete with no RUNNING + * message to update (issue #771). + * + * Merge instead of replace: append any current message the REST page does + * not already cover (checked by id AND logical content via + * [sameLogicalMessage]) and keep chronological order. + */ +internal fun mergeTranscriptWithLive( + restMessages: List, + currentMessages: List, +): List { + val restIds = restMessages.map { it.id }.toSet() + val liveTail = + currentMessages.filter { old -> + old.id !in restIds && restMessages.none { sameLogicalMessage(it, old) } + } + if (liveTail.isEmpty()) return restMessages + return (restMessages + liveTail).sortedBy { it.timestamp } +} + data class ChatUiState( val messages: List = emptyList(), val currentSessionId: String? = null, @@ -538,8 +643,12 @@ class ChatViewModel( } is WsEvent.ToolStart -> { - // Reset streaming state when a tool starts - streamingController.resetStreaming() + // Issue #771: the reducer keeps the streaming message (and its + // reasoning) alive across the tool call so the finalized answer + // retains the thinking card. Only the token buffers are cleared + // here — resetStreaming() would wipe streamingMessage + + // reasoningText and re-introduce the mid-turn reasoning vanish. + streamingController.clearStreamingBuffers() } is WsEvent.RpcResult -> { @@ -1581,7 +1690,7 @@ class ChatViewModel( private fun loadCachedMessages(sessionId: String): Job = viewModelScope.launch(Dispatchers.IO) { - val cachedMessages = repo.loadMessages(sessionId) + val cachedMessages = dedupeCachedMessages(repo.loadMessages(sessionId)) _uiState.update { state -> // Only paint if still showing this session AND no fresher server // page has landed yet (a fast REST fetch must not be clobbered @@ -1609,8 +1718,12 @@ class ChatViewModel( } _uiState.update { state -> if (state.currentSessionId != sessionId) return@update state + // Merge, don't replace: a reload mid-turn must not + // drop live WS bubbles (running tool call, streaming + // answer) the server hasn't persisted yet. Issue #771. + val merged = mergeTranscriptWithLive(chatMessages, state.messages) state.copy( - messages = chatMessages, + messages = merged, isLoading = false, hasOlderMessages = serverOffset > 0 && chatMessages.isNotEmpty(), isLoadingOlder = false, @@ -1807,6 +1920,16 @@ class ChatViewModel( withContext(Dispatchers.IO) { repo.persistMessages(incoming, sessionId) } _uiState.update { current -> if (current.currentSessionId != sessionId) return@update current + // Issue #771: the sync merge was dropping the + // newest tool bubble — the incoming REST page + // didn't include it yet (server persists tool rows + // at completion, but the sync offset may predate + // that), and the fragile toolName/content match + // consumed the WRONG incoming tool for an existing + // one, leaving the newest WS tool with no match + // → dropped. Use sameLogicalMessage (canonical + // result-key match) and always preserve any WS + // message that has no REST counterpart. val unmatchedIncoming = incoming.toMutableList() val mergedList = mutableListOf() @@ -1820,20 +1943,22 @@ class ChatViewModel( mergedList.add(existing) } } else { + // WS message (UUID id, no server index): + // match by canonical content, not fragile + // toolName/content equality. val matchIdx = unmatchedIncoming.indexOfFirst { inc -> - inc.role == existing.role && ( - inc.content == existing.content || - ( - existing.role == MessageRole.TOOL && - inc.toolName != null && - inc.toolName == existing.toolName - ) - ) + sameLogicalMessage(inc, existing) } if (matchIdx >= 0) { - mergedList.add(unmatchedIncoming.removeAt(matchIdx)) + // Prefer the WS copy (richer payload, + // real tool name) when available. + val inc = unmatchedIncoming[matchIdx] + mergedList.add(existing) + unmatchedIncoming.removeAt(matchIdx) } else { + // No REST counterpart (server hasn't + // persisted yet) — KEEP the WS message. mergedList.add(existing) } } @@ -2027,10 +2152,36 @@ class ChatViewModel( .filter { it.reasoningText.isNotBlank() } .associateBy { it.content } + // Tool rows in the REST transcript carry NO tool name — the live WS + // stream was the only source of `toolName`. Match each REST tool row + // to its WS counterpart by RESULT CONTENT (not position — pagination + // and mixed cache state make positional mapping misalign, leaving + // the newest call with a null name → generic "tool" bubble). When a + // match is found the live message is reused wholesale (same id + + // toolName + rich payload), so persistence upserts the same row + // instead of accumulating a second `rest-` copy in Room. Issue #771. + val liveToolByResult = linkedMapOf() + _uiState.value.messages + .filter { it.role == MessageRole.TOOL } + .sortedBy { it.id.startsWith("rest-") } // prefer live WS copies + .forEach { msg -> + canonicalToolResultKey(msg.content)?.let { key -> + liveToolByResult.putIfAbsent(key, msg) + } + } + val baseUrl = AuthManager.getBaseUrl() val token = AuthManager.getToken().orEmpty() - return messages.mapIndexed { index, msg -> + val mapped = mutableListOf() + // The gateway stores a reasoning-model's thinking as its OWN assistant + // row (content = "", reasoning = trace) directly before the answer row. + // Rendering that as a standalone empty assistant bubble is the + // "reasoning box in a separate bubble" artifact — fold it into the + // next assistant message with content instead. Issue #771. + var pendingReasoning: String? = null + + messages.forEachIndexed { index, msg -> val role = when (msg.role?.lowercase()) { "user" -> MessageRole.USER @@ -2047,13 +2198,21 @@ class ChatViewModel( ?: System.currentTimeMillis() val rawContent = msg.contentText - val reasoning = + val rowReasoning = if (msg.reasoningText.isNotBlank()) { msg.reasoningText } else { existingReasoningMap[rawContent]?.reasoningText.orEmpty() } + // Reasoning-only assistant row (the gateway's split storage of a + // reasoning turn): stash the trace, skip the empty bubble, and + // attach it to the next assistant message that has content. + if (role == MessageRole.ASSISTANT && rawContent.isBlank() && rowReasoning.isNotBlank()) { + pendingReasoning = rowReasoning + return@forEachIndexed + } + var finalContent = rawContent var attachments: List? = null if (role == MessageRole.ASSISTANT && rawContent.contains("MEDIA:")) { @@ -2081,16 +2240,54 @@ class ChatViewModel( } } - ChatMessage( - id = "rest-$sessionId-$globalIndex", - role = role, - content = finalContent, - reasoningText = reasoning, - attachments = attachments, - timestamp = timestamp, - isStreaming = false, + val finalReasoning = + if (rowReasoning.isNotBlank()) { + rowReasoning + } else if (role == MessageRole.ASSISTANT && pendingReasoning != null) { + pendingReasoning.also { pendingReasoning = null } + } else { + "" + } + + // Tool rows in the REST transcript carry no tool name. When the + // result payload matches a live WS tool message, reuse it whole — + // keeps the real name, the rich WS payload, AND the same id so + // Room upserts instead of accumulating a duplicate `rest-` row. + if (role == MessageRole.TOOL) { + canonicalToolResultKey(rawContent)?.let { key -> + liveToolByResult[key]?.let { live -> + mapped.add(live) + return@forEachIndexed + } + } + } + + mapped.add( + ChatMessage( + id = "rest-$sessionId-$globalIndex", + role = role, + content = finalContent, + reasoningText = finalReasoning, + attachments = attachments, + timestamp = timestamp, + isStreaming = false, + ), ) } + + // A reasoning-only row with no following answer (interrupted turn): + // don't drop the trace — attach it to the last assistant message. + if (pendingReasoning != null) { + val lastAssistantIdx = mapped.indexOfLast { it.role == MessageRole.ASSISTANT } + if (lastAssistantIdx >= 0) { + val target = mapped[lastAssistantIdx] + if (target.reasoningText.isBlank()) { + mapped[lastAssistantIdx] = target.copy(reasoningText = pendingReasoning!!) + } + } + } + + return mapped } // ── Issue #724: attach host-path MEDIA: files as real attachments ──── diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt index 00b2427f..99172aa1 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt @@ -247,11 +247,31 @@ object ChatWsEventReducer { state: ChatUiState, streamingState: StreamingState, event: WsEvent.ReasoningAvailable, - ): ReducerResult = - ReducerResult( - state = state, - streamingState = streamingState.copy(isReasoning = true), - ) + ): ReducerResult { + // The gateway sends the FULL reasoning trace in reasoning.available + // (payload `text`) once reasoning finishes streaming. When the text + // is present, use it as the authoritative fill — it survives even if + // per-token reasoning.delta events were dropped or wiped by a + // mid-turn tool.start. Attach it to the live streaming message so + // message.complete carries it into the finalized bubble. + val text = event.text + return if (!text.isNullOrBlank() && streamingState.streamingMessage != null) { + ReducerResult( + state = state, + streamingState = + streamingState.copy( + isReasoning = true, + reasoningText = text, + streamingMessage = streamingState.streamingMessage.copy(reasoningText = text), + ), + ) + } else { + ReducerResult( + state = state, + streamingState = streamingState.copy(isReasoning = true), + ) + } + } // ── MessageComplete ─────────────────────────────────────────────── @@ -261,12 +281,15 @@ object ChatWsEventReducer { event: WsEvent.MessageComplete, ): ReducerResult { val streaming = streamingState.streamingMessage + // Prefer the authoritative reasoning carried in the message.complete + // payload (the gateway's assembled full trace), then fall back to + // whatever reasoning.delta deltas accumulated during streaming. + // Fixes: a mid-turn tool.start wiping the streaming reasoning used + // to leave the finalized bubble with no reasoning card at all. val reasoning = - if (streamingState.reasoningText.isNotBlank()) { - streamingState.reasoningText - } else { - streaming?.reasoningText ?: "" - } + event.reasoning + ?.takeIf { it.isNotBlank() } + ?: streamingState.reasoningText.ifBlank { streaming?.reasoningText ?: "" } val msg = streaming?.copy( content = event.text, @@ -392,11 +415,29 @@ object ChatWsEventReducer { effects.add(ReducerEffect.PersistMessage(toolMessage, sid)) } + // Issue #771: KEEP the streaming state across a tool call instead of + // resetting it. A reasoning-model turn streams its thinking BEFORE the + // tool starts (content is still empty here) — wiping the streaming + // message at tool.start made the reasoning bubble vanish mid-turn and + // left the finalized answer with no reasoning card. The stream stays + // alive; post-tool message.delta continues the SAME message and + // message.complete finalizes it with the reasoning intact. + // When an orphan WAS sealed (interim text before the tool), the + // streaming message is done — null it, but keep the reasoning text in + // the shared state so the post-tool fallback message still picks it up. + val finalStreamingState = + if (orphanToPersist != null) { + streamingState.copy(streamingMessage = null) + } else { + streamingState + } + val parsedTodos = if (event.name == "todo") extractTodosFromMap(event.data) else null val nextTodos = parsedTodos ?: newState.todos return ReducerResult( state = newState.copy(messages = newState.messages + toolMessage, todos = nextTodos), + streamingState = finalStreamingState, effects = effects, ) } diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatToolDedupeTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatToolDedupeTest.kt new file mode 100644 index 00000000..07e2b4e3 --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatToolDedupeTest.kt @@ -0,0 +1,408 @@ +package com.m57.hermescontrol.ui.chat + +import com.m57.hermescontrol.data.ws.WsEvent +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * Issue #771 — duplicate tool bubble ("terminal" + generic "Tool") regression. + * + * Real captured payloads for `echo meow`: + * - WS tool.complete content: full payload incl. `result` + * - REST transcript row content: result-only, no tool name + */ +class ChatToolDedupeTest { + private val wsToolContent = + """{"tool_id":"call_00_g45kdmnSROQL4RMaGZrn3669","name":"terminal","args":{"command":"echo meow"},"duration_s":0.988396167755127,"result":{"output":"meow\nrenamed '/tmp/hermes-snap-ecb25e16f404.sh.tmp.21749' -> '/tmp/hermes-snap-ecb25e16f404.sh'","exit_code":0.0,"error":null}}""" + + private val restToolContent = + """{"output": "meow\nrenamed '/tmp/hermes-snap-ecb25e16f404.sh.tmp.21749' -> '/tmp/hermes-snap-ecb25e16f404.sh'", "exit_code": 0, "error": null}""" + + // ── canonicalToolResultKey ───────────────────────────────────────────── + + @Test + fun canonicalKey_wsPayloadAndRestRow_match() { + val wsKey = canonicalToolResultKey(wsToolContent) + val restKey = canonicalToolResultKey(restToolContent) + assertEquals(wsKey, restKey) + } + + @Test + fun canonicalKey_isIntFloatAgnostic() { + // exit_code 0.0 (WS, float) vs 0 (REST, int) must not break the match + val wsKey = canonicalToolResultKey(wsToolContent) + val restKey = + canonicalToolResultKey( + "{\"output\":\"meow\\nrenamed '/tmp/hermes-snap-ecb25e16f404.sh.tmp.21749' " + + "-> '/tmp/hermes-snap-ecb25e16f404.sh'\",\"exit_code\":0,\"error\":null}", + ) + assertEquals(wsKey, restKey) + } + + @Test + fun canonicalKey_differentResults_doNotMatch() { + val a = canonicalToolResultKey(wsToolContent) + val b = canonicalToolResultKey("""{"output":"other","exit_code":1,"error":null}""") + assertFalse(a == b) + } + + @Test + fun canonicalKey_unparseable_returnsNull() { + assertNull(canonicalToolResultKey("not json at all")) + } + + // ── sameLogicalMessage ───────────────────────────────────────────────── + + @Test + fun sameLogicalMessage_wsToolAndRestRow_areSameCall() { + val ws = ChatMessage(role = MessageRole.TOOL, content = wsToolContent, toolName = "terminal") + val rest = ChatMessage(role = MessageRole.TOOL, content = restToolContent) + assertTrue(sameLogicalMessage(ws, rest)) + } + + @Test + fun sameLogicalMessage_differentCalls_notSame() { + val ws = ChatMessage(role = MessageRole.TOOL, content = wsToolContent, toolName = "terminal") + val other = + ChatMessage( + role = MessageRole.TOOL, + content = """{"output":"meow 2","exit_code":0,"error":null}""", + ) + assertFalse(sameLogicalMessage(ws, other)) + } + + @Test + fun sameLogicalMessage_userMessages_matchOnContent() { + val ws = ChatMessage(role = MessageRole.USER, content = "run echo meow") + val rest = ChatMessage(role = MessageRole.USER, content = "run echo meow") + assertTrue(sameLogicalMessage(ws, rest)) + } + + @Test + fun sameLogicalMessage_differentRoles_notSame() { + val ws = ChatMessage(role = MessageRole.TOOL, content = wsToolContent) + val rest = ChatMessage(role = MessageRole.ASSISTANT, content = wsToolContent) + assertFalse(sameLogicalMessage(ws, rest)) + } + + // ── dedupeCachedMessages ─────────────────────────────────────────────── + + @Test + fun dedupe_dropsRestCopiesWhenWsCopyExists() { + val wsUser = ChatMessage(role = MessageRole.USER, content = "run echo meow", timestamp = 1) + val restUser = ChatMessage(role = MessageRole.USER, content = "run echo meow", id = "rest-s-0", timestamp = 1) + val wsTool = ChatMessage(role = MessageRole.TOOL, content = wsToolContent, toolName = "terminal", timestamp = 2) + val restTool = ChatMessage(role = MessageRole.TOOL, content = restToolContent, id = "rest-s-1", timestamp = 2) + val wsAssistant = ChatMessage(role = MessageRole.ASSISTANT, content = "done!!", timestamp = 3) + val restAssistant = + ChatMessage(role = MessageRole.ASSISTANT, content = "done!!", id = "rest-s-2", timestamp = 3) + + val deduped = dedupeCachedMessages(listOf(restUser, wsTool, restTool, wsAssistant, restAssistant)) + + assertEquals(3, deduped.size) + // WS copies win; rest- copies of the same logical message are dropped + assertTrue(deduped.contains(wsTool)) + assertTrue(deduped.none { it.id == "rest-s-1" }) + assertTrue(deduped.none { it.id == "rest-s-2" }) + // rest- user kept when no WS copy exists + assertTrue(deduped.contains(restUser)) + } + + @Test + fun dedupe_onlyRestRows_unchanged() { + val restTool = ChatMessage(role = MessageRole.TOOL, content = restToolContent, id = "rest-s-0") + val result = dedupeCachedMessages(listOf(restTool)) + assertEquals(1, result.size) + assertEquals("rest-s-0", result[0].id) + } + + @Test + fun dedupe_noRestRows_unchanged() { + val wsTool = ChatMessage(role = MessageRole.TOOL, content = wsToolContent, toolName = "terminal") + val result = dedupeCachedMessages(listOf(wsTool)) + assertEquals(1, result.size) + } + + // ── mergeTranscriptWithLive ──────────────────────────────────────────── + + @Test + fun merge_midTurnReload_keepsInFlightToolBubble() { + // Reload lands while the tool is RUNNING — the REST page has the + // user row only (server persists the tool row at completion). + val current = + listOf( + ChatMessage(role = MessageRole.USER, content = "run echo meow", timestamp = 1), + ChatMessage( + role = MessageRole.TOOL, + content = wsToolContent, + toolName = "terminal", + toolStatus = ToolStatus.RUNNING, + timestamp = 2, + ), + ) + val restPage = + listOf(ChatMessage(role = MessageRole.USER, content = "run echo meow", id = "rest-s-0", timestamp = 1)) + + val merged = mergeTranscriptWithLive(restPage, current) + + assertEquals(2, merged.size) + // The RUNNING tool bubble survives the reload + assertEquals(ToolStatus.RUNNING, merged[1].toolStatus) + assertEquals("terminal", merged[1].toolName) + } + + @Test + fun merge_reusedLiveToolId_noDuplicate() { + // Post-completion reload: mapServerMessages reuses the live WS tool + // message (same id) — the old copy is covered by id, no duplicate. + val liveTool = + ChatMessage( + role = MessageRole.TOOL, + content = wsToolContent, + toolName = "terminal", + toolStatus = ToolStatus.COMPLETED, + timestamp = 2, + ) + val current = listOf(liveTool) + val restPage = + listOf( + ChatMessage(role = MessageRole.USER, content = "run echo meow", id = "rest-s-0", timestamp = 1), + liveTool, + ) + + val merged = mergeTranscriptWithLive(restPage, current) + + assertEquals(2, merged.size) + assertEquals(1, merged.count { it.role == MessageRole.TOOL }) + } + + @Test + fun merge_restToolCopyWithCanonicalMatch_noDuplicate() { + // Rest- copy + WS copy of the same call (canonical match) — the WS + // copy is dropped from the tail, only the REST row stays. + val wsTool = + ChatMessage( + role = MessageRole.TOOL, + content = wsToolContent, + toolName = "terminal", + toolStatus = ToolStatus.COMPLETED, + timestamp = 2, + ) + val restTool = + ChatMessage(role = MessageRole.TOOL, content = restToolContent, id = "rest-s-1", timestamp = 2) + val current = listOf(wsTool) + val restPage = + listOf( + ChatMessage(role = MessageRole.USER, content = "run echo meow", id = "rest-s-0", timestamp = 1), + restTool, + ) + + val merged = mergeTranscriptWithLive(restPage, current) + + assertEquals(2, merged.size) + assertEquals(1, merged.count { it.role == MessageRole.TOOL }) + assertEquals("rest-s-1", merged.single { it.role == MessageRole.TOOL }.id) + } + + @Test + fun merge_preservesChronologicalOrder() { + val current = + listOf( + ChatMessage(role = MessageRole.USER, content = "run echo meow", timestamp = 1), + ChatMessage( + role = MessageRole.TOOL, + content = wsToolContent, + toolName = "terminal", + toolStatus = ToolStatus.RUNNING, + timestamp = 2, + ), + ) + val restPage = + listOf(ChatMessage(role = MessageRole.USER, content = "run echo meow", id = "rest-s-0", timestamp = 1)) + + val merged = mergeTranscriptWithLive(restPage, current) + + assertEquals(1, merged[0].timestamp) + assertEquals(2, merged[1].timestamp) + } + + // ── end-to-end: full turn + mid-turn reload (the user's exact scenario) ── + + @Test + fun fullTurn_withMidTurnReload_toolBubbleSurvivesUntilStreamEnd() { + val state = ChatUiState(currentSessionId = "session-1") + + // 1. WS turn: user → reasoning → tool.start → tool.complete + val userMsg = ChatMessage(role = MessageRole.USER, content = "echo meow", timestamp = 1) + var s = state.copy(messages = listOf(userMsg)) + var stream = StreamingState() + stream = ChatWsEventReducer.reduce(s, stream, WsEvent.MessageStart("session-1"), "session-1").streamingState + stream = + ChatWsEventReducer.reduce( + s, + stream, + WsEvent.ReasoningDelta("thinking about meow", "session-1"), + "session-1", + ).streamingState + var r = + ChatWsEventReducer.reduce( + s, + stream, + WsEvent.ToolStart("terminal", mapOf("args_text" to "echo meow"), "session-1"), + "session-1", + ) + s = r.state + stream = r.streamingState + r = + ChatWsEventReducer.reduce( + s, + stream, + WsEvent.ToolComplete("terminal", mapOf("output" to "meow"), "session-1"), + "session-1", + ) + s = r.state + stream = r.streamingState + + // 2. MID-TURN reload lands while the tool is still running server-side + // (server persists the tool row only at completion → page has no tool). + val restPage = + listOf( + ChatMessage(role = MessageRole.USER, content = "echo meow", id = "rest-s-0", timestamp = 1), + // assistant reasoning carrier row skipped by mapper; tool row ABSENT + ) + val merged = mergeTranscriptWithLive(restPage, s.messages) + s = s.copy(messages = merged) + + // Tool bubble survived the reload + assertEquals(1, s.messages.count { it.role == MessageRole.TOOL }) + assertEquals(ToolStatus.COMPLETED, s.messages.single { it.role == MessageRole.TOOL }.toolStatus) + assertEquals("terminal", s.messages.single { it.role == MessageRole.TOOL }.toolName) + + // 3. Rest of the turn: answer streams → message.complete + stream = ChatWsEventReducer.reduce(s, stream, WsEvent.MessageStart("session-1"), "session-1").streamingState + r = + ChatWsEventReducer.reduce( + s, + stream, + WsEvent.MessageComplete(text = "meow, done!", sessionId = "session-1"), + "session-1", + ) + s = r.state + + // Final list: user + tool + assistant — tool bubble STILL there + assertEquals(3, s.messages.size) + assertEquals(listOf(MessageRole.USER, MessageRole.TOOL, MessageRole.ASSISTANT), s.messages.map { it.role }) + assertEquals("terminal", s.messages[1].toolName) + } + + // ── syncCurrentSession merge: the exact logcat scenario ────────────── + // + // Logcat from the user's device: + // MessageComplete: messages.size=34 tools=8 + // syncCurrentSession: messages.size=35 tools=8 + // sync merge: before=35 after=34 tools=7 ← TOOL DROPPED + // + // The 5-second sync poll fires right after the turn ends (isAgentTyping + // = false, streamingMessage = null). Its REST page is one tool short + // (server offset predates the newest tool's persistence), and the old + // toolName/content match consumed the wrong incoming tool, leaving the + // newest WS tool with no counterpart → dropped. + + @Test + fun syncMerge_newestToolSurvives_whenRestPageLacksIt() { + // 3 existing messages: user + tool(WS, terminal, echo meow) + assistant + val existingTool = + ChatMessage( + id = "ws-uuid-1", + role = MessageRole.TOOL, + content = wsToolContent, + toolName = "terminal", + toolStatus = ToolStatus.COMPLETED, + timestamp = 2, + ) + val current = + listOf( + ChatMessage(id = "rest-s-0", role = MessageRole.USER, content = "echo meow", timestamp = 1), + existingTool, + ChatMessage(id = "rest-s-2", role = MessageRole.ASSISTANT, content = "done!", timestamp = 3), + ) + + // Incoming REST page is SHORT: 2 rows, tool row missing (server lag) + val incoming = + listOf( + ChatMessage(id = "rest-s-0", role = MessageRole.USER, content = "echo meow", timestamp = 1), + ChatMessage(id = "rest-s-2", role = MessageRole.ASSISTANT, content = "done!", timestamp = 3), + ) + + // Replay the sync merge logic + val unmatchedIncoming = incoming.toMutableList() + val mergedList = mutableListOf() + + for (existing in current) { + val existingServerIndex = if (existing.id.startsWith("rest-")) 0 else null + if (existingServerIndex != null) { + val matchIdx = unmatchedIncoming.indexOfFirst { it.id == existing.id } + if (matchIdx >= 0) { + mergedList.add(unmatchedIncoming.removeAt(matchIdx)) + } else { + mergedList.add(existing) + } + } else { + val matchIdx = unmatchedIncoming.indexOfFirst { inc -> sameLogicalMessage(inc, existing) } + if (matchIdx >= 0) { + mergedList.add(existing) + unmatchedIncoming.removeAt(matchIdx) + } else { + mergedList.add(existing) + } + } + } + mergedList.addAll(unmatchedIncoming) + val merged = mergedList.distinctBy { it.id } + + // Tool MUST survive with its name + assertEquals(3, merged.size) + assertEquals(1, merged.count { it.role == MessageRole.TOOL }) + assertEquals("terminal", merged.single { it.role == MessageRole.TOOL }.toolName) + assertEquals("ws-uuid-1", merged.single { it.role == MessageRole.TOOL }.id) + } + + @Test + fun syncMatch_oldWay_wouldDropTheTool() { + // Prove the OLD match logic (toolName/content) would match the wrong + // incoming and drop the newest WS tool when the REST page is short. + val existingTool = + ChatMessage( + id = "ws-uuid-1", + role = MessageRole.TOOL, + content = wsToolContent, + toolName = "terminal", + toolStatus = ToolStatus.COMPLETED, + timestamp = 2, + ) + val current = + listOf( + ChatMessage(id = "rest-s-0", role = MessageRole.USER, content = "echo meow", timestamp = 1), + existingTool, + ChatMessage(id = "rest-s-2", role = MessageRole.ASSISTANT, content = "done!", timestamp = 3), + ) + + // Old logic: match by role + (content equality OR toolName equality) + // With no incoming tool row, there's nothing to match — the WS tool + // should be kept. But if there WAS a different terminal tool in + // incoming, the old match would consume it for the wrong existing. + // The sameLogicalMessage match is stricter (canonical result key). + val incoming = + listOf( + ChatMessage(id = "rest-s-0", role = MessageRole.USER, content = "echo meow", timestamp = 1), + ChatMessage(id = "rest-s-2", role = MessageRole.ASSISTANT, content = "done!", timestamp = 3), + ) + + // With the NEW match, the WS tool has no canonical match in incoming + // → it's kept. Verify sameLogicalMessage returns false for all incoming. + assertTrue(incoming.none { sameLogicalMessage(it, existingTool) }) + } +} diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt index ca0d2c67..86982a78 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt @@ -482,4 +482,159 @@ class ChatWsEventReducerTest { assertEquals(MessageRole.SYSTEM, msg.role) assertEquals("💾 Self-improvement review: Skill 'android-ci' patched", msg.content) } + + // ── Issue #771: reasoning survives a tool call (regression) ──────────── + // + // Real gateway capture for `run echo hi` (reasoning model): + // message.start + // reasoning.delta × N ← thinking streams BEFORE the tool + // tool.generating / tool.start / tool.complete (terminal) + // message.delta × N ← final answer streams AFTER the tool + // reasoning.available (full text) + // message.complete (text + reasoning payload field) + // + // Previously the reducer returned a FRESH StreamingState() at tool.start, + // wiping the streamed reasoning — the reasoning bubble vanished mid-turn + // and the finalized answer had no reasoning card. + + @Test + fun testToolCallTurn_reasoningSurvivesToolStart_andFinalizesWithCard() { + val state = ChatUiState(currentSessionId = "session-1") + + // 1. message.start + val start = ChatWsEventReducer.reduce(state, StreamingState(), WsEvent.MessageStart("session-1"), "session-1") + + // 2. reasoning.delta — thinking streams before the tool + val reasoningTokens = listOf("The user just said", " \"run echo hi\".", " Let me run it.") + var reasoningState = start.streamingState + for (token in reasoningTokens) { + reasoningState = + ChatWsEventReducer.reduce( + state, + reasoningState, + WsEvent.ReasoningDelta(token, "session-1"), + "session-1", + ).streamingState + } + assertEquals(true, reasoningState.isReasoning) + assertEquals("The user just said \"run echo hi\". Let me run it.", reasoningState.reasoningText) + // Streaming message alive (reasoning copy lands on it via the + // controller's flush — the reducer owns the shared text). + assertEquals(true, reasoningState.streamingMessage?.isStreaming) + + // 3. tool.start — must NOT wipe the streaming message or its reasoning + var result = + ChatWsEventReducer.reduce( + state, + reasoningState, + WsEvent.ToolStart("terminal", mapOf("args_text" to "echo hi"), "session-1"), + "session-1", + ) + assertEquals(1, result.state.messages.size) + assertEquals(MessageRole.TOOL, result.state.messages.first().role) + assertEquals("terminal", result.state.messages.first().toolName) + // Issue #771: streaming message + reasoning survive the tool call + assertEquals(true, result.streamingState.streamingMessage?.isStreaming) + assertEquals("The user just said \"run echo hi\". Let me run it.", result.streamingState.reasoningText) + + // 4. tool.complete — tool bubble completes; stream untouched + result = + ChatWsEventReducer.reduce( + result.state, + result.streamingState, + WsEvent.ToolComplete("terminal", mapOf("output" to "hi"), "session-1"), + "session-1", + ) + assertEquals(ToolStatus.COMPLETED, result.state.messages.first().toolStatus) + + // 5. reasoning.available — authoritative full-trace fill + val fullReasoning = "The user just said \"run echo hi\". That's a simple terminal command. Let me just run it." + result = + ChatWsEventReducer.reduce( + result.state, + result.streamingState, + WsEvent.ReasoningAvailable("session-1", fullReasoning), + "session-1", + ) + assertEquals(fullReasoning, result.streamingState.reasoningText) + assertEquals(fullReasoning, result.streamingState.streamingMessage?.reasoningText) + + // 6. message.complete — finalized bubble keeps the reasoning card + result = + ChatWsEventReducer.reduce( + result.state, + result.streamingState, + WsEvent.MessageComplete( + text = "done!! `echo hi` ran clean", + sessionId = "session-1", + reasoning = fullReasoning, + ), + "session-1", + ) + val assistant = result.state.messages.last() + assertEquals(MessageRole.ASSISTANT, assistant.role) + assertEquals("done!! `echo hi` ran clean", assistant.content) + assertEquals(fullReasoning, assistant.reasoningText) + assertFalse(assistant.isStreaming) + // Stream tail cleared after finalize — no ghost duplicate bubble + assertEquals(null, result.streamingState.streamingMessage) + // Exactly one assistant message — no orphan duplication + assertEquals(1, result.state.messages.count { it.role == MessageRole.ASSISTANT }) + } + + @Test + fun testMessageComplete_reasoningPayloadField_isAuthoritativeFallback() { + // Even if reasoning.delta events were entirely lost (throttled/wipe), + // the message.complete payload carries the full trace (real gateway + // capture: payload keys text/usage/status/reasoning). + val state = ChatUiState(currentSessionId = "session-1") + val start = ChatWsEventReducer.reduce(state, StreamingState(), WsEvent.MessageStart("session-1"), "session-1") + + val result = + ChatWsEventReducer.reduce( + start.state, + start.streamingState, + WsEvent.MessageComplete(text = "answer", sessionId = "session-1", reasoning = "payload reasoning"), + "session-1", + ) + + assertEquals("payload reasoning", result.state.messages.last().reasoningText) + } + + @Test + fun testToolStart_withInterimText_sealsOrphan_butClearsStreamTail() { + // When the agent streams interim text BEFORE calling the tool, the + // orphan is sealed into messages (desktop parity), and the streaming + // tail is cleared — but the reasoning text stays in shared state so + // the post-tool answer can still pick it up. + val state = ChatUiState(currentSessionId = "session-1") + val streaming = + StreamingState( + streamingMessage = + ChatMessage( + role = MessageRole.ASSISTANT, + content = "Let me run that for you.", + reasoningText = "thinking...", + isStreaming = true, + ), + isReasoning = true, + reasoningText = "thinking...", + ) + + val result = + ChatWsEventReducer.reduce( + state, + streaming, + WsEvent.ToolStart("terminal", mapOf("args_text" to "echo hi"), "session-1"), + "session-1", + ) + + assertEquals(2, result.state.messages.size) // sealed orphan + tool + assertEquals("Let me run that for you.", result.state.messages[0].content) + assertFalse(result.state.messages[0].isStreaming) + assertEquals("thinking...", result.state.messages[0].reasoningText) + assertEquals(null, result.streamingState.streamingMessage) + // Reasoning preserved for the post-tool message + assertEquals("thinking...", result.streamingState.reasoningText) + } }