feat(#1285): reconcile pending/streamed turns with persisted receipts (#1299)

This commit is contained in:
m57 authored and GitHub committed 2026-09-26 23:12:22 +03:00
1 parent 7ae97a1099
commit 5486856bd0
10 files changed
+392 -5

No files matched your search

@@ -107,6 +107,12 @@ data class ChatMessage(
* Transient; UNKNOWN provenance and the persisted row remain unchanged.
*/
val isRestoredUnconfirmed: Boolean = false,
/**
* Gateway SQLite row id (#1285): from REST `id`, the `prompt.submit` `user_row_id`
* receipt, or `persisted_turn.final_assistant_row_id`. Null means unproven, never rejected.
* Scoped to the owning profile store; transient.
*/
val serverRowId: Long? = null,
)
/** Cached REST rows already carry their canonical identity in the persisted primary key. */
@@ -0,0 +1,30 @@
package com.m57.hermescontrol.ui.chat
/**
* `message.complete` `persisted_turn` (hermes-agent `PersistedTurn`, #1285). Only the anchors Mobile
* can bind without inference are kept; `row_ids` for intermediate rows would need positional
* guessing. A missing id is unproven, never a rejection, and only [complete] permits retiring the
* whole local turn. Row ids are scoped to the owning profile store.
*/
internal data class PersistedTurn(
val complete: Boolean,
val userRowId: Long? = null,
val finalAssistantRowId: Long? = null,
)
/** Decoded JSON numbers may arrive as Int, Long, or Double; booleans and non-positive ids are not ids. */
internal fun positiveRowId(value: Any?): Long? =
when (value) {
is Int, is Long -> (value as Number).toLong()
is Double -> value.takeIf { it % 1.0 == 0.0 }?.toLong()
else -> null
}?.takeIf { it > 0L }
internal fun parsePersistedTurn(payload: Map<*, *>?): PersistedTurn? {
val raw = payload?.get("persisted_turn") as? Map<*, *> ?: return null
return PersistedTurn(
complete = raw["complete"] == true,
userRowId = positiveRowId(raw["user_row_id"]),
finalAssistantRowId = positiveRowId(raw["final_assistant_row_id"]),
)
}
@@ -21,6 +21,8 @@ data class PendingSend(
val attempts: Int = 0,
/** Legacy source URIs are retained for an explicit user retry, never auto-dispatched. */
val requiresAttachmentRecovery: Boolean = false,
/** `prompt.submit` `user_row_id` receipt (#1285); null means unproven, not rejected. */
val userRowId: Long? = null,
)
@Serializable
@@ -259,6 +259,7 @@ internal fun mapServerMessages(
displayKind = msg.display_kind,
tokenCount = tokenCount,
completionId = completionId,
serverRowId = msg.id?.toLong()?.takeIf { it > 0L },
),
)
}
@@ -274,12 +275,13 @@ internal fun mapServerMessages(
// A canonical tool-result row settles a cached tool.start, not the reverse.
local.copy(
restId = message.canonicalRestId,
serverRowId = message.serverRowId,
content = message.content,
toolStatus = ToolStatus.COMPLETED,
isHistoricalCache = false,
)
} else {
local?.copy(restId = message.canonicalRestId) ?: message
local?.copy(restId = message.canonicalRestId, serverRowId = message.serverRowId) ?: message
}
}
}
@@ -151,6 +151,8 @@ internal class TranscriptComparison(
b: ChatMessage,
): Boolean {
if (a.role != b.role) return false
// #1285: two proven row ids decide identity; a missing one is unproven, so fall through.
if (a.serverRowId != null && b.serverRowId != null) return a.serverRowId == b.serverRowId
val aRestId = a.canonicalRestId
val bRestId = b.canonicalRestId
if (aRestId != null && bRestId != null) return aRestId == bRestId
@@ -213,10 +215,20 @@ internal fun matchTranscriptMessages(
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 byRowId =
existing
.withIndex()
.mapNotNull { (index, message) ->
message.serverRowId?.let { (message.role to 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 ->
(
byId[message.id]
?: message.canonicalRestId?.let { byRestId[it] }
?: message.serverRowId?.let { byRowId[message.role to it] }
)?.let { match ->
if (!used[match]) {
used[match] = true
matches[index] = existing[match]
@@ -333,6 +345,7 @@ internal fun dedupeCachedMessages(
aliases[message.id]?.let {
message.copy(
restId = it.canonicalRestId,
serverRowId = message.serverRowId ?: it.serverRowId,
completionId = message.completionId ?: it.completionId,
displayKind = message.displayKind ?: it.displayKind,
isRestoredUnconfirmed = false,
@@ -402,6 +415,7 @@ internal fun mergeCachedTranscriptPage(
id = match.id,
content = preservedContent ?: rich.content,
restId = match.canonicalRestId ?: message.canonicalRestId,
serverRowId = match.serverRowId ?: message.serverRowId,
completionId = match.completionId ?: message.completionId,
displayKind = match.displayKind ?: message.displayKind,
isRestoredUnconfirmed =
@@ -439,6 +453,7 @@ internal fun mergeTranscriptWithLive(
match?.role == MessageRole.USER -> {
match.copy(
restId = (message.canonicalRestId ?: match.canonicalRestId).takeUnless { it == match.id },
serverRowId = message.serverRowId ?: match.serverRowId,
displayKind = message.displayKind ?: match.displayKind,
isHistoricalCache = false,
isRestoredUnconfirmed = false,
@@ -457,6 +472,7 @@ internal fun mergeTranscriptWithLive(
}
match.copy(
restId = message.canonicalRestId ?: match.canonicalRestId,
serverRowId = message.serverRowId ?: match.serverRowId,
content = mergedContent,
timestamp = message.timestamp,
isStreaming = message.isStreaming,
@@ -1063,6 +1063,7 @@ class ChatViewModel(
if (isCurrentSession(event.sessionId)) {
mainTurnBusy = false
lastMainCompletionAt = System.currentTimeMillis()
retireReceiptForPersistedTurn(parsePersistedTurn(event.rawPayload))
refreshSendReceipts()
}
// Buffers cleared before reduce; ViewModel resets them after
@@ -1602,7 +1603,8 @@ class ChatViewModel(
result: Any?,
) {
val messageId = outgoingRequestById.remove(requestId) ?: return
val status = rpcResultMap(result)?.get("status") as? String
val resultMap = rpcResultMap(result)
val status = resultMap?.get("status") as? String
val accepted =
when (method) {
WsMethods.SESSION_REDIRECT -> status == "redirected" || status == "queued"
@@ -1611,6 +1613,24 @@ class ChatViewModel(
}
when {
accepted -> {
// #1285: the submit ack names the row written for THIS input. Absent = unproven.
positiveRowId(resultMap?.get("user_row_id"))?.let { rowId ->
sendStore.update(messageId) { it.copy(userRowId = rowId) }
_uiState.update { state ->
state.copy(
messages =
state.messages.map {
if (it.id == messageId &&
it.serverRowId == null
) {
it.copy(serverRowId = rowId)
} else {
it
}
},
)
}
}
markPendingSend(messageId, PendingSendState.ACCEPTED)
if (sendStore.all().any { it.id == messageId && lastMainCompletionAt >= it.createdAt }) {
refreshSendReceipts()
@@ -2489,6 +2509,23 @@ class ChatViewModel(
publishPendingSends()
}
/**
* #1285: only a complete receipt proves the whole turn is durable, and only the receipt
* already bound to that exact `user_row_id` is retired. Partial or id-less receipts leave
* every local row to REST reconciliation; no row is ever inferred from text or position.
*/
private fun retireReceiptForPersistedTurn(turn: PersistedTurn?) {
val userRowId = turn?.userRowId?.takeIf { turn.complete } ?: return
val sessionId = _uiState.value.currentSessionId ?: return
val scope = sendScope()
sendStore
.all()
.filter {
it.scope == scope && it.sessionId == sessionId && it.userRowId == userRowId &&
it.state in setOf(PendingSendState.SENDING, PendingSendState.ACCEPTED, PendingSendState.UNKNOWN)
}.forEach { removePendingSend(it.id) }
}
private fun removeUnconfirmedBubble(messageId: String) {
val row = _uiState.value.messages.firstOrNull { it.id == messageId && it.canonicalRestId == null } ?: return
_uiState.update { state -> state.copy(messages = state.messages.filterNot { it.id == row.id }) }
@@ -4648,6 +4685,7 @@ class ChatViewModel(
content = it.text,
attachments = it.attachments,
timestamp = it.createdAt,
serverRowId = it.userRowId,
)
}
}
@@ -515,6 +515,9 @@ object ChatWsEventReducer {
effects = effects,
)
}
// #1285: bind the committed final row only when this bubble carries the whole reply body;
// a stripped commentary prefix means the local bubble is not that row's exact content.
val finalRowId = parsePersistedTurn(event.rawPayload)?.finalAssistantRowId?.takeIf { text == event.text }
val tps = turnUsage?.avgTps.validTpsOrNull() ?: usageState.latestTps.validTpsOrNull()
val tokenCount =
if (turnUsage?.outputTokens != null) {
@@ -531,6 +534,7 @@ object ChatWsEventReducer {
tokenCount = tokenCount,
tps = tps,
completionId = event.completionId,
serverRowId = finalRowId ?: streaming.serverRowId,
) ?: ChatMessage(
role = MessageRole.ASSISTANT,
content = text,
@@ -539,6 +543,7 @@ object ChatWsEventReducer {
tokenCount = tokenCount,
tps = tps,
completionId = event.completionId,
serverRowId = finalRowId,
)
val effects = mutableListOf<ReducerEffect>()
val sid = state.currentSessionId