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 c2d46f96..027ba385 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 @@ -1117,6 +1117,9 @@ class ChatViewModel( } is WsEvent.MessageToken -> { + // The interrupted partial is already sealed. Wait for its authoritative + // completion instead of letting trailing deltas create a second stream. + if (_streamingState.value.interruptedMessage != null && isCurrentSession(event.sessionId)) return if (isCurrentSession(event.sessionId)) mainTurnBusy = true streamingController.handleMessageToken(event) } @@ -2019,9 +2022,23 @@ class ChatViewModel( if (targetsCurrentTurn && status in setOf("interrupted", "not_interrupted")) { // Never let a late result for an older same-session turn // erase a newer stream or make its replacement race it. + streamingController.flushPendingTransition() + val interruptedStream = _streamingState.value sealStreamingMessageIfAny() + val interruptedMessage = + interruptedStream.streamingMessage?.let { stream -> + _uiState.value.messages.firstOrNull { it.id == stream.id } + } _uiState.update { it.copy(isAgentTyping = false) } - _streamingState.update { StreamingState() } + _streamingState.update { + interruptedStream.copy( + streamingMessage = null, + interruptedMessage = interruptedMessage, + isThinking = false, + thinkingText = "", + isReasoning = false, + ) + } streamingController.resetStreaming() if (status == "interrupted") addSystemMessage("Session interrupted") mainTurnBusy = false 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 1a0c983b..9c2b5e04 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 @@ -457,7 +457,7 @@ object ChatWsEventReducer { val finalSnapshot = event.rawPayload?.let(::parseUsageSnapshot) val usageState = finalSnapshot?.let { applyUsageSnapshot(state, it) } ?: state val turnUsage = finalSnapshot?.deltaFrom(streamingState.turnUsageBaseline) - val streaming = streamingState.streamingMessage + val streaming = streamingState.interruptedMessage ?: 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. @@ -663,7 +663,7 @@ object ChatWsEventReducer { streamingState.streamingMessage ?: return ReducerResult( state = state.copy(isAgentTyping = false), - streamingState = StreamingState(), + streamingState = streamingState.takeIf { it.interruptedMessage != null } ?: StreamingState(), ) val reasoning = if (streamingState.reasoningText.isNotBlank()) { diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/StreamingState.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/StreamingState.kt index 03ef553d..24c01180 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/StreamingState.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/StreamingState.kt @@ -4,6 +4,8 @@ import com.m57.hermescontrol.data.model.UsageSnapshotResponse data class StreamingState( val streamingMessage: ChatMessage? = null, + /** Sealed by an interrupt acknowledgement; late completion must update this row, not append another. */ + val interruptedMessage: ChatMessage? = null, val isThinking: Boolean = false, val thinkingText: String = "", val isReasoning: Boolean = false, diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt index 69b18af1..9f4ffe28 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt @@ -952,6 +952,100 @@ class ChatPagingMergeTest { ) } + private fun laterTurnRows(assistantText: String) = + listOf( + SessionMessage( + id = 0, + role = "user", + content = JsonPrimitive("long running task"), + timestamp = JsonPrimitive(1), + ), + SessionMessage( + id = 1, + role = "assistant", + content = JsonPrimitive(assistantText), + timestamp = JsonPrimitive(2), + ), + SessionMessage( + id = 2, + role = "user", + content = JsonPrimitive("next question"), + timestamp = JsonPrimitive(3), + ), + SessionMessage( + id = 3, + role = "assistant", + content = JsonPrimitive("next answer"), + timestamp = JsonPrimitive(4), + ), + ) + + @Test + fun stopNoticesStayInPlaceWhenInterruptedReplyIsStillLiveOnly() { + val prompt = ChatMessage(id = "rest-session-0", role = MessageRole.USER, content = "long running task") + val liveReply = ChatMessage(id = "live-reply", role = MessageRole.ASSISTANT, content = "Working on it...") + val stop = ChatMessage(id = "uuid-stop", role = MessageRole.USER, content = "/stop") + val processes = + ChatMessage(id = "uuid-procs", role = MessageRole.SYSTEM, content = "Stopped 2 background processes.") + val interrupted = ChatMessage(id = "uuid-int", role = MessageRole.SYSTEM, content = "Session interrupted") + + val merged = + applyServerPage(listOf(prompt, liveReply, stop, processes, interrupted), laterTurnRows("Working on it...")) + + assertEquals( + listOf( + "rest-session-0", + "rest-session-1", + "uuid-stop", + "uuid-procs", + "uuid-int", + "rest-session-2", + "rest-session-3", + ), + merged.map { it.canonicalRestId ?: it.id }, + ) + } + + @Test + fun stopNoticesStayInPlaceWhenLongInterruptedReplyIsAPrefixOfTheServerCopy() { + val long = "Working on the migration plan step by step, first the schema then the data" + val prompt = ChatMessage(id = "rest-session-0", role = MessageRole.USER, content = "long running task") + val liveReply = ChatMessage(id = "live-reply", role = MessageRole.ASSISTANT, content = long) + val stop = ChatMessage(id = "uuid-stop", role = MessageRole.USER, content = "/stop") + val interrupted = ChatMessage(id = "uuid-int", role = MessageRole.SYSTEM, content = "Session interrupted") + + val merged = applyServerPage(listOf(prompt, liveReply, stop, interrupted), laterTurnRows("$long, then indexes")) + + val ids = merged.map { it.canonicalRestId ?: it.id } + assertTrue("notices before later turns: $ids", ids.indexOf("uuid-int") < ids.indexOf("rest-session-2")) + } + + @Test + fun processResultArrivingAfterInterruptNoticeStaysBeforeLaterTurns() { + val prompt = ChatMessage(id = "rest-session-0", role = MessageRole.USER, content = "long running task") + val reply = ChatMessage(id = "rest-session-1", role = MessageRole.ASSISTANT, content = "Working on it...") + val stop = ChatMessage(id = "uuid-stop", role = MessageRole.USER, content = "/stop") + val interrupted = ChatMessage(id = "uuid-int", role = MessageRole.SYSTEM, content = "Session interrupted") + val processes = + ChatMessage(id = "uuid-procs", role = MessageRole.SYSTEM, content = "No background processes to stop.") + + val merged = + applyServerPage(listOf(prompt, reply, stop, interrupted, processes), laterTurnRows("Working on it...")) + + assertEquals( + listOf( + "rest-session-0", + "rest-session-1", + "uuid-stop", + "uuid-int", + "uuid-procs", + "rest-session-2", + "rest-session-3", + ), + merged.map { it.canonicalRestId ?: it.id }, + ) + } + @Test fun commandEchoAndOutputStayInPlaceAcrossFutureSyncs() { val prompt = ChatMessage(id = "rest-session-0", role = MessageRole.USER, content = "what model are you?") diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt index 7a35896f..65c97925 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt @@ -3922,6 +3922,41 @@ class ChatViewModelTest { verify(exactly = 1) { HermesWsClient.sendMessage(sessionId, "Fresh turn", any(), false) } } + @Test + fun lateCompletionAfterInterruptUpdatesSealedReplyInPlace() = + runTest { + val (viewModel, sessionId) = createViewModelWithSession() + mockEventsFlow.emit(WsEvent.MessageStart(sessionId)) + mockEventsFlow.emit(WsEvent.MessageToken("partial reply", sessionId)) + advanceUntilIdle() + val originalId = + viewModel.streamingState.value.streamingMessage!! + .id + + viewModel.interruptSession() + advanceUntilIdle() + val id = sentRequestMethods.last { it.first == WsMethods.SESSION_INTERRUPT }.second + mockEventsFlow.emit(WsEvent.RpcResult(id, mapOf("status" to "interrupted"))) + advanceUntilIdle() + assertNull(viewModel.streamingState.value.streamingMessage) + + mockEventsFlow.emit(WsEvent.MessageToken("late tail", sessionId)) + advanceUntilIdle() + assertNull(viewModel.streamingState.value.streamingMessage) + mockEventsFlow.emit(WsEvent.MessageComplete("partial reply with final tail", sessionId)) + advanceUntilIdle() + + val messages = viewModel.uiState.value.messages + val replies = messages.filter { it.role == MessageRole.ASSISTANT } + assertEquals(1, replies.size) + assertEquals(originalId, replies.single().id) + assertEquals("partial reply with final tail", replies.single().content) + assertTrue( + messages.indexOf(replies.single()) < messages.indexOfFirst { it.content == "Session interrupted" }, + ) + assertNull(viewModel.streamingState.value.interruptedMessage) + } + @Test fun hardInterruptWaitsForAckBeforeSubmitting() = runTest { 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 6d3dc887..fc25edfa 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 @@ -9,6 +9,43 @@ import org.junit.Assert.assertTrue import org.junit.Test class ChatWsEventReducerTest { + @Test + fun interruptedIdentitySurvivesDoneButNotANewStart() { + val partial = ChatMessage(id = "partial", role = MessageRole.ASSISTANT, content = "old reply") + val state = ChatUiState(currentSessionId = "session-1", messages = listOf(partial)) + val interrupted = StreamingState(interruptedMessage = partial) + val done = + ChatWsEventReducer.reduce( + state, + interrupted, + WsEvent.MessageDone("session-1"), + "session-1", + ) + assertEquals(partial, done.streamingState.interruptedMessage) + val started = + ChatWsEventReducer.reduce( + done.state, + done.streamingState, + WsEvent.MessageStart("session-1"), + "session-1", + ) + assertNull(started.streamingState.interruptedMessage) + val complete = + ChatWsEventReducer.reduce( + started.state, + started.streamingState, + WsEvent.MessageComplete("new reply", "session-1"), + "session-1", + ) + assertEquals( + "old reply", + complete.state.messages + .first() + .content, + ) + assertEquals(2, complete.state.messages.count { it.role == MessageRole.ASSISTANT }) + } + @Test fun testMessageComplete_clearsResolvedClarifyRequest() { val state =