fix(chat): deduplicate message ids (#810)
This commit is contained in:
1 parent
356ccc98b5
commit
87442e1a92
4 files changed
+61
-13
No files matched your search
@@ -154,25 +154,27 @@ internal fun mergeTranscriptWithLive(
|
|||||||
restMessages: List<ChatMessage>,
|
restMessages: List<ChatMessage>,
|
||||||
currentMessages: List<ChatMessage>,
|
currentMessages: List<ChatMessage>,
|
||||||
): List<ChatMessage> {
|
): List<ChatMessage> {
|
||||||
|
val dedupedRest = restMessages.dedupeById()
|
||||||
|
val dedupedCurrent = currentMessages.dedupeById()
|
||||||
// User rows can carry live/cache-only metadata (for example whether the
|
// User rows can carry live/cache-only metadata (for example whether the
|
||||||
// bubble continues the active turn). Keep that richer copy when REST
|
// bubble continues the active turn). Keep that richer copy when REST
|
||||||
// returns the same logical row.
|
// returns the same logical row.
|
||||||
val currentUsers = currentMessages.filter { it.role == MessageRole.USER }.toMutableList()
|
val currentUsers = dedupedCurrent.filter { it.role == MessageRole.USER }.toMutableList()
|
||||||
val mergedRest =
|
val mergedRest =
|
||||||
restMessages.map { rest ->
|
dedupedRest.map { rest ->
|
||||||
if (rest.role != MessageRole.USER) return@map rest
|
if (rest.role != MessageRole.USER) return@map rest
|
||||||
val matchIndex =
|
val matchIndex =
|
||||||
currentUsers.indexOfFirst { it.id == rest.id }.takeIf { it >= 0 }
|
currentUsers.indexOfFirst { it.id == rest.id }.takeIf { it >= 0 }
|
||||||
?: currentUsers.indexOfFirst { sameLogicalMessage(rest, it) }
|
?: currentUsers.indexOfFirst { sameLogicalMessage(rest, it) }
|
||||||
if (matchIndex >= 0) currentUsers.removeAt(matchIndex) else rest
|
if (matchIndex >= 0) currentUsers.removeAt(matchIndex) else rest
|
||||||
}
|
}
|
||||||
val restIds = restMessages.map { it.id }.toSet()
|
val restIds = dedupedRest.map { it.id }.toSet()
|
||||||
val liveTail =
|
val liveTail =
|
||||||
currentMessages.filter { old ->
|
dedupedCurrent.filter { old ->
|
||||||
old.id !in restIds && restMessages.none { sameLogicalMessage(it, old) }
|
old.id !in restIds && mergedRest.none { sameLogicalMessage(it, old) }
|
||||||
}
|
}
|
||||||
if (liveTail.isEmpty()) return mergedRest
|
if (liveTail.isEmpty()) return mergedRest.dedupeById()
|
||||||
return (mergedRest + liveTail).sortedBy { it.timestamp }
|
return (mergedRest + liveTail).dedupeById().sortedBy { it.timestamp }
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Merge a REST page without matching it against already-settled transcript rows. */
|
/** Merge a REST page without matching it against already-settled transcript rows. */
|
||||||
@@ -186,8 +188,10 @@ internal fun mergeIncrementalTranscriptPage(
|
|||||||
currentMessages.indexOfLast { message ->
|
currentMessages.indexOfLast { message ->
|
||||||
serverMessageIndex(message.id, sessionId)?.let { it < offset } == true
|
serverMessageIndex(message.id, sessionId)?.let { it < offset } == true
|
||||||
}
|
}
|
||||||
return currentMessages.take(settledEnd + 1) +
|
return (
|
||||||
mergeTranscriptWithLive(restMessages, currentMessages.drop(settledEnd + 1))
|
currentMessages.take(settledEnd + 1) +
|
||||||
|
mergeTranscriptWithLive(restMessages, currentMessages.drop(settledEnd + 1))
|
||||||
|
).dedupeById()
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun serverMessageIndex(
|
private fun serverMessageIndex(
|
||||||
@@ -195,6 +199,8 @@ private fun serverMessageIndex(
|
|||||||
sessionId: String,
|
sessionId: String,
|
||||||
): Int? = id.removePrefix("rest-$sessionId-").takeIf { it != id }?.toIntOrNull()
|
): Int? = id.removePrefix("rest-$sessionId-").takeIf { it != id }?.toIntOrNull()
|
||||||
|
|
||||||
|
internal fun List<ChatMessage>.dedupeById(): List<ChatMessage> = associateBy { it.id }.values.toList()
|
||||||
|
|
||||||
data class ChatUiState(
|
data class ChatUiState(
|
||||||
val messages: List<ChatMessage> = emptyList(),
|
val messages: List<ChatMessage> = emptyList(),
|
||||||
val currentSessionId: String? = null,
|
val currentSessionId: String? = null,
|
||||||
|
|||||||
@@ -8,6 +8,10 @@ import kotlinx.serialization.json.JsonArray
|
|||||||
import kotlinx.serialization.json.JsonObject
|
import kotlinx.serialization.json.JsonObject
|
||||||
import kotlinx.serialization.json.JsonPrimitive
|
import kotlinx.serialization.json.JsonPrimitive
|
||||||
|
|
||||||
|
private fun List<ChatMessage>.upsertById(message: ChatMessage): List<ChatMessage> {
|
||||||
|
return (this + message).dedupeById()
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Pure state reducer for WebSocket events.
|
* Pure state reducer for WebSocket events.
|
||||||
*
|
*
|
||||||
@@ -168,7 +172,7 @@ object ChatWsEventReducer {
|
|||||||
val finalized = streamingState.streamingMessage.copy(isStreaming = false)
|
val finalized = streamingState.streamingMessage.copy(isStreaming = false)
|
||||||
orphan = finalized
|
orphan = finalized
|
||||||
state.copy(
|
state.copy(
|
||||||
messages = state.messages + finalized,
|
messages = state.messages.upsertById(finalized),
|
||||||
isAgentTyping = true,
|
isAgentTyping = true,
|
||||||
subagentIndicators = cleanedSubagents,
|
subagentIndicators = cleanedSubagents,
|
||||||
)
|
)
|
||||||
@@ -319,7 +323,7 @@ object ChatWsEventReducer {
|
|||||||
return ReducerResult(
|
return ReducerResult(
|
||||||
state =
|
state =
|
||||||
state.copy(
|
state.copy(
|
||||||
messages = state.messages + msg,
|
messages = state.messages.upsertById(msg),
|
||||||
isAgentTyping = false,
|
isAgentTyping = false,
|
||||||
),
|
),
|
||||||
effects = effects,
|
effects = effects,
|
||||||
@@ -358,7 +362,7 @@ object ChatWsEventReducer {
|
|||||||
return ReducerResult(
|
return ReducerResult(
|
||||||
state =
|
state =
|
||||||
state.copy(
|
state.copy(
|
||||||
messages = state.messages + msg,
|
messages = state.messages.upsertById(msg),
|
||||||
isAgentTyping = false,
|
isAgentTyping = false,
|
||||||
),
|
),
|
||||||
effects = effects,
|
effects = effects,
|
||||||
@@ -401,7 +405,7 @@ object ChatWsEventReducer {
|
|||||||
)
|
)
|
||||||
orphanToPersist = finalized
|
orphanToPersist = finalized
|
||||||
state.copy(
|
state.copy(
|
||||||
messages = state.messages + finalized,
|
messages = state.messages.upsertById(finalized),
|
||||||
)
|
)
|
||||||
} else {
|
} else {
|
||||||
state
|
state
|
||||||
|
|||||||
@@ -68,6 +68,16 @@ class ChatViewModelTest {
|
|||||||
/** Counter used to generate unique WS request IDs. */
|
/** Counter used to generate unique WS request IDs. */
|
||||||
private var reqCount = 0
|
private var reqCount = 0
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun mergeTranscriptWithLive_collapsesDuplicateIdsKeepingLatestMessage() {
|
||||||
|
val stale = ChatMessage(id = "duplicate-id", role = MessageRole.ASSISTANT, content = "stale")
|
||||||
|
val latest = stale.copy(content = "latest")
|
||||||
|
|
||||||
|
val merged = mergeTranscriptWithLive(emptyList(), listOf(stale, latest))
|
||||||
|
|
||||||
|
assertEquals(listOf(latest), merged)
|
||||||
|
}
|
||||||
|
|
||||||
@Before
|
@Before
|
||||||
fun setUp() {
|
fun setUp() {
|
||||||
Dispatchers.setMain(testDispatcher)
|
Dispatchers.setMain(testDispatcher)
|
||||||
|
|||||||
@@ -377,6 +377,34 @@ class ChatWsEventReducerTest {
|
|||||||
assertEquals("", persistEffect?.message?.reasoningText)
|
assertEquals("", persistEffect?.message?.reasoningText)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testMessageComplete_replacesExistingStreamingMessageWithSameId() {
|
||||||
|
val streaming =
|
||||||
|
ChatMessage(
|
||||||
|
id = "037cae0c-494e-4a36-862f-3f3e8a235950",
|
||||||
|
role = MessageRole.ASSISTANT,
|
||||||
|
content = "partial",
|
||||||
|
isStreaming = true,
|
||||||
|
)
|
||||||
|
val result =
|
||||||
|
ChatWsEventReducer.reduce(
|
||||||
|
state =
|
||||||
|
ChatUiState(
|
||||||
|
messages =
|
||||||
|
listOf(
|
||||||
|
streaming.copy(content = "stale", isStreaming = false),
|
||||||
|
streaming.copy(isStreaming = false),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
streamingState = StreamingState(streamingMessage = streaming),
|
||||||
|
event = WsEvent.MessageComplete(text = "complete", sessionId = "session-1"),
|
||||||
|
currentSessionId = "session-1",
|
||||||
|
)
|
||||||
|
|
||||||
|
assertEquals(1, result.state.messages.size)
|
||||||
|
assertEquals("complete", result.state.messages.single().content)
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun testMessageStart_prunesCompletedSubagents() {
|
fun testMessageStart_prunesCompletedSubagents() {
|
||||||
val completedSubagent =
|
val completedSubagent =
|
||||||
|
|||||||
Reference in new issue
Block a user