fix(chat): keep interrupted replies and notices in place (#1413)
This commit is contained in:
6 files changed
+188
-3
No files matched your search
@@ -1117,6 +1117,9 @@ class ChatViewModel(
|
|||||||
}
|
}
|
||||||
|
|
||||||
is WsEvent.MessageToken -> {
|
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
|
if (isCurrentSession(event.sessionId)) mainTurnBusy = true
|
||||||
streamingController.handleMessageToken(event)
|
streamingController.handleMessageToken(event)
|
||||||
}
|
}
|
||||||
@@ -2019,9 +2022,23 @@ class ChatViewModel(
|
|||||||
if (targetsCurrentTurn && status in setOf("interrupted", "not_interrupted")) {
|
if (targetsCurrentTurn && status in setOf("interrupted", "not_interrupted")) {
|
||||||
// Never let a late result for an older same-session turn
|
// Never let a late result for an older same-session turn
|
||||||
// erase a newer stream or make its replacement race it.
|
// erase a newer stream or make its replacement race it.
|
||||||
|
streamingController.flushPendingTransition()
|
||||||
|
val interruptedStream = _streamingState.value
|
||||||
sealStreamingMessageIfAny()
|
sealStreamingMessageIfAny()
|
||||||
|
val interruptedMessage =
|
||||||
|
interruptedStream.streamingMessage?.let { stream ->
|
||||||
|
_uiState.value.messages.firstOrNull { it.id == stream.id }
|
||||||
|
}
|
||||||
_uiState.update { it.copy(isAgentTyping = false) }
|
_uiState.update { it.copy(isAgentTyping = false) }
|
||||||
_streamingState.update { StreamingState() }
|
_streamingState.update {
|
||||||
|
interruptedStream.copy(
|
||||||
|
streamingMessage = null,
|
||||||
|
interruptedMessage = interruptedMessage,
|
||||||
|
isThinking = false,
|
||||||
|
thinkingText = "",
|
||||||
|
isReasoning = false,
|
||||||
|
)
|
||||||
|
}
|
||||||
streamingController.resetStreaming()
|
streamingController.resetStreaming()
|
||||||
if (status == "interrupted") addSystemMessage("Session interrupted")
|
if (status == "interrupted") addSystemMessage("Session interrupted")
|
||||||
mainTurnBusy = false
|
mainTurnBusy = false
|
||||||
|
|||||||
@@ -457,7 +457,7 @@ object ChatWsEventReducer {
|
|||||||
val finalSnapshot = event.rawPayload?.let(::parseUsageSnapshot)
|
val finalSnapshot = event.rawPayload?.let(::parseUsageSnapshot)
|
||||||
val usageState = finalSnapshot?.let { applyUsageSnapshot(state, it) } ?: state
|
val usageState = finalSnapshot?.let { applyUsageSnapshot(state, it) } ?: state
|
||||||
val turnUsage = finalSnapshot?.deltaFrom(streamingState.turnUsageBaseline)
|
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
|
// Prefer the authoritative reasoning carried in the message.complete
|
||||||
// payload (the gateway's assembled full trace), then fall back to
|
// payload (the gateway's assembled full trace), then fall back to
|
||||||
// whatever reasoning.delta deltas accumulated during streaming.
|
// whatever reasoning.delta deltas accumulated during streaming.
|
||||||
@@ -663,7 +663,7 @@ object ChatWsEventReducer {
|
|||||||
streamingState.streamingMessage
|
streamingState.streamingMessage
|
||||||
?: return ReducerResult(
|
?: return ReducerResult(
|
||||||
state = state.copy(isAgentTyping = false),
|
state = state.copy(isAgentTyping = false),
|
||||||
streamingState = StreamingState(),
|
streamingState = streamingState.takeIf { it.interruptedMessage != null } ?: StreamingState(),
|
||||||
)
|
)
|
||||||
val reasoning =
|
val reasoning =
|
||||||
if (streamingState.reasoningText.isNotBlank()) {
|
if (streamingState.reasoningText.isNotBlank()) {
|
||||||
|
|||||||
@@ -4,6 +4,8 @@ import com.m57.hermescontrol.data.model.UsageSnapshotResponse
|
|||||||
|
|
||||||
data class StreamingState(
|
data class StreamingState(
|
||||||
val streamingMessage: ChatMessage? = null,
|
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 isThinking: Boolean = false,
|
||||||
val thinkingText: String = "",
|
val thinkingText: String = "",
|
||||||
val isReasoning: Boolean = false,
|
val isReasoning: Boolean = false,
|
||||||
|
|||||||
@@ -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
|
@Test
|
||||||
fun commandEchoAndOutputStayInPlaceAcrossFutureSyncs() {
|
fun commandEchoAndOutputStayInPlaceAcrossFutureSyncs() {
|
||||||
val prompt = ChatMessage(id = "rest-session-0", role = MessageRole.USER, content = "what model are you?")
|
val prompt = ChatMessage(id = "rest-session-0", role = MessageRole.USER, content = "what model are you?")
|
||||||
|
|||||||
@@ -3922,6 +3922,41 @@ class ChatViewModelTest {
|
|||||||
verify(exactly = 1) { HermesWsClient.sendMessage(sessionId, "Fresh turn", any(), false) }
|
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
|
@Test
|
||||||
fun hardInterruptWaitsForAckBeforeSubmitting() =
|
fun hardInterruptWaitsForAckBeforeSubmitting() =
|
||||||
runTest {
|
runTest {
|
||||||
|
|||||||
@@ -9,6 +9,43 @@ import org.junit.Assert.assertTrue
|
|||||||
import org.junit.Test
|
import org.junit.Test
|
||||||
|
|
||||||
class ChatWsEventReducerTest {
|
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
|
@Test
|
||||||
fun testMessageComplete_clearsResolvedClarifyRequest() {
|
fun testMessageComplete_clearsResolvedClarifyRequest() {
|
||||||
val state =
|
val state =
|
||||||
|
|||||||
Reference in new issue
Block a user