feat(chat): add paged history with stable transcript and scroll identity (#1211)
Co-authored-by: m57 <41645758+Hy4ri@users.noreply.github.com>
This commit is contained in:
AlexDev_andm57 authored and GitHub committed
1 parent
83b2a8fa9b
commit
a78b20bc99
26 files changed
+3942
-884
No files matched your search
@@ -1,23 +1,83 @@
|
||||
package com.m57.hermescontrol.data.local
|
||||
|
||||
import androidx.room3.Dao
|
||||
import androidx.room3.Insert
|
||||
import androidx.room3.OnConflictStrategy
|
||||
import androidx.room3.Query
|
||||
import androidx.room3.Transaction
|
||||
import androidx.room3.Upsert
|
||||
|
||||
@Dao
|
||||
interface ChatMessageDao {
|
||||
@Query("SELECT EXISTS(SELECT 1 FROM chat_messages WHERE session_id = :sessionId)")
|
||||
suspend fun sessionExists(sessionId: String): Boolean
|
||||
|
||||
@Query("SELECT * FROM chat_messages WHERE session_id = :sessionId ORDER BY timestamp ASC")
|
||||
@Query(
|
||||
"SELECT * FROM chat_messages WHERE session_id = :sessionId " +
|
||||
"ORDER BY sort_group, sort_order, id",
|
||||
)
|
||||
suspend fun getMessagesForSession(sessionId: String): List<ChatMessageEntity>
|
||||
|
||||
@Insert(onConflict = OnConflictStrategy.REPLACE)
|
||||
suspend fun upsert(message: ChatMessageEntity)
|
||||
@Query(
|
||||
"SELECT * FROM chat_messages WHERE session_id = :sessionId " +
|
||||
"ORDER BY sort_group DESC, sort_order DESC, id DESC LIMIT :limit",
|
||||
)
|
||||
suspend fun getLatestMessagePage(
|
||||
sessionId: String,
|
||||
limit: Int,
|
||||
): List<ChatMessageEntity>
|
||||
|
||||
@Insert(onConflict = OnConflictStrategy.REPLACE)
|
||||
suspend fun upsertAll(messages: List<ChatMessageEntity>)
|
||||
@Query(
|
||||
"SELECT * FROM chat_messages WHERE session_id = :sessionId " +
|
||||
"AND (sort_group, sort_order, id) < (:beforeGroup, :beforeOrder, :beforeId) " +
|
||||
"ORDER BY sort_group DESC, sort_order DESC, id DESC LIMIT :limit",
|
||||
)
|
||||
suspend fun getMessagePage(
|
||||
sessionId: String,
|
||||
beforeGroup: Int,
|
||||
beforeOrder: Long,
|
||||
beforeId: String,
|
||||
limit: Int,
|
||||
): List<ChatMessageEntity>
|
||||
|
||||
@Query("SELECT * FROM chat_messages WHERE id = :id")
|
||||
suspend fun getMessage(id: String): ChatMessageEntity?
|
||||
|
||||
// Physical insertion sequence includes confirmed aliases, so confirmation cannot reuse a local cursor key.
|
||||
@Query("SELECT COALESCE(MAX(rowid), 0) + 1 FROM chat_messages")
|
||||
suspend fun nextLocalOrder(): Long
|
||||
|
||||
@Upsert
|
||||
suspend fun writeMessage(message: ChatMessageEntity)
|
||||
|
||||
// Allocate once, in the same transaction as the write. Content/clock updates never move a local row.
|
||||
@Transaction
|
||||
suspend fun upsert(message: ChatMessageEntity) {
|
||||
val existing = getMessage(message.id)
|
||||
val restId = message.restId ?: existing?.restId
|
||||
val canonicalOrder = canonicalMessageOrder(restId ?: message.id, message.sessionId)
|
||||
writeMessage(
|
||||
message.copy(
|
||||
restId = restId,
|
||||
completionId = message.completionId ?: existing?.completionId,
|
||||
sortGroup = if (canonicalOrder != null) 0 else 1,
|
||||
sortOrder = canonicalOrder ?: existing?.sortOrder ?: nextLocalOrder(),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
@Transaction
|
||||
suspend fun upsertAll(messages: List<ChatMessageEntity>) {
|
||||
messages.forEach { upsert(it) }
|
||||
}
|
||||
|
||||
// History confirms identity, but must not overwrite a newer WS-owned payload.
|
||||
@Transaction
|
||||
suspend fun confirmIdentity(message: ChatMessageEntity) {
|
||||
val existing = getMessage(message.id)
|
||||
upsert(
|
||||
existing?.copy(restId = message.restId, completionId = existing.completionId ?: message.completionId)
|
||||
?: message,
|
||||
)
|
||||
}
|
||||
|
||||
@Query("DELETE FROM chat_messages WHERE session_id = :sessionId")
|
||||
suspend fun deleteMessagesForSession(sessionId: String)
|
||||
|
||||
@@ -10,13 +10,13 @@ import androidx.room3.PrimaryKey
|
||||
* UI model but is stored independently so it survives process death.
|
||||
*
|
||||
* Messages are scoped by [sessionId] so switching sessions loads the right
|
||||
* thread. The [timestamp] field preserves original ordering even if Room
|
||||
* reorders internally.
|
||||
* thread. [sortOrder] preserves numeric server order or local insertion order;
|
||||
* [timestamp] is display metadata only.
|
||||
*/
|
||||
@Entity(
|
||||
tableName = "chat_messages",
|
||||
indices = [
|
||||
Index(value = ["session_id", "timestamp"]),
|
||||
Index(value = ["session_id", "sort_group", "sort_order", "id"]),
|
||||
],
|
||||
)
|
||||
data class ChatMessageEntity(
|
||||
@@ -48,4 +48,21 @@ data class ChatMessageEntity(
|
||||
val tps: Double? = null,
|
||||
@ColumnInfo(name = "completion_id")
|
||||
val completionId: String? = null,
|
||||
@ColumnInfo(name = "rest_id")
|
||||
val restId: String? = null,
|
||||
@ColumnInfo(name = "sort_group", defaultValue = "1")
|
||||
val sortGroup: Int = 1,
|
||||
@ColumnInfo(name = "sort_order", defaultValue = "0")
|
||||
val sortOrder: Long = 0,
|
||||
)
|
||||
|
||||
/** Only the exact session prefix and a nonnegative decimal suffix identify a canonical row. */
|
||||
internal fun canonicalMessageOrder(
|
||||
id: String,
|
||||
sessionId: String,
|
||||
): Long? {
|
||||
val prefix = "rest-$sessionId-"
|
||||
if (!id.startsWith(prefix)) return null
|
||||
val suffix = id.removePrefix(prefix)
|
||||
return suffix.takeIf { it.isNotEmpty() && it.all { char -> char in '0'..'9' } }?.toLongOrNull()
|
||||
}
|
||||
@@ -38,6 +38,8 @@ fun ChatMessageEntity.toUiModel(): ChatMessage =
|
||||
tokenCount = tokenCount,
|
||||
tps = tps,
|
||||
completionId = completionId,
|
||||
restId = restId,
|
||||
localOrder = sortOrder.takeIf { sortGroup == 1 && it > 0 },
|
||||
)
|
||||
|
||||
fun ChatMessage.toEntity(sessionId: String): ChatMessageEntity =
|
||||
@@ -56,4 +58,5 @@ fun ChatMessage.toEntity(sessionId: String): ChatMessageEntity =
|
||||
tokenCount = tokenCount,
|
||||
tps = tps,
|
||||
completionId = completionId,
|
||||
restId = restId,
|
||||
)
|
||||
@@ -14,7 +14,7 @@ import java.io.File
|
||||
|
||||
@Database(
|
||||
entities = [ChatMessageEntity::class],
|
||||
version = 8,
|
||||
version = 9,
|
||||
exportSchema = true,
|
||||
)
|
||||
abstract class HermesDatabase : RoomDatabase() {
|
||||
@@ -90,6 +90,31 @@ abstract class HermesDatabase : RoomDatabase() {
|
||||
}
|
||||
}
|
||||
|
||||
val MIGRATION_8_9: Migration =
|
||||
object : Migration(8, 9) {
|
||||
override suspend fun migrate(connection: SQLiteConnection) {
|
||||
connection.execSQL("ALTER TABLE chat_messages ADD COLUMN rest_id TEXT")
|
||||
connection.execSQL("ALTER TABLE chat_messages ADD COLUMN sort_group INTEGER NOT NULL DEFAULT 1")
|
||||
connection.execSQL("ALTER TABLE chat_messages ADD COLUMN sort_order INTEGER NOT NULL DEFAULT 0")
|
||||
// v8 has no durable local sequence. Preserve its surviving physical insertion order.
|
||||
connection.execSQL("UPDATE chat_messages SET sort_order = rowid")
|
||||
val suffix = "substr(id, length(session_id) + 7)"
|
||||
val digits = "ltrim($suffix, '0')"
|
||||
connection.execSQL(
|
||||
"UPDATE chat_messages SET sort_group = 0, sort_order = CAST($suffix AS INTEGER) " +
|
||||
"WHERE substr(id, 1, length(session_id) + 6) = 'rest-' || session_id || '-' " +
|
||||
"AND $suffix != '' AND $suffix NOT GLOB '*[^0-9]*' " +
|
||||
"AND (length($digits) < 19 OR " +
|
||||
"(length($digits) = 19 AND $digits <= '9223372036854775807'))",
|
||||
)
|
||||
connection.execSQL("DROP INDEX index_chat_messages_session_id_timestamp")
|
||||
connection.execSQL(
|
||||
"CREATE INDEX index_chat_messages_session_id_sort_group_sort_order_id " +
|
||||
"ON chat_messages (session_id, sort_group, sort_order, id)",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun get(context: Context): HermesDatabase =
|
||||
withContext(Dispatchers.IO) {
|
||||
instance?.let { return@withContext it }
|
||||
@@ -126,6 +151,7 @@ abstract class HermesDatabase : RoomDatabase() {
|
||||
MIGRATION_5_6,
|
||||
MIGRATION_6_7,
|
||||
MIGRATION_7_8,
|
||||
MIGRATION_8_9,
|
||||
).fallbackToDestructiveMigration(false)
|
||||
.build()
|
||||
.also { instance = it }
|
||||
|
||||
@@ -87,8 +87,16 @@ data class ChatMessage(
|
||||
val tps: Double? = null,
|
||||
/** Stable completion identity correlating with reply notifications for read tracking. */
|
||||
val completionId: String? = null,
|
||||
/** Confirmed transcript identity (#859); [id] remains the stable live/render key. */
|
||||
val restId: String? = null,
|
||||
/** Cache insertion sequence for unconfirmed local rows; null before first persistence. */
|
||||
val localOrder: Long? = null,
|
||||
)
|
||||
|
||||
/** Cached REST rows already carry their canonical identity in the persisted primary key. */
|
||||
internal val ChatMessage.canonicalRestId: String?
|
||||
get() = restId ?: id.takeIf { it.startsWith("rest-") }
|
||||
|
||||
/**
|
||||
* Single live transcript log entry for subagent execution.
|
||||
*/
|
||||
|
||||
@@ -36,6 +36,50 @@ open class ChatPersistenceRepository(
|
||||
suspend fun loadMessages(sessionId: String): List<ChatMessage> =
|
||||
daoProvider().getMessagesForSession(sessionId).map { it.toUiModel() }
|
||||
|
||||
data class Cursor(
|
||||
val group: Int,
|
||||
val order: Long,
|
||||
val id: String,
|
||||
)
|
||||
|
||||
data class Page(
|
||||
val messages: List<ChatMessage>,
|
||||
val cursor: Cursor?,
|
||||
val hasOlder: Boolean,
|
||||
)
|
||||
|
||||
/** Read at most one page plus a lookahead row; never trim a full-session query. */
|
||||
suspend fun loadPage(
|
||||
sessionId: String,
|
||||
before: Cursor?,
|
||||
limit: Int,
|
||||
): Page {
|
||||
require(limit in 1..1_000)
|
||||
val dao = daoProvider()
|
||||
val rows =
|
||||
if (before == null) {
|
||||
dao.getLatestMessagePage(sessionId, limit + 1)
|
||||
} else {
|
||||
dao.getMessagePage(sessionId, before.group, before.order, before.id, limit + 1)
|
||||
}
|
||||
val page = rows.take(limit)
|
||||
val oldest = page.lastOrNull()
|
||||
return Page(
|
||||
messages = page.asReversed().map { it.toUiModel() },
|
||||
cursor = oldest?.let { Cursor(it.sortGroup, it.sortOrder, it.id) } ?: before,
|
||||
hasOlder = rows.size > limit,
|
||||
)
|
||||
}
|
||||
|
||||
/** Record confirmed UUID aliases without replacing their newer locally persisted content. */
|
||||
suspend fun confirmIdentities(
|
||||
messages: List<ChatMessage>,
|
||||
sessionId: String,
|
||||
) {
|
||||
val dao = daoProvider()
|
||||
messages.forEach { dao.confirmIdentity(it.toEntity(sessionId)) }
|
||||
}
|
||||
|
||||
/** Clear all cached messages for a session (e.g. after /undo rewind). */
|
||||
suspend fun clearMessagesForSession(sessionId: String) {
|
||||
daoProvider().deleteMessagesForSession(sessionId)
|
||||
|
||||
@@ -59,7 +59,6 @@ import androidx.compose.runtime.remember
|
||||
import androidx.compose.runtime.rememberCoroutineScope
|
||||
import androidx.compose.runtime.saveable.rememberSaveable
|
||||
import androidx.compose.runtime.setValue
|
||||
import androidx.compose.runtime.snapshotFlow
|
||||
import androidx.compose.ui.Alignment
|
||||
import androidx.compose.ui.Modifier
|
||||
import androidx.compose.ui.draw.clip
|
||||
@@ -115,8 +114,6 @@ import com.m57.hermescontrol.ui.common.NavIcon
|
||||
import com.m57.hermescontrol.ui.model.components.ModelPickerDialog
|
||||
import com.m57.hermescontrol.util.ConnectorUrlValidator
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.collectLatest
|
||||
import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
private const val SESSION_SYNC_INTERVAL_MS = 30_000L
|
||||
@@ -211,7 +208,6 @@ fun ChatScreen(
|
||||
val listState = rememberLazyListState(prefetchStrategy = ChatTimelineNoPrefetchStrategy)
|
||||
val scrollScope = rememberCoroutineScope()
|
||||
val scrollController = rememberChatScrollController(listState, scrollScope)
|
||||
var isOlderPagingArmed by remember(state.currentSessionId) { mutableStateOf(false) }
|
||||
var showContextSheet by remember { mutableStateOf(false) }
|
||||
var pendingSavePath by rememberSaveable { mutableStateOf<String?>(null) }
|
||||
var pendingSaveName by rememberSaveable { mutableStateOf<String?>(null) }
|
||||
@@ -265,38 +261,9 @@ fun ChatScreen(
|
||||
}
|
||||
}
|
||||
|
||||
// Arm + trigger older-message paging when the user scrolls near the top.
|
||||
// Before prepending, capture the anchor (first visible message id + offset)
|
||||
// so the same content stays under the reader's eye after the insert
|
||||
// (issue #682). Restored in a LaunchedEffect once the page actually lands.
|
||||
val pagingAnchor = remember { mutableStateOf<Pair<String, Int>?>(null) }
|
||||
LaunchedEffect(listState, state.currentSessionId, state.hasOlderMessages, state.isLoadingOlder) {
|
||||
if (!state.hasOlderMessages || state.isLoadingOlder) return@LaunchedEffect
|
||||
snapshotFlow { listState.firstVisibleItemIndex }
|
||||
.distinctUntilChanged()
|
||||
.collectLatest { firstVisibleIndex ->
|
||||
if (firstVisibleIndex > 2) {
|
||||
isOlderPagingArmed = true
|
||||
} else if (isOlderPagingArmed || state.messages.size <= 3) {
|
||||
val anchorId = state.messages.getOrNull(firstVisibleIndex)?.id
|
||||
val offset = scrollController.captureAnchorOffset()
|
||||
pagingAnchor.value = if (anchorId != null) anchorId to offset else null
|
||||
viewModel.loadOlderMessages()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// After older history is prepended, restore the anchor message (by id, so
|
||||
// streaming-token inserts during paging don't skew the index) plus offset.
|
||||
LaunchedEffect(state.messages) {
|
||||
val anchor = pagingAnchor.value ?: return@LaunchedEffect
|
||||
val (anchorId, offset) = anchor
|
||||
val index = state.messages.indexOfFirst { it.id == anchorId }
|
||||
if (index >= 0) {
|
||||
scrollController.scrollToItem(index, offset)
|
||||
pagingAnchor.value = null
|
||||
}
|
||||
}
|
||||
// FullBleedChatList preserves the live lazy row key and pixel offset on prepend.
|
||||
// Never restore a message index captured before an asynchronous fetch: it is not
|
||||
// a lazy row index and the reader may have moved in the meantime.
|
||||
|
||||
// Continuous bottom-follow tracking from LazyListState (issue #682).
|
||||
LaunchedEffect(Unit) {
|
||||
@@ -705,6 +672,8 @@ fun ChatScreen(
|
||||
maxToolCallsPerTurn = state.maxToolCallsPerTurn,
|
||||
isLoading = state.isLoading,
|
||||
isLoadingOlder = state.isLoadingOlder,
|
||||
hasOlderMessages = state.hasOlderMessages,
|
||||
pagingSessionId = state.currentSessionId,
|
||||
listState = listState,
|
||||
scrollController = scrollController,
|
||||
lastAnimatedMessageId = lastAnimatedMessageId,
|
||||
|
||||
@@ -10,10 +10,9 @@ import com.m57.hermescontrol.notification.ReplyNotificationTracker
|
||||
/**
|
||||
* Maps REST transcript rows ([SessionMessage]) into UI [ChatMessage]s.
|
||||
*
|
||||
* Verbatim extraction of ChatViewModel.mapServerMessages (corral 2 of the
|
||||
* 2026-08 delegate refactor). Pure with respect to the VM: it reads the live
|
||||
* message list supplied by the caller and one paging-mode flag — no state
|
||||
* mutation, no I/O beyond URL construction.
|
||||
* Reads the supplied transcript snapshot and recovers notification identity through
|
||||
* ReplyNotificationTracker's synchronized lookup. Safe on the history dispatcher;
|
||||
* it never mutates ViewModel state or dismisses a notification.
|
||||
*/
|
||||
internal fun mapServerMessages(
|
||||
sessionId: String,
|
||||
@@ -24,15 +23,11 @@ internal fun mapServerMessages(
|
||||
isPagingOlder: Boolean = false,
|
||||
context: android.content.Context? = null,
|
||||
): List<ChatMessage> {
|
||||
val existingReasoningMap =
|
||||
liveMessages
|
||||
.filter { it.reasoningText.isNotBlank() }
|
||||
.associateBy { it.content }
|
||||
|
||||
val existingById = liveMessages.associateBy { it.canonicalRestId ?: it.id }
|
||||
val liveByExactId =
|
||||
liveMessages
|
||||
.filter { !it.completionId.isNullOrBlank() }
|
||||
.associateBy { it.id }
|
||||
.associateBy { it.canonicalRestId ?: it.id }
|
||||
|
||||
val unmappedLiveWsAssistants =
|
||||
if (isPagingOlder) {
|
||||
@@ -40,112 +35,91 @@ internal fun mapServerMessages(
|
||||
} else {
|
||||
liveMessages
|
||||
.filter {
|
||||
it.role == MessageRole.ASSISTANT && !it.id.startsWith("rest-") &&
|
||||
it.role == MessageRole.ASSISTANT && it.canonicalRestId == null &&
|
||||
!it.completionId.isNullOrBlank()
|
||||
}
|
||||
}
|
||||
|
||||
val wsCompletionIdByRestIndex = mutableMapOf<Int, String>()
|
||||
if (!isPagingOlder) {
|
||||
if (unmappedLiveWsAssistants.isNotEmpty()) {
|
||||
val remainingWs = unmappedLiveWsAssistants.toMutableList()
|
||||
for (i in messages.indices.reversed()) {
|
||||
if (remainingWs.isEmpty()) break
|
||||
val m = messages[i]
|
||||
val r =
|
||||
when (m.role?.lowercase()) {
|
||||
"user" -> MessageRole.USER
|
||||
"system" -> MessageRole.SYSTEM
|
||||
"tool" -> MessageRole.TOOL
|
||||
else -> MessageRole.ASSISTANT
|
||||
}
|
||||
if (r != MessageRole.ASSISTANT) continue
|
||||
val rawContent = m.contentText
|
||||
// Reasoning-only or empty tool placeholder: skip, does not become visible prose
|
||||
if (rawContent.isBlank() && !rawContent.contains("MEDIA:")) continue
|
||||
|
||||
val restId =
|
||||
if (latestPaging) {
|
||||
m.id?.let { "rest-$sessionId-$it" } ?: "rest-$sessionId-${offset + i}"
|
||||
} else {
|
||||
"rest-$sessionId-${offset + i}"
|
||||
}
|
||||
if (liveByExactId.containsKey(restId)) continue
|
||||
|
||||
val canonicalContent =
|
||||
if (rawContent.contains("MEDIA:")) {
|
||||
HostMediaExtractor.strip(rawContent).trim()
|
||||
} else {
|
||||
rawContent.trim()
|
||||
}
|
||||
|
||||
val wsIdx = remainingWs.indexOfLast { it.content.trim() == canonicalContent }
|
||||
if (wsIdx >= 0) {
|
||||
remainingWs.removeAt(wsIdx).completionId?.let { compId ->
|
||||
wsCompletionIdByRestIndex[i] = compId
|
||||
}
|
||||
}
|
||||
}
|
||||
fun restIdAt(index: Int): String =
|
||||
if (latestPaging) {
|
||||
"rest-$sessionId-${requireNotNull(messages[index].id) { "Latest transcript row has no stable id" }}"
|
||||
} else {
|
||||
"rest-$sessionId-${offset + index}"
|
||||
}
|
||||
|
||||
// REST-only recovery is fail-closed: only a durable server row ID may
|
||||
// transfer notification identity after the WebSocket copy is gone.
|
||||
val activeTarget = ReplyNotificationTracker.getActiveTarget(context)
|
||||
if (
|
||||
activeTarget != null &&
|
||||
activeTarget.sessionId == sessionId &&
|
||||
activeTarget.serverMessageId != null &&
|
||||
wsCompletionIdByRestIndex.values.none { it == activeTarget.completionId }
|
||||
) {
|
||||
val wsCompletionIdByRestIndex = mutableMapOf<Int, String>()
|
||||
if (!isPagingOlder) {
|
||||
// #129: reserve durable identity before matching any repeated live prose.
|
||||
// REST-only recovery remains fail-closed when the target row is absent.
|
||||
val activeTarget = ReplyNotificationTracker.getActiveTarget(context)?.takeIf { it.sessionId == sessionId }
|
||||
val durableTarget = activeTarget?.takeIf { it.serverMessageId != null }
|
||||
if (durableTarget != null) {
|
||||
val exactIndex =
|
||||
messages.indexOfFirst { message ->
|
||||
message.id == activeTarget.serverMessageId &&
|
||||
message.id == durableTarget.serverMessageId &&
|
||||
message.role.equals("assistant", ignoreCase = true)
|
||||
}
|
||||
if (exactIndex >= 0 && !liveByExactId.containsKey("rest-$sessionId-${activeTarget.serverMessageId}")) {
|
||||
wsCompletionIdByRestIndex[exactIndex] = activeTarget.completionId
|
||||
if (exactIndex >= 0 && !liveByExactId.containsKey(restIdAt(exactIndex))) {
|
||||
wsCompletionIdByRestIndex[exactIndex] = durableTarget.completionId
|
||||
}
|
||||
}
|
||||
val confirmedCompletions =
|
||||
liveMessages.filter { it.canonicalRestId != null }.mapNotNull { it.completionId }.toSet()
|
||||
val remainingWs =
|
||||
unmappedLiveWsAssistants
|
||||
.filter {
|
||||
it.completionId !in confirmedCompletions && it.completionId != durableTarget?.completionId
|
||||
}.toMutableList()
|
||||
// #129: reserve known completions and live aliases. A metadata-free cached REST
|
||||
// echo must remain eligible for its pending WS completion before older duplicates.
|
||||
val reservedRestIds = liveByExactId.keys + liveMessages.mapNotNull { it.restId }
|
||||
// Retain upstream's newest-to-newest, one-to-one WS echo matching.
|
||||
for (i in messages.indices.reversed()) {
|
||||
if (remainingWs.isEmpty()) break
|
||||
if (i in wsCompletionIdByRestIndex || restIdAt(i) in reservedRestIds) continue
|
||||
val m = messages[i]
|
||||
if (m.role?.lowercase() in listOf("user", "system", "tool")) continue
|
||||
val rawContent = m.contentText
|
||||
if (rawContent.isBlank()) continue
|
||||
val canonicalContent =
|
||||
if (rawContent.contains("MEDIA:")) HostMediaExtractor.strip(rawContent).trim() else rawContent.trim()
|
||||
val wsIdx = remainingWs.indexOfLast { it.content.trim() == canonicalContent }
|
||||
if (wsIdx >= 0) {
|
||||
remainingWs.removeAt(wsIdx).completionId?.let { wsCompletionIdByRestIndex[i] = it }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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<String, ChatMessage>()
|
||||
liveMessages
|
||||
.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)
|
||||
// Reasoning follows the same reserved identities as completions, never a reusable text lookup.
|
||||
val assistants = liveMessages.filter { it.role == MessageRole.ASSISTANT }
|
||||
val assistantsByRestId = assistants.groupBy { it.canonicalRestId }
|
||||
val assistantsByCompletion = assistants.groupBy { it.completionId }
|
||||
val reasoningSources = arrayOfNulls<ChatMessage>(messages.size)
|
||||
messages.forEachIndexed { index, row ->
|
||||
if (row.role?.lowercase() in listOf("user", "system", "tool")) return@forEachIndexed
|
||||
val exact = assistantsByRestId[restIdAt(index)].orEmpty()
|
||||
val completion = liveByExactId[restIdAt(index)]?.completionId ?: wsCompletionIdByRestIndex[index]
|
||||
val candidates = exact + completion?.let { assistantsByCompletion[it] }.orEmpty()
|
||||
reasoningSources[index] = candidates.firstOrNull { it.reasoningText.isNotBlank() } ?: candidates.firstOrNull()
|
||||
}
|
||||
if (!isPagingOlder) {
|
||||
val remaining =
|
||||
assistants.filter { it.canonicalRestId == null && it.completionId == null }.toMutableList()
|
||||
for (index in messages.indices.reversed()) {
|
||||
val row = messages[index]
|
||||
if (reasoningSources[index] != null || row.contentText.isBlank() ||
|
||||
row.role?.lowercase() in listOf("user", "system", "tool") ||
|
||||
liveByExactId[restIdAt(index)]?.completionId != null || index in wsCompletionIdByRestIndex
|
||||
) {
|
||||
continue
|
||||
}
|
||||
val content = HostMediaExtractor.strip(row.contentText).trim()
|
||||
val match = remaining.indexOfLast { it.content.trim() == content }
|
||||
if (match >= 0) reasoningSources[index] = remaining.removeAt(match)
|
||||
}
|
||||
|
||||
// Issue #842: REST transcript rows carry the gateway's `tool_call_id`
|
||||
// — prefer matching live bubbles by that 1:1 identity. It works for
|
||||
// EVERY tool shape, including MCP/web rows whose REST copy is raw
|
||||
// `<untrusted_tool_result>` text with no JSON key to canonicalize.
|
||||
val liveToolByCallId = linkedMapOf<String, ChatMessage>()
|
||||
liveMessages
|
||||
.filter { it.role == MessageRole.TOOL && it.toolCallId.isNotBlank() }
|
||||
.sortedBy { it.id.startsWith("rest-") } // prefer live WS copies
|
||||
.forEach { msg ->
|
||||
liveToolByCallId.putIfAbsent(msg.toolCallId, msg)
|
||||
}
|
||||
}
|
||||
|
||||
val mapped = mutableListOf<ChatMessage>()
|
||||
// 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()) {
|
||||
@@ -154,47 +128,32 @@ internal fun mapServerMessages(
|
||||
"tool" -> MessageRole.TOOL
|
||||
else -> MessageRole.ASSISTANT
|
||||
}
|
||||
val globalIndex = offset + index
|
||||
// Issue #859: under newest-anchored paging use the server's
|
||||
// AUTOINCREMENT row id as the stable key — from-end positions shift
|
||||
// as the transcript grows and would collide across hydrations
|
||||
// (distinctBy would silently drop the newest copy). Legacy paging
|
||||
// keeps the absolute-position key its count-based sync math needs.
|
||||
val restId =
|
||||
if (latestPaging) {
|
||||
msg.id?.let { "rest-$sessionId-$it" } ?: "rest-$sessionId-$globalIndex"
|
||||
} else {
|
||||
"rest-$sessionId-$globalIndex"
|
||||
}
|
||||
val restId = restIdAt(index)
|
||||
val timestamp =
|
||||
msg.timestampText
|
||||
?.toDoubleOrNull()
|
||||
?.times(1000)
|
||||
?.toLong()
|
||||
?: existingById[restId]?.timestamp
|
||||
?: System.currentTimeMillis()
|
||||
|
||||
val rawContent = msg.contentText
|
||||
val rowReasoning =
|
||||
if (msg.reasoningText.isNotBlank()) {
|
||||
msg.reasoningText
|
||||
} else {
|
||||
existingReasoningMap[rawContent]?.reasoningText.orEmpty()
|
||||
msg.reasoningText.ifBlank {
|
||||
if (role == MessageRole.ASSISTANT) {
|
||||
reasoningSources[index]?.reasoningText.orEmpty()
|
||||
} else {
|
||||
existingById[restId]?.reasoningText.orEmpty()
|
||||
}
|
||||
}
|
||||
|
||||
// Empty assistant row — two cases stored by the gateway:
|
||||
// 1. Reasoning-only: thinking-model split storage (content = "",
|
||||
// reasoning = trace). Stash the trace and fold it into the
|
||||
// next assistant message that has content (issue #771).
|
||||
// 2. Tool-call placeholder: non-reasoning models emit content = ""
|
||||
// with tool_calls metadata and no reasoning. These carry no
|
||||
// user-visible text and must not render as empty bubbles
|
||||
// (issue #956).
|
||||
if (role == MessageRole.ASSISTANT && rawContent.isBlank()) {
|
||||
if (rowReasoning.isNotBlank()) {
|
||||
pendingReasoning = rowReasoning
|
||||
}
|
||||
return@forEachIndexed
|
||||
}
|
||||
// Retain canonical reasoning rows across page boundaries; hide only textless placeholders.
|
||||
if (role == MessageRole.ASSISTANT && rawContent.isBlank() && rowReasoning.isBlank()) return@forEachIndexed
|
||||
|
||||
// Context-reference preprocessing can persist model-facing attachment
|
||||
// contents after the user's text. A cold reload / second device has no
|
||||
@@ -232,34 +191,6 @@ internal fun mapServerMessages(
|
||||
}
|
||||
}
|
||||
|
||||
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) {
|
||||
// Prefer the gateway call id (1:1, works for every tool
|
||||
// shape), then fall back to result-content matching.
|
||||
liveToolByCallId[msg.toolCallId]?.let { live ->
|
||||
mapped.add(live)
|
||||
return@forEachIndexed
|
||||
}
|
||||
canonicalToolResultKey(rawContent)?.let { key ->
|
||||
liveToolByResult[key]?.let { live ->
|
||||
mapped.add(live)
|
||||
return@forEachIndexed
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val completionId =
|
||||
if (role == MessageRole.ASSISTANT) {
|
||||
liveByExactId[restId]?.completionId ?: wsCompletionIdByRestIndex[index]
|
||||
@@ -272,7 +203,7 @@ internal fun mapServerMessages(
|
||||
id = restId,
|
||||
role = role,
|
||||
content = finalContent,
|
||||
reasoningText = finalReasoning,
|
||||
reasoningText = rowReasoning,
|
||||
toolCallId = msg.toolCallId,
|
||||
attachments = attachments,
|
||||
timestamp = timestamp,
|
||||
@@ -284,17 +215,12 @@ internal fun mapServerMessages(
|
||||
)
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
// REST echoes must not reserve a match before the richer WS copy of that tool.
|
||||
val liveTools = liveMessages.filter { it.role == MessageRole.TOOL && !it.id.startsWith("rest-") }
|
||||
val mappedTools = mapped.filter { it.role == MessageRole.TOOL }
|
||||
val matches = matchTranscriptMessages(mappedTools, liveTools)
|
||||
val toolsById = mappedTools.indices.associate { index -> mappedTools[index].id to matches[index] }
|
||||
return mapped.map { message ->
|
||||
toolsById[message.id]?.copy(restId = message.canonicalRestId) ?: message
|
||||
}
|
||||
|
||||
return mapped
|
||||
}
|
||||
@@ -107,48 +107,120 @@ internal fun canonicalToolResultKey(content: String): String? {
|
||||
internal fun sameLogicalMessage(
|
||||
a: ChatMessage,
|
||||
b: ChatMessage,
|
||||
): Boolean {
|
||||
if (a.role != b.role) return false
|
||||
if (a.role == MessageRole.TOOL) {
|
||||
// Issue #842: prefer the gateway's tool call id when both sides carry
|
||||
// it — the REST transcript stores `tool_call_id` and the live WS
|
||||
// bubble keeps it from `tool.start`. Content canonicalization cannot
|
||||
// cover MCP/web tools: the REST side stores the payload as raw
|
||||
// `<untrusted_tool_result>` text (not JSON), so it has no key at all.
|
||||
if (a.toolCallId.isNotBlank() && b.toolCallId.isNotBlank()) {
|
||||
return a.toolCallId == b.toolCallId
|
||||
): Boolean = TranscriptComparison().same(a, b)
|
||||
|
||||
/** One operation owns this cache; even invalid JSON is parsed at most once per content. */
|
||||
internal class TranscriptComparison(
|
||||
private val canonicalize: (String) -> String? = ::canonicalToolResultKey,
|
||||
) {
|
||||
private val toolKeys = HashMap<String, String?>()
|
||||
private val trimmed = HashMap<String, String>()
|
||||
private val captions = HashMap<String, String>()
|
||||
|
||||
fun toolKey(content: String): String? {
|
||||
if (!toolKeys.containsKey(content)) toolKeys[content] = canonicalize(content)
|
||||
return toolKeys[content]
|
||||
}
|
||||
|
||||
fun same(
|
||||
a: ChatMessage,
|
||||
b: ChatMessage,
|
||||
): Boolean {
|
||||
if (a.role != b.role) return false
|
||||
val aRestId = a.canonicalRestId
|
||||
val bRestId = b.canonicalRestId
|
||||
if (aRestId != null && bRestId != null) return aRestId == bRestId
|
||||
// #129: completion-bearing replies need confirmed identity, not repeated prose.
|
||||
// Older pages deliberately do not acquire a live reply's completion in the mapper.
|
||||
if (a.role == MessageRole.ASSISTANT && (a.completionId != null || b.completionId != null)) {
|
||||
return a.completionId != null && a.completionId == b.completionId
|
||||
}
|
||||
val ka = canonicalToolResultKey(a.content)
|
||||
val kb = canonicalToolResultKey(b.content)
|
||||
return ka != null && ka == kb
|
||||
if (a.role == MessageRole.ASSISTANT && a.content.isBlank() && b.content.isBlank() &&
|
||||
(a.reasoningText.isNotBlank() || b.reasoningText.isNotBlank())
|
||||
) {
|
||||
return a.reasoningText.isNotBlank() && a.reasoningText == b.reasoningText
|
||||
}
|
||||
if (a.role == MessageRole.TOOL) {
|
||||
if (a.toolCallId.isNotBlank() && b.toolCallId.isNotBlank()) return a.toolCallId == b.toolCallId
|
||||
val key = toolKey(a.content)
|
||||
return key != null && key == toolKey(b.content)
|
||||
}
|
||||
val ta = trimmed.getOrPut(a.content) { a.content.trim() }
|
||||
val tb = trimmed.getOrPut(b.content) { b.content.trim() }
|
||||
if (ta == tb) return true
|
||||
if (a.role == MessageRole.USER &&
|
||||
captions.getOrPut(ta) { stripAttachmentRefLines(ta) } ==
|
||||
captions.getOrPut(tb) { stripAttachmentRefLines(tb) }
|
||||
) {
|
||||
return true
|
||||
}
|
||||
return ta.length >= 40 && tb.length >= 40 && (tb.startsWith(ta) || ta.startsWith(tb))
|
||||
}
|
||||
val ta = a.content.trim()
|
||||
val tb = b.content.trim()
|
||||
if (ta == tb) return true
|
||||
// Attachment dedupe: when the user sends an image/file, the backend
|
||||
// persists an ENRICHED copy of the prompt — the real caption plus
|
||||
// `@image:`/`@file:` reference lines and a `[screenshot]` marker
|
||||
// (tui_gateway _build_image_ref_message / run_agent flattening). The
|
||||
// optimistic bubble carries the raw caption only, so the exact compare
|
||||
// above fails and the REST sync renders the same message twice. Compare
|
||||
// USER rows on the caption with injection lines stripped (both sides —
|
||||
// stripping a clean string is a no-op). USER-only so assistant text that
|
||||
// legitimately mentions these tokens is never collapsed.
|
||||
if (a.role == MessageRole.USER) {
|
||||
val ca = stripAttachmentRefLines(ta)
|
||||
val cb = stripAttachmentRefLines(tb)
|
||||
if (ca == cb) return true
|
||||
}
|
||||
|
||||
/** Consume each occurrence once, reserving exact IDs before considering content echoes. */
|
||||
internal fun matchTranscriptMessages(
|
||||
incoming: List<ChatMessage>,
|
||||
existing: List<ChatMessage>,
|
||||
comparison: TranscriptComparison = TranscriptComparison(),
|
||||
allowAssistantContentMatches: Boolean = true,
|
||||
): List<ChatMessage?> {
|
||||
val byId = existing.withIndex().associate { it.value.id to it.index }
|
||||
val byRestId =
|
||||
existing.withIndex().mapNotNull { (index, message) -> message.canonicalRestId?.let { it to index } }.toMap()
|
||||
val used = BooleanArray(existing.size)
|
||||
val matches = arrayOfNulls<ChatMessage>(incoming.size)
|
||||
incoming.forEachIndexed { index, message ->
|
||||
(byId[message.id] ?: message.canonicalRestId?.let { byRestId[it] })?.let { match ->
|
||||
if (!used[match]) {
|
||||
used[match] = true
|
||||
matches[index] = existing[match]
|
||||
}
|
||||
}
|
||||
}
|
||||
// Issue #842: a seal race can leave the live orphan a few trailing tokens
|
||||
// short of the streamed narration (the last delta was still in the
|
||||
// throttled buffer when tool.start sealed the message). The backend
|
||||
// persists the COMPLETE copy, so the orphan is a strict prefix of the
|
||||
// REST row. Accept prefix-covering only for substantial texts (>=40
|
||||
// chars) so a short reply can never be swallowed by a longer message
|
||||
// that merely starts with it.
|
||||
return ta.length >= 40 &&
|
||||
tb.length >= 40 &&
|
||||
(tb.startsWith(ta) || ta.startsWith(tb))
|
||||
val byRole = existing.indices.groupBy { existing[it].role }
|
||||
val byCall =
|
||||
existing.indices
|
||||
.filter { existing[it].toolCallId.isNotBlank() }
|
||||
.groupBy { existing[it].toolCallId }
|
||||
val byResult =
|
||||
existing.indices
|
||||
.filter { existing[it].role == MessageRole.TOOL }
|
||||
.groupBy { comparison.toolKey(existing[it].content) }
|
||||
val matchOrder =
|
||||
incoming.indices.filter { incoming[it].role != MessageRole.ASSISTANT } +
|
||||
incoming.indices.reversed().filter { incoming[it].role == MessageRole.ASSISTANT }
|
||||
matchOrder.forEach { index ->
|
||||
val message = incoming[index]
|
||||
if (matches[index] != null) return@forEach
|
||||
val candidates =
|
||||
if (message.role == MessageRole.TOOL) {
|
||||
val callMatches = byCall[message.toolCallId].orEmpty()
|
||||
callMatches + byResult[comparison.toolKey(message.content)].orEmpty()
|
||||
} else {
|
||||
// Prefix matching scans only candidates with the same role.
|
||||
byRole[message.role].orEmpty().let {
|
||||
if (message.role == MessageRole.ASSISTANT) it.asReversed() else it
|
||||
}
|
||||
}
|
||||
val match =
|
||||
candidates.firstOrNull { candidate ->
|
||||
val other = existing[candidate]
|
||||
!used[candidate] &&
|
||||
(
|
||||
allowAssistantContentMatches || message.role != MessageRole.ASSISTANT ||
|
||||
message.completionId != null || other.completionId != null
|
||||
) &&
|
||||
// Distinct IDs within the same source are separate occurrences, not echoes.
|
||||
((message.canonicalRestId == null) != (other.canonicalRestId == null)) &&
|
||||
comparison.same(message, other)
|
||||
}
|
||||
if (match != null) {
|
||||
used[match] = true
|
||||
matches[index] = existing[match]
|
||||
}
|
||||
}
|
||||
return matches.toList()
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -176,55 +248,154 @@ internal fun stripAttachmentRefLines(content: String): String =
|
||||
* same call twice. Drop the `rest-` copy whenever a WS copy of the same
|
||||
* logical message exists (issue #771).
|
||||
*/
|
||||
internal fun dedupeCachedMessages(messages: List<ChatMessage>): List<ChatMessage> {
|
||||
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 }
|
||||
internal fun dedupeCachedMessages(
|
||||
messages: List<ChatMessage>,
|
||||
confirmedOnly: Boolean = false,
|
||||
): List<ChatMessage> {
|
||||
val unique = messages.dedupeById()
|
||||
val rest = unique.filter { it.id.startsWith("rest-") }
|
||||
val live = unique.filterNot { it.id.startsWith("rest-") }
|
||||
if (rest.isEmpty() || live.isEmpty()) return unique
|
||||
val matches =
|
||||
matchTranscriptMessages(rest, live).mapIndexed { index, match ->
|
||||
match?.takeIf {
|
||||
!confirmedOnly || rest[index].canonicalRestId == it.canonicalRestId ||
|
||||
(rest[index].completionId != null && rest[index].completionId == it.completionId)
|
||||
}
|
||||
}
|
||||
val echoes =
|
||||
rest.indices
|
||||
.filter { matches[it] != null }
|
||||
.map { rest[it].id }
|
||||
.toSet()
|
||||
// #859: retain the REST identity when keeping a rich UUID row, including across cache pages.
|
||||
val aliases =
|
||||
rest.indices
|
||||
.mapNotNull { index ->
|
||||
matches[index]?.id?.let { it to rest[index] }
|
||||
}.toMap()
|
||||
return unique.filterNot { it.id in echoes }.map { message ->
|
||||
aliases[message.id]?.let {
|
||||
message.copy(restId = it.canonicalRestId, completionId = message.completionId ?: it.completionId)
|
||||
} ?: message
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
/** Cache pages enrich confirmed identities without changing keys already on screen (#859). */
|
||||
internal fun mergeCachedTranscriptPage(
|
||||
page: List<ChatMessage>,
|
||||
current: List<ChatMessage>,
|
||||
): List<ChatMessage> {
|
||||
val currentById = current.associateBy { it.id }
|
||||
// Legacy cached UUIDs may lack an alias. Restore a known alias before page-local matching
|
||||
// so an older identical REST row cannot claim that UUID again.
|
||||
val incoming =
|
||||
dedupeCachedMessages(
|
||||
page.map { message ->
|
||||
currentById[message.id]?.restId?.let { message.copy(restId = it) } ?: message
|
||||
},
|
||||
)
|
||||
val matches = matchTranscriptMessages(incoming, current)
|
||||
val replacements =
|
||||
incoming
|
||||
.mapIndexedNotNull { index, message ->
|
||||
val match = matches[index] ?: return@mapIndexedNotNull null
|
||||
val rich = if (match.id.startsWith("rest-") && !message.id.startsWith("rest-")) message else match
|
||||
match.id to
|
||||
rich.copy(
|
||||
id = match.id,
|
||||
restId = message.canonicalRestId ?: match.canonicalRestId,
|
||||
completionId = match.completionId ?: message.completionId,
|
||||
)
|
||||
}.toMap()
|
||||
return (
|
||||
current.map { replacements[it.id] ?: it } +
|
||||
incoming.filterIndexed { index, _ -> matches[index] == null }
|
||||
).dedupeById().inTranscriptOrder().reconcileReasoningRows()
|
||||
}
|
||||
|
||||
/** Merge one page with the current snapshot without consuming repeated results more than once. */
|
||||
internal fun mergeTranscriptWithLive(
|
||||
restMessages: List<ChatMessage>,
|
||||
currentMessages: List<ChatMessage>,
|
||||
chronological: Boolean = true,
|
||||
preserveLiveIds: Boolean = false,
|
||||
): List<ChatMessage> {
|
||||
val dedupedRest = restMessages.dedupeById()
|
||||
val dedupedCurrent = currentMessages.dedupeById()
|
||||
// User rows can carry live/cache-only metadata (for example whether the
|
||||
// bubble continues the active turn). Keep that richer copy when REST
|
||||
// returns the same logical row.
|
||||
val currentUsers = dedupedCurrent.filter { it.role == MessageRole.USER }.toMutableList()
|
||||
val mergedRest =
|
||||
dedupedRest.map { rest ->
|
||||
if (rest.role != MessageRole.USER) return@map rest
|
||||
val matchIndex =
|
||||
currentUsers.indexOfFirst { it.id == rest.id }.takeIf { it >= 0 }
|
||||
?: currentUsers.indexOfFirst { sameLogicalMessage(rest, it) }
|
||||
if (matchIndex >= 0) currentUsers.removeAt(matchIndex) else rest
|
||||
val incoming = restMessages.dedupeById()
|
||||
val current = currentMessages.dedupeById()
|
||||
val matches = matchTranscriptMessages(incoming, current, allowAssistantContentMatches = chronological)
|
||||
val consumed = matches.mapNotNull { it?.id }.toSet()
|
||||
val merged =
|
||||
incoming.mapIndexed { index, message ->
|
||||
val match = matches[index]
|
||||
// Keep local user metadata and stable IDs already used by the renderer.
|
||||
when {
|
||||
match?.role == MessageRole.USER -> {
|
||||
match.copy(
|
||||
restId = (message.canonicalRestId ?: match.canonicalRestId).takeUnless { it == match.id },
|
||||
)
|
||||
}
|
||||
|
||||
preserveLiveIds && match != null && !match.id.startsWith("rest-") -> {
|
||||
match.copy(
|
||||
restId = message.canonicalRestId ?: match.canonicalRestId,
|
||||
content = message.content,
|
||||
timestamp = message.timestamp,
|
||||
isStreaming = message.isStreaming,
|
||||
reasoningText = message.reasoningText.ifBlank { match.reasoningText },
|
||||
attachments = message.attachments ?: match.attachments,
|
||||
toolName = message.toolName ?: match.toolName,
|
||||
toolCallId = message.toolCallId.ifBlank { match.toolCallId },
|
||||
toolStatus = message.toolStatus ?: match.toolStatus,
|
||||
displayKind = message.displayKind ?: match.displayKind,
|
||||
tokenCount = message.tokenCount ?: match.tokenCount,
|
||||
completionId = message.completionId ?: match.completionId,
|
||||
)
|
||||
}
|
||||
|
||||
else -> {
|
||||
message.copy(completionId = message.completionId ?: match?.completionId)
|
||||
}
|
||||
}
|
||||
}
|
||||
val restIds = dedupedRest.map { it.id }.toSet()
|
||||
val liveTail =
|
||||
dedupedCurrent.filter { old ->
|
||||
old.id !in restIds && mergedRest.none { sameLogicalMessage(it, old) }
|
||||
val transcript =
|
||||
if (chronological) {
|
||||
current.filterNot { it.id in consumed } + merged
|
||||
} else {
|
||||
merged + current.filterNot { it.id in consumed }
|
||||
}
|
||||
if (liveTail.isEmpty()) return mergedRest.dedupeById()
|
||||
return (mergedRest + liveTail).dedupeById().sortedBy { it.timestamp }
|
||||
// #129: a mapped completion can confirm a cached REST row while its rich UUID
|
||||
// copy is also present. Fold that now-confirmed echo without changing the live key.
|
||||
// The page has already consumed its content matches; consuming another would
|
||||
// collapse a separate repeated occurrence. Only confirmed identities can fold here.
|
||||
return dedupeCachedMessages(transcript.inTranscriptOrder(), confirmedOnly = true).reconcileReasoningRows()
|
||||
}
|
||||
|
||||
/** Numeric canonical history first, then unconfirmed local insertion order; clocks never sort a transcript. */
|
||||
private fun List<ChatMessage>.inTranscriptOrder(): List<ChatMessage> =
|
||||
sortedWith(
|
||||
compareBy<ChatMessage> { it.canonicalOrder == null }
|
||||
.thenBy { it.canonicalOrder ?: it.localOrder ?: Long.MAX_VALUE },
|
||||
)
|
||||
|
||||
private val ChatMessage.canonicalOrder: Long?
|
||||
get() = if (localOrder != null && restId == null) null else canonicalRestId?.substringAfterLast('-')?.toLongOrNull()
|
||||
|
||||
/** Retain the canonical trace once, without copying it across a user/system/tool boundary. */
|
||||
private fun List<ChatMessage>.reconcileReasoningRows(): List<ChatMessage> =
|
||||
mapIndexed { index, message ->
|
||||
val previous = getOrNull(index - 1)
|
||||
if (previous?.role == MessageRole.ASSISTANT && previous.content.isBlank() &&
|
||||
previous.canonicalRestId != null && previous.reasoningText.isNotBlank() &&
|
||||
message.role == MessageRole.ASSISTANT && message.content.isNotBlank() &&
|
||||
message.canonicalRestId != null && message.reasoningText == previous.reasoningText
|
||||
) {
|
||||
message.copy(reasoningText = "")
|
||||
} else {
|
||||
message
|
||||
}
|
||||
}
|
||||
|
||||
/** Merge a REST page without matching it against already-settled transcript rows. */
|
||||
internal fun mergeIncrementalTranscriptPage(
|
||||
restMessages: List<ChatMessage>,
|
||||
@@ -234,7 +405,7 @@ internal fun mergeIncrementalTranscriptPage(
|
||||
): List<ChatMessage> {
|
||||
val settledEnd =
|
||||
currentMessages.indexOfLast { message ->
|
||||
serverMessageIndex(message.id, sessionId)?.let { it < offset } == true
|
||||
serverMessageIndex(message.canonicalRestId ?: message.id, sessionId)?.let { it < offset } == true
|
||||
}
|
||||
return (
|
||||
currentMessages.take(settledEnd + 1) +
|
||||
|
||||
@@ -45,6 +45,8 @@ import kotlinx.coroutines.flow.combine
|
||||
import kotlinx.coroutines.flow.stateIn
|
||||
import kotlinx.coroutines.flow.update
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.coroutines.withContext
|
||||
import kotlinx.serialization.json.JsonElement
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
@@ -374,6 +376,7 @@ class ChatViewModel(
|
||||
slashUsageStore: SlashUsageStore = SlashUsageStore(application.applicationContext),
|
||||
searchDispatcher: kotlinx.coroutines.CoroutineDispatcher = kotlinx.coroutines.Dispatchers.Default,
|
||||
private val ioDispatcher: kotlinx.coroutines.CoroutineDispatcher = kotlinx.coroutines.Dispatchers.IO,
|
||||
private val historyDispatcher: kotlinx.coroutines.CoroutineDispatcher = searchDispatcher,
|
||||
) : AndroidViewModel(application) {
|
||||
constructor(application: Application) : this(application, startCleanup = true)
|
||||
|
||||
@@ -443,6 +446,15 @@ class ChatViewModel(
|
||||
*/
|
||||
private var latestPaging = false
|
||||
private var isSyncingMessages = false
|
||||
private var cacheCursor: ChatPersistenceRepository.Cursor? = null
|
||||
private var cacheHasOlder = false
|
||||
private var cacheLoaded = false
|
||||
private var serverHasOlder = false
|
||||
private var cacheJob: Job? = null
|
||||
private var hydrationJob: Job? = null
|
||||
private var olderJob: Job? = null
|
||||
private var syncJob: Job? = null
|
||||
private val historyFetchMutex = Mutex()
|
||||
|
||||
// ── Session resume recovery (desktop parity) ────────────────────────
|
||||
// Bounded auto-retry with exponential backoff, mirroring the desktop's
|
||||
@@ -642,7 +654,19 @@ class ChatViewModel(
|
||||
activeHydrationRequestSequence = ++hydrationRequestSequence
|
||||
// Preserve metadata so late results and errors are rejected as stale.
|
||||
sessionGeneration++
|
||||
_uiState.update { it.copy(isSessionReady = false) }
|
||||
// #129: invalidate and cancel paged work together, including its busy flags.
|
||||
hydrationJob?.cancel()
|
||||
olderJob?.cancel()
|
||||
syncJob?.cancel()
|
||||
cacheJob?.cancel()
|
||||
isSyncingMessages = false
|
||||
cancelResumeRetry()
|
||||
_uiState.update {
|
||||
it.copy(isSessionReady = false, isLoadingOlder = false, isResumeRetrying = false)
|
||||
}
|
||||
_uiState.value.currentSessionId?.let { sessionId ->
|
||||
if (!cacheLoaded) loadCachedMessages(sessionId, sessionGeneration)
|
||||
}
|
||||
// Fail any in-flight awaited RPCs so callers don't hang
|
||||
// across the disconnect (delegated to HermesWsClient, issue #526).
|
||||
wsClient.rejectAllPending()
|
||||
@@ -2904,151 +2928,201 @@ class ChatViewModel(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* CPU work never runs in StateFlow.update: it may retry. If a WS event or another
|
||||
* page changes the transcript during compute, recompute against that newer snapshot.
|
||||
*/
|
||||
private suspend fun mergeHistoryPage(
|
||||
isCurrent: () -> Boolean,
|
||||
cached: Boolean = false,
|
||||
prepend: Boolean = false,
|
||||
mapPage: (List<ChatMessage>) -> List<ChatMessage>,
|
||||
): List<ChatMessage>? {
|
||||
while (isCurrent()) {
|
||||
val snapshot = _uiState.value
|
||||
val computed =
|
||||
withContext(historyDispatcher) {
|
||||
val page = mapPage(snapshot.messages)
|
||||
val merged =
|
||||
if (cached) {
|
||||
mergeCachedTranscriptPage(page, snapshot.messages)
|
||||
} else {
|
||||
mergeTranscriptWithLive(
|
||||
page,
|
||||
snapshot.messages,
|
||||
chronological = !prepend,
|
||||
preserveLiveIds = true,
|
||||
)
|
||||
}
|
||||
val stableMessages = if (merged == snapshot.messages) snapshot.messages else merged
|
||||
Triple(page, stableMessages, hydrateTodosFromMessages(merged).ifEmpty { snapshot.todos })
|
||||
}
|
||||
if (!isCurrent()) return null
|
||||
var applied = false
|
||||
_uiState.update { current ->
|
||||
applied = isCurrent() && current.messages === snapshot.messages && current.todos === snapshot.todos
|
||||
if (applied) current.copy(messages = computed.second, todos = computed.third) else current
|
||||
}
|
||||
if (applied) return computed.first
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
private fun publishHistoryAvailability() {
|
||||
_uiState.update { it.copy(hasOlderMessages = cacheHasOlder || serverHasOlder) }
|
||||
}
|
||||
|
||||
private suspend fun readCachedPage(
|
||||
sessionId: String,
|
||||
generation: Long,
|
||||
isCurrent: () -> Boolean = { isCurrentSessionRequest(sessionId, generation) },
|
||||
): Boolean {
|
||||
val before = cacheCursor
|
||||
// Room suspend queries own their IO executor; entity mapping/dedup runs on Default.
|
||||
val page = withContext(historyDispatcher) { repo.loadPage(sessionId, before, MESSAGE_PAGE_SIZE) }
|
||||
val valid = { isCurrent() && cacheCursor == before }
|
||||
if (!valid()) return false
|
||||
mergeHistoryPage(valid, cached = true) { page.messages } ?: return false
|
||||
if (!valid()) return false
|
||||
cacheCursor = page.cursor
|
||||
cacheHasOlder = page.hasOlder
|
||||
cacheLoaded = true
|
||||
publishHistoryAvailability()
|
||||
return true
|
||||
}
|
||||
|
||||
private fun loadCachedMessages(
|
||||
sessionId: String,
|
||||
generation: Long,
|
||||
): Job =
|
||||
viewModelScope.launch(ioDispatcher) {
|
||||
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
|
||||
// by a stale cache read that loses the race).
|
||||
if (isCurrentSessionRequest(sessionId, generation) &&
|
||||
state.messages.isEmpty() &&
|
||||
cachedMessages.isNotEmpty()
|
||||
) {
|
||||
state.copy(
|
||||
messages = cachedMessages,
|
||||
isLoading = false,
|
||||
todos = hydrateTodosFromMessages(cachedMessages),
|
||||
)
|
||||
} else {
|
||||
state
|
||||
): Job {
|
||||
cacheJob?.cancel()
|
||||
return viewModelScope
|
||||
.launch {
|
||||
try {
|
||||
readCachedPage(sessionId, generation)
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
if (isCurrentSessionRequest(sessionId, generation)) {
|
||||
// The first read can be retried by the same older-history action.
|
||||
cacheHasOlder = true
|
||||
publishHistoryAvailability()
|
||||
_uiState.update { it.copy(errorMessage = "Failed to load cached messages: ${e.message}") }
|
||||
}
|
||||
} finally {
|
||||
if (isCurrentSessionRequest(sessionId, generation) && _uiState.value.messages.isNotEmpty()) {
|
||||
_uiState.update { it.copy(isLoading = false) }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}.also { cacheJob = it }
|
||||
}
|
||||
|
||||
private fun loadSessionMessages(
|
||||
sessionId: String,
|
||||
generation: Long,
|
||||
) {
|
||||
if (activeHydrationRequestSequence != 0L && cacheJob?.isActive == true) {
|
||||
cacheJob?.cancel()
|
||||
if (!cacheLoaded) loadCachedMessages(sessionId, generation)
|
||||
}
|
||||
val requestSequence = ++hydrationRequestSequence
|
||||
activeHydrationRequestSequence = requestSequence
|
||||
viewModelScope.launch {
|
||||
// Initial page: ask for the NEWEST page directly (order=latest —
|
||||
// offset is measured back from the newest message, page returned
|
||||
// chronologically; verified in hermes_state.py get_messages).
|
||||
// No count-based anchor, so a stale session-list message_count can
|
||||
// no longer land the page at the wrong position (issue #859).
|
||||
// Legacy backends without the `order` param ignore it and echo no
|
||||
// `pagination` — detect that and fall back to the count-based
|
||||
// anchor so nothing regresses.
|
||||
val latestResult = fetchMessagePage(sessionId, 0, MESSAGE_PAGE_SIZE, order = "latest")
|
||||
if (!isCurrentHydration(sessionId, generation, requestSequence)) return@launch
|
||||
val (result, requestedOffset) =
|
||||
if (latestResult is NetworkResult.Success && latestResult.data.pagination?.order == "latest") {
|
||||
latestPaging = true
|
||||
latestResult to 0
|
||||
} else if (latestResult is NetworkResult.Success) {
|
||||
latestPaging = false
|
||||
val messageCount = fetchServerMessageCount(sessionId, generation, requestSequence)
|
||||
if (!isCurrentHydration(sessionId, generation, requestSequence)) return@launch
|
||||
val offset = (messageCount - MESSAGE_PAGE_SIZE).coerceAtLeast(0)
|
||||
fetchMessagePage(sessionId, offset, MESSAGE_PAGE_SIZE) to offset
|
||||
} else {
|
||||
latestPaging = false
|
||||
latestResult to 0
|
||||
}
|
||||
if (!isCurrentHydration(sessionId, generation, requestSequence)) return@launch
|
||||
when (result) {
|
||||
is NetworkResult.Success -> {
|
||||
// REST 200 — the gateway has the row for this session.
|
||||
sessionHasServerPresence = true
|
||||
val serverOffset = result.data.pagination?.offset ?: result.data.offset ?: requestedOffset
|
||||
val chatMessages =
|
||||
mapServerMessages(
|
||||
sessionId,
|
||||
result.data.messages.orEmpty(),
|
||||
serverOffset,
|
||||
latestPaging,
|
||||
_uiState.value.messages,
|
||||
context = getApplication(),
|
||||
)
|
||||
loadedMessageOffset = serverOffset
|
||||
withContext(ioDispatcher) {
|
||||
repo.persistMessages(chatMessages, sessionId)
|
||||
}
|
||||
if (!isCurrentHydration(sessionId, generation, requestSequence)) return@launch
|
||||
_uiState.update { state ->
|
||||
if (!isCurrentHydration(sessionId, generation, requestSequence)) 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)
|
||||
val hasOlder =
|
||||
if (latestPaging) {
|
||||
// Newest-anchored: older messages exist iff the
|
||||
// page came back FULL (a short page means we hit
|
||||
// the oldest boundary).
|
||||
val returned = result.data.pagination?.returned ?: chatMessages.size
|
||||
returned >= MESSAGE_PAGE_SIZE && chatMessages.isNotEmpty()
|
||||
hydrationJob?.cancel()
|
||||
olderJob?.cancel()
|
||||
syncJob?.cancel()
|
||||
isSyncingMessages = false
|
||||
_uiState.update { it.copy(isLoadingOlder = false) }
|
||||
val valid = { isCurrentHydration(sessionId, generation, requestSequence) }
|
||||
hydrationJob =
|
||||
viewModelScope.launch {
|
||||
try {
|
||||
val latestResult = fetchMessagePage(sessionId, 0, MESSAGE_PAGE_SIZE, order = "latest")
|
||||
if (!valid()) return@launch
|
||||
// Backends without stable latest IDs use the legacy absolute-position protocol.
|
||||
val useLatest =
|
||||
latestResult is NetworkResult.Success &&
|
||||
latestResult.data.pagination?.order == "latest" &&
|
||||
latestResult.data.messages.all { it.id != null }
|
||||
val (result, requestedOffset) =
|
||||
if (useLatest || latestResult is NetworkResult.Failure) {
|
||||
latestResult to 0
|
||||
} else {
|
||||
val count = fetchServerMessageCount(sessionId, generation, requestSequence)
|
||||
if (!valid()) return@launch
|
||||
val offset = (count - MESSAGE_PAGE_SIZE).coerceAtLeast(0)
|
||||
fetchMessagePage(sessionId, offset, MESSAGE_PAGE_SIZE) to offset
|
||||
}
|
||||
if (!valid()) return@launch
|
||||
when (result) {
|
||||
is NetworkResult.Success -> {
|
||||
val serverOffset = result.data.pagination?.offset ?: result.data.offset ?: requestedOffset
|
||||
val raw = result.data.messages
|
||||
val page =
|
||||
mergeHistoryPage(valid) { current ->
|
||||
mapServerMessages(
|
||||
sessionId,
|
||||
raw,
|
||||
serverOffset,
|
||||
useLatest,
|
||||
current,
|
||||
context = getApplication(),
|
||||
)
|
||||
} ?: return@launch
|
||||
persistHistoryPage(page, sessionId)
|
||||
if (!valid()) return@launch
|
||||
latestPaging = useLatest
|
||||
// Cursor counts RAW rows, including hidden reasoning/placeholder rows.
|
||||
loadedMessageOffset = if (useLatest) serverOffset + raw.size else serverOffset
|
||||
serverHasOlder = raw.isNotEmpty() &&
|
||||
if (useLatest) raw.size >= MESSAGE_PAGE_SIZE else serverOffset > 0
|
||||
sessionHasServerPresence = true
|
||||
publishHistoryAvailability()
|
||||
hydratedGeneration = generation
|
||||
finishResumeWhenHydrated(generation)
|
||||
}
|
||||
|
||||
is NetworkResult.Failure -> {
|
||||
val error = result.error.message
|
||||
if (error.contains("404", ignoreCase = true) ||
|
||||
error.contains("not found", ignoreCase = true)
|
||||
) {
|
||||
serverHasOlder = false
|
||||
publishHistoryAvailability()
|
||||
hydratedGeneration = generation
|
||||
finishResumeWhenHydrated(generation)
|
||||
} else {
|
||||
serverOffset > 0 && chatMessages.isNotEmpty()
|
||||
if (hydratedGeneration == generation) hydratedGeneration = -1L
|
||||
handleResumeFailure(sessionId, generation, "Failed to load messages: $error")
|
||||
}
|
||||
state.copy(
|
||||
messages = merged,
|
||||
todos = hydrateTodosFromMessages(merged),
|
||||
isLoading = false,
|
||||
hasOlderMessages = hasOlder,
|
||||
isLoadingOlder = false,
|
||||
)
|
||||
}
|
||||
hydratedGeneration = generation
|
||||
finishResumeWhenHydrated(generation)
|
||||
}
|
||||
|
||||
is NetworkResult.Failure -> {
|
||||
val errorMsg = result.error.message
|
||||
val is404 =
|
||||
errorMsg.contains("404", ignoreCase = true) ||
|
||||
errorMsg.contains("not found", ignoreCase = true)
|
||||
if (is404) {
|
||||
// A 404 from GET /api/sessions/{id}/messages indicates the session has no
|
||||
// persisted rows in state.db (e.g. newly created/lazy unprompted session, or empty history).
|
||||
// It is NOT a fatal error: keep the session open with empty transcript, clear loading,
|
||||
// and mark hydrated cleanly.
|
||||
_uiState.update {
|
||||
if (!isCurrentHydration(sessionId, generation, requestSequence)) return@update it
|
||||
it.copy(
|
||||
isLoading = false,
|
||||
isLoadingOlder = false,
|
||||
hasOlderMessages = false,
|
||||
)
|
||||
}
|
||||
hydratedGeneration = generation
|
||||
finishResumeWhenHydrated(generation)
|
||||
return@launch
|
||||
}
|
||||
if (hydratedGeneration == generation) hydratedGeneration = -1L
|
||||
_uiState.update {
|
||||
if (!isCurrentHydration(sessionId, generation, requestSequence)) return@update it
|
||||
it.copy(
|
||||
isLoading = false,
|
||||
isLoadingOlder = false,
|
||||
)
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
if (valid()) {
|
||||
hydratedGeneration = -1L
|
||||
handleResumeFailure(sessionId, generation, "Failed to load messages: ${e.message}")
|
||||
}
|
||||
// Route the transcript failure through the bounded resume
|
||||
// retry (desktop parity) instead of a one-shot snackbar —
|
||||
// a transient backend/network blip recovers on its own,
|
||||
// and a persistent failure ends in an explicit Retry.
|
||||
handleResumeFailure(
|
||||
sessionId,
|
||||
generation,
|
||||
"Failed to load messages: ${result.error.message}",
|
||||
)
|
||||
} finally {
|
||||
if (valid()) _uiState.update { it.copy(isLoading = false) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun persistHistoryPage(
|
||||
page: List<ChatMessage>,
|
||||
sessionId: String,
|
||||
) {
|
||||
// A mapped page can reuse a live WS message. Never overwrite its newer persisted
|
||||
// version with the snapshot used for mapping; WS owns persistence of those IDs.
|
||||
val pageIds = page.mapNotNull { it.canonicalRestId }.toSet()
|
||||
val aliases = _uiState.value.messages.filter { it.restId in pageIds && !it.id.startsWith("rest-") }
|
||||
withContext(historyDispatcher) {
|
||||
repo.persistMessages(
|
||||
page.mapNotNull { message -> message.canonicalRestId?.let { message.copy(id = it, restId = null) } },
|
||||
sessionId,
|
||||
)
|
||||
repo.confirmIdentities(aliases, sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3096,6 +3170,15 @@ class ChatViewModel(
|
||||
runtimeSessionId = null
|
||||
pendingInitialPrompt = null
|
||||
ActiveSessionHolder.clear()
|
||||
cacheJob?.cancel()
|
||||
hydrationJob?.cancel()
|
||||
olderJob?.cancel()
|
||||
syncJob?.cancel()
|
||||
activeHydrationRequestSequence = 0L
|
||||
cacheCursor = null
|
||||
cacheHasOlder = false
|
||||
cacheLoaded = false
|
||||
serverHasOlder = false
|
||||
loadedMessageOffset = 0
|
||||
latestPaging = false
|
||||
isSyncingMessages = false
|
||||
@@ -3372,222 +3455,150 @@ class ChatViewModel(
|
||||
fun loadOlderMessages() {
|
||||
val state = _uiState.value
|
||||
val sessionId = state.currentSessionId ?: return
|
||||
if (!state.hasOlderMessages || state.isLoadingOlder || olderJob?.isActive == true) return
|
||||
if (cacheJob?.isActive == true) return
|
||||
val fromCache = cacheHasOlder || !cacheLoaded
|
||||
if (!fromCache && (hydrationJob?.isActive == true || isSyncingMessages || !serverHasOlder)) return
|
||||
val generation = sessionGeneration
|
||||
if (!state.hasOlderMessages || state.isLoadingOlder) return
|
||||
if (!latestPaging && loadedMessageOffset <= 0) return
|
||||
val oldOffset = loadedMessageOffset
|
||||
// latest: offsets count BACK from the newest message, so older pages go
|
||||
// UP; legacy: absolute offsets go DOWN toward 0 (issue #859).
|
||||
val newOffset =
|
||||
if (latestPaging) {
|
||||
oldOffset + MESSAGE_PAGE_SIZE
|
||||
} else {
|
||||
(oldOffset - MESSAGE_PAGE_SIZE).coerceAtLeast(0)
|
||||
}
|
||||
// Legacy: the final page can be short; requesting a full page at a
|
||||
// clamped offset OVERLAPS already-loaded rows (duplicate stable keys
|
||||
// crash LazyColumn), so size the request to the real gap. Latest:
|
||||
// pages are disjoint from-end ranges, so a full page never overlaps.
|
||||
val limit = if (latestPaging) MESSAGE_PAGE_SIZE else oldOffset - newOffset
|
||||
val requestSequence = activeHydrationRequestSequence
|
||||
val valid = { isCurrentHydration(sessionId, generation, requestSequence) }
|
||||
_uiState.update { it.copy(isLoadingOlder = true) }
|
||||
viewModelScope.launch {
|
||||
val result =
|
||||
fetchMessagePage(
|
||||
sessionId,
|
||||
newOffset,
|
||||
limit,
|
||||
order = if (latestPaging) "latest" else null,
|
||||
)
|
||||
when (result) {
|
||||
is NetworkResult.Success -> {
|
||||
if (!isCurrentSessionRequest(sessionId, generation)) return@launch
|
||||
val returnedOffset = result.data.pagination?.offset ?: result.data.offset ?: newOffset
|
||||
val older =
|
||||
mapServerMessages(
|
||||
sessionId,
|
||||
result.data.messages.orEmpty(),
|
||||
returnedOffset,
|
||||
latestPaging,
|
||||
_uiState.value.messages,
|
||||
isPagingOlder = true,
|
||||
)
|
||||
loadedMessageOffset = returnedOffset
|
||||
withContext(ioDispatcher) { repo.persistMessages(older, sessionId) }
|
||||
_uiState.update { current ->
|
||||
if (!isCurrentSessionRequest(sessionId, generation)) return@update current
|
||||
val hasOlder =
|
||||
if (latestPaging) {
|
||||
// Full page = more older messages behind it; a
|
||||
// short (or empty) page is the oldest boundary.
|
||||
val returned = result.data.pagination?.returned ?: older.size
|
||||
returned >= limit && older.isNotEmpty()
|
||||
} else {
|
||||
returnedOffset < oldOffset && older.isNotEmpty() && returnedOffset > 0
|
||||
olderJob =
|
||||
viewModelScope.launch {
|
||||
try {
|
||||
while (cacheHasOlder || !cacheLoaded) {
|
||||
val beforeMessages = _uiState.value.messages
|
||||
if (!readCachedPage(sessionId, generation, valid)) return@launch
|
||||
val afterMessages = _uiState.value.messages
|
||||
val addedRows =
|
||||
withContext(historyDispatcher) {
|
||||
val previousIds = beforeMessages.mapTo(HashSet()) { it.id }
|
||||
afterMessages.any { it.id !in previousIds }
|
||||
}
|
||||
current.copy(
|
||||
messages = (older + current.messages).distinctBy { it.id },
|
||||
isLoadingOlder = false,
|
||||
hasOlderMessages = hasOlder,
|
||||
)
|
||||
if (!valid()) return@launch
|
||||
if (addedRows) return@launch
|
||||
// Hydration persists canonical rows behind the local cache cursor. Consume
|
||||
// those already-visible echoes without spending another older-history action.
|
||||
}
|
||||
}
|
||||
if (hydrationJob?.isActive == true || isSyncingMessages || !serverHasOlder) return@launch
|
||||
// Cache reads may suspend through hydration/sync; use the current server cursor.
|
||||
val useLatest = latestPaging
|
||||
val oldOffset = loadedMessageOffset
|
||||
val offset = if (useLatest) oldOffset else (oldOffset - MESSAGE_PAGE_SIZE).coerceAtLeast(0)
|
||||
val limit = if (useLatest) MESSAGE_PAGE_SIZE else oldOffset - offset
|
||||
if (limit <= 0) return@launch
|
||||
val result = fetchMessagePage(sessionId, offset, limit, order = if (useLatest) "latest" else null)
|
||||
if (!valid()) return@launch
|
||||
when (result) {
|
||||
is NetworkResult.Success -> {
|
||||
val returnedOffset = result.data.pagination?.offset ?: result.data.offset ?: offset
|
||||
val raw = result.data.messages
|
||||
val page =
|
||||
mergeHistoryPage(valid, prepend = true) { current ->
|
||||
mapServerMessages(
|
||||
sessionId,
|
||||
raw,
|
||||
returnedOffset,
|
||||
useLatest,
|
||||
current,
|
||||
isPagingOlder = true,
|
||||
context = getApplication(),
|
||||
)
|
||||
} ?: return@launch
|
||||
persistHistoryPage(page, sessionId)
|
||||
if (!valid()) return@launch
|
||||
loadedMessageOffset = if (useLatest) returnedOffset + raw.size else returnedOffset
|
||||
serverHasOlder = raw.isNotEmpty() &&
|
||||
if (useLatest) {
|
||||
loadedMessageOffset > oldOffset && raw.size >= limit
|
||||
} else {
|
||||
returnedOffset < oldOffset && returnedOffset > 0
|
||||
}
|
||||
publishHistoryAvailability()
|
||||
}
|
||||
|
||||
is NetworkResult.Failure -> {
|
||||
_uiState.update {
|
||||
if (isCurrentSessionRequest(sessionId, generation)) {
|
||||
it.copy(isLoadingOlder = false)
|
||||
} else {
|
||||
it
|
||||
is NetworkResult.Failure -> {
|
||||
_uiState.update {
|
||||
it.copy(
|
||||
errorMessage = "Failed to load older messages: ${result.error.message}",
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
if (valid()) {
|
||||
_uiState.update {
|
||||
it.copy(
|
||||
errorMessage = "Failed to load older messages: ${e.message}",
|
||||
)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (valid()) _uiState.update { it.copy(isLoadingOlder = false) }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun syncCurrentSession() {
|
||||
// Issue #840: a session created but never prompted has no server row
|
||||
// yet — the REST transcript 404s and the resume-retry machinery spins
|
||||
// forever (visible in logcat as repeated /messages 404s). MessageStart
|
||||
// flips the flag once the first prompt persists the row.
|
||||
if (!sessionHasServerPresence) return
|
||||
val state = _uiState.value
|
||||
val sessionId = state.currentSessionId ?: return
|
||||
val generation = sessionGeneration
|
||||
if (isSyncingMessages || state.isLoading || state.isLoadingOlder || state.isAgentTyping ||
|
||||
_streamingState.value.streamingMessage != null
|
||||
if (isSyncingMessages || hydrationJob?.isActive == true || state.isLoading || state.isLoadingOlder ||
|
||||
state.isAgentTyping || _streamingState.value.streamingMessage != null
|
||||
) {
|
||||
return
|
||||
}
|
||||
val generation = sessionGeneration
|
||||
val requestSequence = activeHydrationRequestSequence
|
||||
val valid = { isCurrentHydration(sessionId, generation, requestSequence) }
|
||||
val useLatest = latestPaging
|
||||
val nextOffset =
|
||||
if (latestPaging) {
|
||||
// Newest-anchored paging can't compute an absolute
|
||||
// "after my last row" offset (from-end offsets shift as the
|
||||
// transcript grows), so refetch the newest page; the logical
|
||||
// merge below keeps existing copies and only adds new rows
|
||||
// (issue #859).
|
||||
if (useLatest) {
|
||||
0
|
||||
} else {
|
||||
state.messages
|
||||
.mapNotNull { serverMessageIndex(it.id, sessionId) }
|
||||
.mapNotNull { serverMessageIndex(it.canonicalRestId ?: it.id, sessionId) }
|
||||
.maxOrNull()
|
||||
?.plus(1)
|
||||
?: loadedMessageOffset
|
||||
}
|
||||
isSyncingMessages = true
|
||||
viewModelScope.launch {
|
||||
try {
|
||||
val result =
|
||||
fetchMessagePage(
|
||||
sessionId,
|
||||
nextOffset,
|
||||
MESSAGE_PAGE_SIZE,
|
||||
order = if (latestPaging) "latest" else null,
|
||||
)
|
||||
when (result) {
|
||||
is NetworkResult.Success -> {
|
||||
if (!isCurrentSessionRequest(sessionId, generation)) return@launch
|
||||
val incoming =
|
||||
mapServerMessages(
|
||||
sessionId,
|
||||
result.data.messages.orEmpty(),
|
||||
nextOffset,
|
||||
latestPaging,
|
||||
_uiState.value.messages,
|
||||
isPagingOlder = true,
|
||||
)
|
||||
if (incoming.isEmpty()) return@launch
|
||||
withContext(ioDispatcher) { repo.persistMessages(incoming, sessionId) }
|
||||
_uiState.update { current ->
|
||||
if (!isCurrentSessionRequest(sessionId, generation)) 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: MutableList<ChatMessage?> = incoming.toMutableList()
|
||||
val incomingById = HashMap<String, Int>(unmatchedIncoming.size)
|
||||
for (i in unmatchedIncoming.indices) {
|
||||
val id = unmatchedIncoming[i]?.id
|
||||
if (id != null && !incomingById.containsKey(id)) {
|
||||
incomingById[id] = i
|
||||
}
|
||||
}
|
||||
|
||||
val mergedList = mutableListOf<ChatMessage>()
|
||||
|
||||
for (existing in current.messages) {
|
||||
val existingServerIndex = serverMessageIndex(existing.id, sessionId)
|
||||
if (existingServerIndex != null) {
|
||||
val matchIdx = incomingById[existing.id]
|
||||
if (matchIdx != null && unmatchedIncoming[matchIdx] != null) {
|
||||
mergedList.add(unmatchedIncoming[matchIdx]!!)
|
||||
unmatchedIncoming[matchIdx] = null
|
||||
} else if (latestPaging) {
|
||||
// Newest-anchored paging re-keys rows by
|
||||
// from-end position, so a transcript that
|
||||
// grew since the last fetch shifted ids —
|
||||
// match by logical content and keep the
|
||||
// existing copy (issue #859).
|
||||
val logicalIdx =
|
||||
unmatchedIncoming.indexOfFirst { inc ->
|
||||
inc != null && sameLogicalMessage(inc, existing)
|
||||
}
|
||||
if (logicalIdx >= 0) {
|
||||
mergedList.add(existing)
|
||||
unmatchedIncoming[logicalIdx] = null
|
||||
} else {
|
||||
mergedList.add(existing)
|
||||
}
|
||||
} else {
|
||||
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 != null && sameLogicalMessage(inc, existing)
|
||||
}
|
||||
if (matchIdx >= 0) {
|
||||
// Prefer the WS copy (richer payload,
|
||||
// real tool name) when available.
|
||||
mergedList.add(existing)
|
||||
unmatchedIncoming[matchIdx] = null
|
||||
} else {
|
||||
// No REST counterpart (server hasn't
|
||||
// persisted yet) — KEEP the WS message.
|
||||
mergedList.add(existing)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (inc in unmatchedIncoming) {
|
||||
if (inc != null) {
|
||||
mergedList.add(inc)
|
||||
}
|
||||
}
|
||||
val merged = mergedList.distinctBy { it.id }
|
||||
if (sameMessages(current.messages, merged)) {
|
||||
current
|
||||
} else {
|
||||
current.copy(messages = merged)
|
||||
}
|
||||
}
|
||||
syncJob =
|
||||
viewModelScope.launch {
|
||||
try {
|
||||
val result =
|
||||
fetchMessagePage(sessionId, nextOffset, MESSAGE_PAGE_SIZE, if (useLatest) "latest" else null)
|
||||
if (!valid()) return@launch
|
||||
if (result is NetworkResult.Success) {
|
||||
val offset = result.data.pagination?.offset ?: result.data.offset ?: nextOffset
|
||||
val page =
|
||||
mergeHistoryPage(valid) { current ->
|
||||
mapServerMessages(
|
||||
sessionId,
|
||||
result.data.messages,
|
||||
offset,
|
||||
useLatest,
|
||||
current,
|
||||
// Sync fetches recent replies, so confirm live completion identities.
|
||||
isPagingOlder = false,
|
||||
context = getApplication(),
|
||||
)
|
||||
} ?: return@launch
|
||||
persistHistoryPage(page, sessionId)
|
||||
// Never derive the older cursor from displayed rows or reset it to this latest page.
|
||||
// Append-only growth shifts from-end offsets toward newer rows: the next older
|
||||
// request may overlap, but stable IDs remove echoes without skipping any history.
|
||||
}
|
||||
|
||||
is NetworkResult.Failure -> {}
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
if (valid()) _uiState.update { it.copy(errorMessage = "Failed to sync messages: ${e.message}") }
|
||||
} finally {
|
||||
if (valid()) isSyncingMessages = false
|
||||
}
|
||||
} finally {
|
||||
if (generation == sessionGeneration) isSyncingMessages = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal fun onModelSwitchInitiated() {
|
||||
@@ -3926,15 +3937,17 @@ class ChatViewModel(
|
||||
offset: Int,
|
||||
limit: Int,
|
||||
order: String? = null,
|
||||
) = withContext(ioDispatcher) {
|
||||
safeApiCall {
|
||||
ApiClient.hermesApi.getSessionMessages(
|
||||
sessionId = sessionId,
|
||||
limit = limit,
|
||||
offset = offset,
|
||||
includeCompacted = true,
|
||||
order = order,
|
||||
)
|
||||
) = historyFetchMutex.withLock {
|
||||
withContext(ioDispatcher) {
|
||||
safeApiCall {
|
||||
ApiClient.hermesApi.getSessionMessages(
|
||||
sessionId = sessionId,
|
||||
limit = limit,
|
||||
offset = offset,
|
||||
includeCompacted = true,
|
||||
order = order,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
package com.m57.hermescontrol.ui.chat.components
|
||||
|
||||
/** One older-page request per deliberate upward gesture, including an overscroll at index zero. */
|
||||
internal class ChatHistoryPrefetch {
|
||||
private var upwardPixels = 0f
|
||||
private var requested = false
|
||||
|
||||
fun onScroll(
|
||||
firstVisibleIndex: Int,
|
||||
deltaY: Float,
|
||||
userInput: Boolean,
|
||||
canLoad: Boolean,
|
||||
): Boolean {
|
||||
if (userInput && deltaY < 0f) upwardPixels = 0f
|
||||
if (deltaY <= 0f || !canLoad || requested) return false
|
||||
// A fling retains the drag's intent until scrolling stops. Programmatic
|
||||
// scrolling without an upward user gesture must never load history.
|
||||
if (!userInput && upwardPixels == 0f) return false
|
||||
upwardPixels += deltaY
|
||||
if (firstVisibleIndex > 25 || upwardPixels < 24f) return false
|
||||
requested = true
|
||||
return true
|
||||
}
|
||||
|
||||
fun endGesture() {
|
||||
upwardPixels = 0f
|
||||
requested = false
|
||||
}
|
||||
}
|
||||
@@ -33,9 +33,9 @@ import kotlinx.coroutines.yield
|
||||
* 5. **Unread indicator** — when follow is paused, [pendingCount] accumulates the
|
||||
* number of tail updates; shows on the FAB. Tapping the FAB resumes following
|
||||
* and clears the count.
|
||||
* 6. **Paging anchor preservation** — capture the first visible item + offset
|
||||
* with [captureAnchor] before prepending older history, then
|
||||
* [restoreAnchor] after insertion so the same content stays under the eye.
|
||||
* 6. **Paging anchors** — the full-bleed renderer preserves the live lazy row
|
||||
* key and offset when older history is inserted; this controller owns no
|
||||
* fetch-time anchor or animated compensation.
|
||||
* 7. **Serialized scroll commands** — every scroll (send, FAB, session switch,
|
||||
* search navigation, auto-follow) is launched from [scope] so animations
|
||||
* don't compete.
|
||||
@@ -189,30 +189,17 @@ class ChatScrollController(
|
||||
}
|
||||
}
|
||||
|
||||
/** An explicit history gesture must not be undone by a short list's bottom-follow. */
|
||||
fun pauseFollowing() {
|
||||
isFollowingBottom = false
|
||||
}
|
||||
|
||||
/** FAB tap: resume following + clear unread. */
|
||||
fun resumeFollowing() = jumpToBottom(animated = true)
|
||||
|
||||
/** True when the FAB should be visible (not following bottom + content exists). */
|
||||
fun showFab(contentPresent: Boolean): Boolean = !isFollowingBottom && contentPresent
|
||||
|
||||
/**
|
||||
* Current pixel scroll offset of the first visible item — captured before
|
||||
* prepending older history so it can be restored after insertion.
|
||||
*/
|
||||
fun captureAnchorOffset(): Int = listState.firstVisibleItemScrollOffset
|
||||
|
||||
/**
|
||||
* Restore the viewport to a specific item (resolved by id in the caller, so
|
||||
* the restore is robust even if streaming tokens arrive during the page
|
||||
* insert). Keeps the reader's eye on the same content after prepend.
|
||||
*/
|
||||
fun scrollToItem(
|
||||
index: Int,
|
||||
offset: Int,
|
||||
) {
|
||||
scope.launch { listState.scrollToItem(index, offset) }
|
||||
}
|
||||
|
||||
/**
|
||||
* Navigate to a search match (serialized through [scope]).
|
||||
*
|
||||
@@ -258,7 +245,7 @@ fun tailContentKey(
|
||||
clarifyRequest: Any?,
|
||||
): Any =
|
||||
listOf(
|
||||
messages.size,
|
||||
messages.lastOrNull(),
|
||||
streamingMessage?.hashCode() ?: 0,
|
||||
isThinking,
|
||||
subagentIndicators.size,
|
||||
|
||||
@@ -4,7 +4,6 @@ import androidx.compose.foundation.layout.Box
|
||||
import androidx.compose.foundation.layout.Column
|
||||
import androidx.compose.foundation.layout.PaddingValues
|
||||
import androidx.compose.foundation.layout.fillMaxSize
|
||||
import androidx.compose.foundation.layout.fillMaxWidth
|
||||
import androidx.compose.foundation.layout.padding
|
||||
import androidx.compose.foundation.layout.size
|
||||
import androidx.compose.foundation.lazy.LazyColumn
|
||||
@@ -12,9 +11,17 @@ import androidx.compose.material3.CircularProgressIndicator
|
||||
import androidx.compose.runtime.Composable
|
||||
import androidx.compose.runtime.CompositionLocalProvider
|
||||
import androidx.compose.runtime.LaunchedEffect
|
||||
import androidx.compose.runtime.SideEffect
|
||||
import androidx.compose.runtime.mutableStateOf
|
||||
import androidx.compose.runtime.remember
|
||||
import androidx.compose.runtime.rememberUpdatedState
|
||||
import androidx.compose.runtime.snapshotFlow
|
||||
import androidx.compose.ui.Alignment
|
||||
import androidx.compose.ui.Modifier
|
||||
import androidx.compose.ui.geometry.Offset
|
||||
import androidx.compose.ui.input.nestedscroll.NestedScrollConnection
|
||||
import androidx.compose.ui.input.nestedscroll.NestedScrollSource
|
||||
import androidx.compose.ui.input.nestedscroll.nestedScroll
|
||||
import androidx.compose.ui.platform.LocalDensity
|
||||
import androidx.compose.ui.res.stringResource
|
||||
import androidx.compose.ui.unit.Density
|
||||
@@ -32,6 +39,7 @@ import com.m57.hermescontrol.ui.chat.ToolCallDivider
|
||||
import com.m57.hermescontrol.ui.chat.VaultCodePromptUi
|
||||
import com.m57.hermescontrol.ui.chat.VaultSaveLoginPromptUi
|
||||
import com.m57.hermescontrol.ui.chat.VaultUnlockPromptUi
|
||||
import com.m57.hermescontrol.ui.chat.components.ChatHistoryPrefetch
|
||||
import com.m57.hermescontrol.ui.chat.components.ChatScrollController
|
||||
import com.m57.hermescontrol.ui.chat.components.ClarifyBubble
|
||||
import com.m57.hermescontrol.ui.chat.components.ReasoningCard
|
||||
@@ -93,6 +101,9 @@ fun FullBleedChatList(
|
||||
savingAttachmentPath: String? = null,
|
||||
openingAttachmentPath: String? = null,
|
||||
onImageClick: (ImageViewerModel) -> Unit = {},
|
||||
hasOlderMessages: Boolean = false,
|
||||
pagingSessionId: String? = null,
|
||||
onLoadOlder: () -> Unit = viewModel::loadOlderMessages,
|
||||
) {
|
||||
if (messages.isEmpty() && !isLoading && !isAgentTyping) {
|
||||
Box(
|
||||
@@ -105,19 +116,133 @@ fun FullBleedChatList(
|
||||
)
|
||||
}
|
||||
} else {
|
||||
val toolMilestones = remember(messages) { toolCallMilestones(messages) }
|
||||
// Keep only the incoming older prefix out of layout during a drag/fling.
|
||||
// The input remains authoritative: overlap updates, new messages and streaming
|
||||
// stay live, and a session change resets the boundary immediately.
|
||||
val previousFirstId = remember(listState, pagingSessionId) { mutableStateOf(messages.firstOrNull()?.id) }
|
||||
val firstRenderedId = previousFirstId.value
|
||||
val olderPrefixSize =
|
||||
remember(messages, firstRenderedId) {
|
||||
if (firstRenderedId != null && firstRenderedId != messages.firstOrNull()?.id) {
|
||||
messages.indexOfFirst { it.id == firstRenderedId }.coerceAtLeast(0)
|
||||
} else {
|
||||
0
|
||||
}
|
||||
}
|
||||
val hiddenPrefixSize = if (olderPrefixSize > 0 && listState.isScrollInProgress) olderPrefixSize else 0
|
||||
val renderedMessages =
|
||||
remember(messages, hiddenPrefixSize) {
|
||||
if (hiddenPrefixSize > 0) messages.subList(hiddenPrefixSize, messages.size) else messages
|
||||
}
|
||||
val toolMilestones = remember(renderedMessages) { toolCallMilestones(renderedMessages) }
|
||||
val settledTurns = remember(renderedMessages) { groupIntoTurns(renderedMessages) }
|
||||
val settledIds = remember(renderedMessages) { renderedMessages.mapTo(HashSet()) { it.id } }
|
||||
val turns =
|
||||
remember(messages, streamingState.streamingMessage) {
|
||||
groupIntoTurnsWithStreaming(messages, streamingState.streamingMessage)
|
||||
remember(settledTurns, streamingState.streamingMessage) {
|
||||
appendStreamingTurn(settledTurns, settledIds, streamingState.streamingMessage)
|
||||
}
|
||||
val agentStatus = deriveAgentStatus(isAgentTyping, streamingState, messages)
|
||||
// messageId → LazyColumn item index. The lazy list has EXTRA items
|
||||
// vs the message list (reasoning hoists, tool rows), so search-match
|
||||
// scrolling must resolve the lazy index, not use the message index.
|
||||
val lazyIndexById =
|
||||
remember(turns, isLoadingOlder) {
|
||||
messageIdToLazyIndex(turns, leadingItems = if (isLoadingOlder) 1 else 0)
|
||||
val agentStatus = deriveAgentStatus(isAgentTyping, streamingState, renderedMessages)
|
||||
// A single ordered definition supplies both emitted tail rows and anchor keys.
|
||||
val tailItems =
|
||||
buildMap<String, @Composable () -> Unit> {
|
||||
agentStatus?.let { status ->
|
||||
put("agent_status") {
|
||||
AgentStatusIndicator(status = status)
|
||||
}
|
||||
}
|
||||
|
||||
// Clarify bubble — rendered at the very bottom
|
||||
if (clarifyRequest != null) {
|
||||
put("clarify_bubble") {
|
||||
ClarifyBubble(
|
||||
clarifyRequest = clarifyRequest,
|
||||
onRespondSingle = { option -> onRespondClarify?.invoke(option) },
|
||||
onRespondBatch = { answers -> onRespondClarifyBatch?.invoke(answers) },
|
||||
onDismiss = { onDismissClarify?.invoke() },
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (vaultUnlockPrompt != null) {
|
||||
put("vault_unlock_card") {
|
||||
VaultUnlockCard(
|
||||
prompt = vaultUnlockPrompt,
|
||||
onConfirm = { password -> onRespondVaultUnlock?.invoke(password) },
|
||||
onDismiss = { onDismissVaultUnlock?.invoke() },
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (vaultSaveLoginPrompt != null) {
|
||||
put("vault_save_login_card") {
|
||||
VaultSaveLoginCard(
|
||||
prompt = vaultSaveLoginPrompt,
|
||||
onConfirm = { identifier, password ->
|
||||
onRespondVaultSaveLogin?.invoke(identifier, password)
|
||||
},
|
||||
onDismiss = { onDismissVaultSaveLogin?.invoke() },
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (vaultCodePrompt != null) {
|
||||
put("vault_code_card") {
|
||||
VaultCodeCard(
|
||||
prompt = vaultCodePrompt,
|
||||
onConfirm = { code -> onRespondVaultCode?.invoke(code) },
|
||||
onDismiss = { onDismissVaultCode?.invoke() },
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
// custom23 prepend regression: a 150-row page can move the anchor beyond
|
||||
// Compose's nearest-key lookup window. Capture the LIVE layout at insertion,
|
||||
// not when fetching starts, and request its new index before the next measure.
|
||||
// A staged prefix only reaches here at idle, so this cannot stop its drag/fling.
|
||||
SideEffect {
|
||||
val oldFirstId = previousFirstId.value
|
||||
val firstId = renderedMessages.firstOrNull()?.id
|
||||
if (oldFirstId != null && oldFirstId != firstId && renderedMessages.any { it.id == oldFirstId }) {
|
||||
val firstVisibleIndex = listState.firstVisibleItemIndex
|
||||
val anchor = listState.layoutInfo.visibleItemsInfo.firstOrNull { it.index == firstVisibleIndex }
|
||||
if (anchor != null) {
|
||||
val newIndex = (fullBleedItemKeys(turns) + tailItems.keys).indexOf(anchor.key)
|
||||
if (newIndex >= 0 && newIndex != anchor.index) {
|
||||
listState.requestScrollToItem(newIndex, listState.firstVisibleItemScrollOffset)
|
||||
}
|
||||
}
|
||||
}
|
||||
previousFirstId.value = firstId
|
||||
}
|
||||
val prefetch = remember(pagingSessionId) { ChatHistoryPrefetch() }
|
||||
val canLoad = rememberUpdatedState(hasOlderMessages && !isLoadingOlder && !isLoading)
|
||||
val loadOlder = rememberUpdatedState(onLoadOlder)
|
||||
val prefetchConnection =
|
||||
remember(listState, prefetch, scrollController) {
|
||||
object : NestedScrollConnection {
|
||||
override fun onPreScroll(
|
||||
available: Offset,
|
||||
source: NestedScrollSource,
|
||||
): Offset {
|
||||
if (prefetch.onScroll(
|
||||
firstVisibleIndex = listState.firstVisibleItemIndex,
|
||||
deltaY = available.y,
|
||||
userInput = source == NestedScrollSource.UserInput,
|
||||
canLoad = canLoad.value,
|
||||
)
|
||||
) {
|
||||
scrollController.pauseFollowing()
|
||||
loadOlder.value()
|
||||
}
|
||||
return Offset.Zero
|
||||
}
|
||||
}
|
||||
}
|
||||
LaunchedEffect(listState, prefetch) {
|
||||
snapshotFlow { listState.isScrollInProgress }.collect { scrolling ->
|
||||
if (!scrolling) prefetch.endGesture()
|
||||
}
|
||||
}
|
||||
|
||||
// Scroll the current search match into view, word-focused. Lives here
|
||||
// (not in ChatLifecycleEffects) because only this composable knows
|
||||
@@ -128,6 +253,7 @@ fun FullBleedChatList(
|
||||
searchState.currentIndex,
|
||||
searchState.matchIndices,
|
||||
searchState.matchOffsets,
|
||||
renderedMessages.firstOrNull()?.id,
|
||||
) {
|
||||
if (searchState.isActive &&
|
||||
searchState.currentIndex >= 0 &&
|
||||
@@ -135,6 +261,9 @@ fun FullBleedChatList(
|
||||
) {
|
||||
val messageIndex = searchState.matchIndices[searchState.currentIndex]
|
||||
if (messageIndex < 0 || messageIndex >= messages.size) return@LaunchedEffect
|
||||
// Search indices refer to incoming messages; resolve by id into the
|
||||
// rendered rows. A hit in a staged prefix is retried when it becomes visible.
|
||||
val lazyIndexById = messageIdToLazyIndex(turns)
|
||||
val lazyIndex = lazyIndexById[messages[messageIndex].id] ?: return@LaunchedEffect
|
||||
val contentOffset = searchState.matchOffsets.getOrElse(searchState.currentIndex) { 0 }
|
||||
val contentLength = messages[messageIndex].content.length
|
||||
@@ -153,202 +282,163 @@ fun FullBleedChatList(
|
||||
}
|
||||
|
||||
CompositionLocalProvider(LocalDensity provides chatDensity) {
|
||||
LazyColumn(
|
||||
state = listState,
|
||||
modifier = Modifier.fillMaxSize(),
|
||||
contentPadding = PaddingValues(vertical = 8.dp),
|
||||
) {
|
||||
if (isLoadingOlder) {
|
||||
item(key = "loading-older") {
|
||||
Box(
|
||||
modifier = Modifier.fillMaxWidth().padding(12.dp),
|
||||
contentAlignment = Alignment.Center,
|
||||
) {
|
||||
CircularProgressIndicator(modifier = Modifier.size(24.dp))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
turns.forEach { turn ->
|
||||
when (turn) {
|
||||
is ChatTurn.User -> {
|
||||
// Eager captures: item lambda reads these at
|
||||
// composition time (lazy), so capture now.
|
||||
val userMessage = turn.message
|
||||
item(key = "user-${userMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 12.dp)) {
|
||||
renderChatBubble(
|
||||
message = userMessage,
|
||||
searchQuery = if (searchState.isActive) searchState.query else "",
|
||||
isCurrentMatch =
|
||||
searchState.currentMatchId != null &&
|
||||
searchState.currentMatchId == userMessage.id,
|
||||
onOpenAttachment = viewModel::openAttachment,
|
||||
onSaveAttachment = onSaveAttachment,
|
||||
savingAttachmentPath = savingAttachmentPath,
|
||||
openingAttachmentPath = openingAttachmentPath,
|
||||
onImageClick = onImageClick,
|
||||
messageStatsEnabled = messageStatsEnabled,
|
||||
showUserMessageTokens = showUserMessageTokens,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
is ChatTurn.Agent -> {
|
||||
// Reasoning hoist: the turn's reasoning renders at the
|
||||
// TOP of the turn — above tool rows — so thinking
|
||||
// leads, then the tool work, then the answer. The
|
||||
// matching prose entry renders without its own card.
|
||||
val turnReasoning =
|
||||
turn.entries
|
||||
.filterIsInstance<AgentEntry.Prose>()
|
||||
.firstOrNull { it.message.reasoningText.isNotBlank() }
|
||||
if (turnReasoning != null) {
|
||||
val reasoning = turnReasoning.message
|
||||
item(key = "reasoning-${reasoning.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 6.dp)) {
|
||||
ReasoningCard(
|
||||
reasoningText = reasoning.reasoningText,
|
||||
isStreaming = reasoning.isStreaming,
|
||||
Box(Modifier.fillMaxSize()) {
|
||||
LazyColumn(
|
||||
state = listState,
|
||||
modifier = Modifier.fillMaxSize().nestedScroll(prefetchConnection),
|
||||
contentPadding = PaddingValues(vertical = 8.dp),
|
||||
) {
|
||||
turns.forEach { turn ->
|
||||
when (turn) {
|
||||
is ChatTurn.User -> {
|
||||
// Eager captures: item lambda reads these at
|
||||
// composition time (lazy), so capture now.
|
||||
val userMessage = turn.message
|
||||
item(key = "user-${userMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 12.dp)) {
|
||||
renderChatBubble(
|
||||
message = userMessage,
|
||||
searchQuery = if (searchState.isActive) searchState.query else "",
|
||||
isCurrentMatch =
|
||||
searchState.currentMatchId != null &&
|
||||
searchState.currentMatchId == userMessage.id,
|
||||
onOpenAttachment = viewModel::openAttachment,
|
||||
onSaveAttachment = onSaveAttachment,
|
||||
savingAttachmentPath = savingAttachmentPath,
|
||||
openingAttachmentPath = openingAttachmentPath,
|
||||
onImageClick = onImageClick,
|
||||
messageStatsEnabled = messageStatsEnabled,
|
||||
showUserMessageTokens = showUserMessageTokens,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
turn.entries.forEach { entry ->
|
||||
when (entry) {
|
||||
is AgentEntry.Prose -> {
|
||||
val proseMessage = entry.message
|
||||
val hoistedReasoning =
|
||||
turnReasoning != null &&
|
||||
proseMessage.id == turnReasoning.message.id
|
||||
if (proseMessage.hasVisibleAgentContent()) {
|
||||
item(key = "prose-${proseMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 12.dp)) {
|
||||
if (proseMessage.isStreaming && typingEffectEnabled) {
|
||||
StreamingFullBleedWithTypingEffect(
|
||||
streaming = proseMessage,
|
||||
typingDelayMs = typingEffectDelayMs,
|
||||
showReasoning = !hoistedReasoning,
|
||||
)
|
||||
|
||||
is ChatTurn.Agent -> {
|
||||
// Reasoning hoist: the turn's reasoning renders at the
|
||||
// TOP of the turn — above tool rows — so thinking
|
||||
// leads, then the tool work, then the answer. The
|
||||
// matching prose entry renders without its own card.
|
||||
val turnReasoning =
|
||||
turn.entries
|
||||
.filterIsInstance<AgentEntry.Prose>()
|
||||
.firstOrNull { it.message.reasoningText.isNotBlank() }
|
||||
if (turnReasoning != null) {
|
||||
val reasoning = turnReasoning.message
|
||||
item(key = "reasoning-${reasoning.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 6.dp)) {
|
||||
ReasoningCard(
|
||||
reasoningText = reasoning.reasoningText,
|
||||
isStreaming = reasoning.isStreaming,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
turn.entries.forEach { entry ->
|
||||
when (entry) {
|
||||
is AgentEntry.Prose -> {
|
||||
val proseMessage = entry.message
|
||||
val hoistedReasoning =
|
||||
turnReasoning != null &&
|
||||
proseMessage.id == turnReasoning.message.id
|
||||
if (!proseMessage.hasVisibleAgentContent() && !hoistedReasoning &&
|
||||
proseMessage.reasoningText.isNotBlank()
|
||||
) {
|
||||
item(key = "reasoning-${proseMessage.id}") {
|
||||
ReasoningCard(
|
||||
reasoningText = proseMessage.reasoningText,
|
||||
isStreaming = proseMessage.isStreaming,
|
||||
)
|
||||
}
|
||||
}
|
||||
if (proseMessage.hasVisibleAgentContent()) {
|
||||
item(key = "prose-${proseMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 12.dp)) {
|
||||
if (proseMessage.isStreaming && typingEffectEnabled) {
|
||||
StreamingFullBleedWithTypingEffect(
|
||||
streaming = proseMessage,
|
||||
typingDelayMs = typingEffectDelayMs,
|
||||
showReasoning = !hoistedReasoning,
|
||||
)
|
||||
} else {
|
||||
FullBleedAgentMessage(
|
||||
message = proseMessage,
|
||||
// Highlight only bubbles that actually contain a match —
|
||||
// the rest skip the highlight scan entirely.
|
||||
searchQuery =
|
||||
if (searchState.isActive &&
|
||||
proseMessage.id in searchState.matchedIds
|
||||
) {
|
||||
searchState.query
|
||||
} else {
|
||||
""
|
||||
},
|
||||
isCurrentMatch =
|
||||
searchState.currentMatchId != null &&
|
||||
searchState.currentMatchId == proseMessage.id,
|
||||
showReasoning = !hoistedReasoning,
|
||||
onOpenAttachment = viewModel::openAttachment,
|
||||
onSaveAttachment = onSaveAttachment,
|
||||
savingAttachmentPath = savingAttachmentPath,
|
||||
openingAttachmentPath = openingAttachmentPath,
|
||||
canSaveAttachment = savingAttachmentPath == null,
|
||||
onImageClick = onImageClick,
|
||||
messageStatsEnabled = messageStatsEnabled,
|
||||
showAssistantMessageTokens = showAssistantMessageTokens,
|
||||
showTokensPerSecond = showTokensPerSecond,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
is AgentEntry.ToolRow -> {
|
||||
val toolMessage = entry.message
|
||||
val milestone = toolMilestones[toolMessage.id]
|
||||
item(key = "tool-${toolMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 6.dp)) {
|
||||
FullBleedToolRow(toolMessage)
|
||||
milestone?.let { count ->
|
||||
ToolCallDivider(count = count, maxPerTurn = maxToolCallsPerTurn)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
is AgentEntry.SystemEvent -> {
|
||||
val sysMessage = entry.message
|
||||
item(key = "sys-${sysMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 6.dp)) {
|
||||
if (sysMessage.displayKind != null) {
|
||||
// Timeline marker (issue #904):
|
||||
// model/personality switches and
|
||||
// auto-continues render as a chip.
|
||||
TimelineMarkerChip(message = sysMessage)
|
||||
} else {
|
||||
FullBleedAgentMessage(
|
||||
message = proseMessage,
|
||||
// Highlight only bubbles that actually contain a match —
|
||||
// the rest skip the highlight scan entirely.
|
||||
searchQuery =
|
||||
if (searchState.isActive &&
|
||||
proseMessage.id in searchState.matchedIds
|
||||
) {
|
||||
searchState.query
|
||||
} else {
|
||||
""
|
||||
},
|
||||
isCurrentMatch =
|
||||
searchState.currentMatchId != null &&
|
||||
searchState.currentMatchId == proseMessage.id,
|
||||
showReasoning = !hoistedReasoning,
|
||||
onOpenAttachment = viewModel::openAttachment,
|
||||
onSaveAttachment = onSaveAttachment,
|
||||
savingAttachmentPath = savingAttachmentPath,
|
||||
openingAttachmentPath = openingAttachmentPath,
|
||||
canSaveAttachment = savingAttachmentPath == null,
|
||||
onImageClick = onImageClick,
|
||||
messageStatsEnabled = messageStatsEnabled,
|
||||
showAssistantMessageTokens = showAssistantMessageTokens,
|
||||
showTokensPerSecond = showTokensPerSecond,
|
||||
FullBleedSystemEvent(
|
||||
message = sysMessage,
|
||||
onRespondApproval = viewModel::respondToApproval,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
is AgentEntry.ToolRow -> {
|
||||
val toolMessage = entry.message
|
||||
val milestone = toolMilestones[toolMessage.id]
|
||||
item(key = "tool-${toolMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 6.dp)) {
|
||||
FullBleedToolRow(toolMessage)
|
||||
milestone?.let { count ->
|
||||
ToolCallDivider(count = count, maxPerTurn = maxToolCallsPerTurn)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
is AgentEntry.SystemEvent -> {
|
||||
val sysMessage = entry.message
|
||||
item(key = "sys-${sysMessage.id}") {
|
||||
Column(modifier = Modifier.padding(bottom = 6.dp)) {
|
||||
if (sysMessage.displayKind != null) {
|
||||
// Timeline marker (issue #904):
|
||||
// model/personality switches and
|
||||
// auto-continues render as a chip.
|
||||
TimelineMarkerChip(message = sysMessage)
|
||||
} else {
|
||||
FullBleedSystemEvent(
|
||||
message = sysMessage,
|
||||
onRespondApproval = viewModel::respondToApproval,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
agentStatus?.let { status ->
|
||||
item(key = "agent_status") {
|
||||
AgentStatusIndicator(status = status)
|
||||
tailItems.forEach { (key, content) ->
|
||||
item(key = key) { content() }
|
||||
}
|
||||
}
|
||||
|
||||
// Clarify bubble — rendered at the very bottom
|
||||
if (clarifyRequest != null) {
|
||||
item(key = "clarify_bubble") {
|
||||
ClarifyBubble(
|
||||
clarifyRequest = clarifyRequest,
|
||||
onRespondSingle = { option -> onRespondClarify?.invoke(option) },
|
||||
onRespondBatch = { answers -> onRespondClarifyBatch?.invoke(answers) },
|
||||
onDismiss = { onDismissClarify?.invoke() },
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (vaultUnlockPrompt != null) {
|
||||
item(key = "vault_unlock_card") {
|
||||
VaultUnlockCard(
|
||||
prompt = vaultUnlockPrompt,
|
||||
onConfirm = { password -> onRespondVaultUnlock?.invoke(password) },
|
||||
onDismiss = { onDismissVaultUnlock?.invoke() },
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (vaultSaveLoginPrompt != null) {
|
||||
item(key = "vault_save_login_card") {
|
||||
VaultSaveLoginCard(
|
||||
prompt = vaultSaveLoginPrompt,
|
||||
onConfirm = { identifier, password ->
|
||||
onRespondVaultSaveLogin?.invoke(identifier, password)
|
||||
},
|
||||
onDismiss = { onDismissVaultSaveLogin?.invoke() },
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (vaultCodePrompt != null) {
|
||||
item(key = "vault_code_card") {
|
||||
VaultCodeCard(
|
||||
prompt = vaultCodePrompt,
|
||||
onConfirm = { code -> onRespondVaultCode?.invoke(code) },
|
||||
onDismiss = { onDismissVaultCode?.invoke() },
|
||||
)
|
||||
}
|
||||
// Loading history must never become the list's first visible anchor.
|
||||
if (isLoadingOlder) {
|
||||
CircularProgressIndicator(
|
||||
modifier = Modifier.align(Alignment.TopCenter).padding(12.dp).size(24.dp),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -128,19 +128,22 @@ fun groupIntoTurns(messages: List<ChatMessage>): List<ChatTurn> {
|
||||
fun groupIntoTurnsWithStreaming(
|
||||
messages: List<ChatMessage>,
|
||||
streamingMessage: ChatMessage?,
|
||||
): List<ChatTurn> = appendStreamingTurn(groupIntoTurns(messages), messages.mapTo(HashSet()) { it.id }, streamingMessage)
|
||||
|
||||
/** Reuse every settled turn; only the last agent turn needs a changing streaming entry. */
|
||||
internal fun appendStreamingTurn(
|
||||
settled: List<ChatTurn>,
|
||||
settledIds: Set<String>,
|
||||
streamingMessage: ChatMessage?,
|
||||
): List<ChatTurn> {
|
||||
if (streamingMessage == null || messages.any { it.id == streamingMessage.id }) {
|
||||
return groupIntoTurns(messages)
|
||||
}
|
||||
val turns = groupIntoTurns(messages).toMutableList()
|
||||
if (streamingMessage == null || streamingMessage.id in settledIds) return settled
|
||||
val prose = AgentEntry.Prose(streamingMessage)
|
||||
val last = turns.lastOrNull()
|
||||
if (last is ChatTurn.Agent) {
|
||||
turns[turns.lastIndex] = ChatTurn.Agent(last.entries + prose)
|
||||
val last = settled.lastOrNull()
|
||||
return if (last is ChatTurn.Agent) {
|
||||
settled.dropLast(1) + ChatTurn.Agent(last.entries + prose)
|
||||
} else {
|
||||
turns += ChatTurn.Agent(listOf(prose))
|
||||
settled + ChatTurn.Agent(listOf(prose))
|
||||
}
|
||||
return turns
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -156,8 +159,9 @@ fun groupIntoTurnsWithStreaming(
|
||||
*
|
||||
* Mirrors the item emission order in [FullBleedChatList] exactly:
|
||||
* user turn → 1 item; agent turn → optional reasoning item, then one item per
|
||||
* visible entry. [leadingItems] accounts for fixed items emitted before the
|
||||
* turns (e.g. the `loading-older` spinner when paging).
|
||||
* visible entry, including later reasoning-only rows at their own positions.
|
||||
* [leadingItems] accounts for fixed items emitted before the
|
||||
* turns. The history spinner is an overlay and consumes no index.
|
||||
*
|
||||
* @return messageId → LazyColumn item index of the message's content item
|
||||
* (prose items for agent messages; user items for user messages).
|
||||
@@ -165,48 +169,54 @@ fun groupIntoTurnsWithStreaming(
|
||||
fun messageIdToLazyIndex(
|
||||
turns: List<ChatTurn>,
|
||||
leadingItems: Int = 0,
|
||||
): Map<String, Int> {
|
||||
val map = mutableMapOf<String, Int>()
|
||||
var itemIndex = leadingItems
|
||||
turns.forEach { turn ->
|
||||
when (turn) {
|
||||
is ChatTurn.User -> {
|
||||
map[turn.message.id] = itemIndex
|
||||
itemIndex++
|
||||
): Map<String, Int> =
|
||||
fullBleedItemKeys(turns)
|
||||
.mapIndexedNotNull { index, key ->
|
||||
if (key.startsWith("user-") || key.startsWith("prose-")) {
|
||||
key.substringAfter('-') to index + leadingItems
|
||||
} else {
|
||||
null
|
||||
}
|
||||
}.toMap()
|
||||
|
||||
is ChatTurn.Agent -> {
|
||||
val hasReasoning =
|
||||
turn.entries
|
||||
.filterIsInstance<AgentEntry.Prose>()
|
||||
.any { it.message.reasoningText.isNotBlank() }
|
||||
if (hasReasoning) {
|
||||
// The reasoning-<id> hoist item occupies one slot.
|
||||
itemIndex++
|
||||
/** Lazy row identities, including hoisted reasoning and grouped tool/system entries. */
|
||||
internal fun fullBleedItemKeys(turns: List<ChatTurn>): List<String> =
|
||||
buildList {
|
||||
turns.forEach { turn ->
|
||||
when (turn) {
|
||||
is ChatTurn.User -> {
|
||||
add("user-${turn.message.id}")
|
||||
}
|
||||
turn.entries.forEach { entry ->
|
||||
when (entry) {
|
||||
is AgentEntry.Prose -> {
|
||||
if (entry.message.hasVisibleAgentContent()) {
|
||||
map[entry.message.id] = itemIndex
|
||||
itemIndex++
|
||||
|
||||
is ChatTurn.Agent -> {
|
||||
val hoisted =
|
||||
turn.entries
|
||||
.filterIsInstance<AgentEntry.Prose>()
|
||||
.firstOrNull { it.message.reasoningText.isNotBlank() }
|
||||
hoisted?.let { add("reasoning-${it.message.id}") }
|
||||
turn.entries.forEach { entry ->
|
||||
when (entry) {
|
||||
is AgentEntry.Prose -> {
|
||||
if (entry.message.hasVisibleAgentContent()) {
|
||||
add("prose-${entry.message.id}")
|
||||
} else if (entry != hoisted && entry.message.reasoningText.isNotBlank()) {
|
||||
add("reasoning-${entry.message.id}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
is AgentEntry.ToolRow -> {
|
||||
itemIndex++
|
||||
}
|
||||
is AgentEntry.ToolRow -> {
|
||||
add("tool-${entry.message.id}")
|
||||
}
|
||||
|
||||
is AgentEntry.SystemEvent -> {
|
||||
itemIndex++
|
||||
is AgentEntry.SystemEvent -> {
|
||||
add("sys-${entry.message.id}")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return map
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the message id of the CURRENT search match once, so per-item
|
||||
|
||||
Reference in new issue
Block a user