diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatMessage.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatMessage.kt index 4a33fb46..badd3eb2 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatMessage.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatMessage.kt @@ -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. */ diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatPersistedReceipts.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatPersistedReceipts.kt new file mode 100644 index 00000000..08d030fc --- /dev/null +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatPersistedReceipts.kt @@ -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"]), + ) +} diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatSendStore.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatSendStore.kt index a4a9263e..4b659f0e 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatSendStore.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatSendStore.kt @@ -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 diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatServerMessageMapper.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatServerMessageMapper.kt index 9bc3a25b..a451dcc0 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatServerMessageMapper.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatServerMessageMapper.kt @@ -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 } } } diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatTranscriptMerge.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatTranscriptMerge.kt index 474b164b..4d0214f1 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatTranscriptMerge.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatTranscriptMerge.kt @@ -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(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, diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt index a46daaab..b4d9ba7e 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatViewModel.kt @@ -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, ) } } diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt index 28d248be..f5dad2b2 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducer.kt @@ -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() val sid = state.currentSessionId diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt index 815a315f..69b18af1 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPagingMergeTest.kt @@ -316,7 +316,7 @@ class ChatPagingMergeTest { val mapped = mapServerMessages("s", listOf(serverTool(42, "call-42")), 0, true, listOf(live, echo)) - assertEquals(listOf(live.copy(restId = "rest-s-42")), mapped) + assertEquals(listOf(live.copy(restId = "rest-s-42", serverRowId = 42L)), mapped) } @Test @@ -328,7 +328,10 @@ class ChatPagingMergeTest { val mapped = mapServerMessages("s", listOf(serverTool(42), serverTool(43)), 0, true, cache) assertEquals( - listOf(first.copy(restId = "rest-s-42"), second.copy(restId = "rest-s-43")), + listOf( + first.copy(restId = "rest-s-42", serverRowId = 42L), + second.copy(restId = "rest-s-43", serverRowId = 43L), + ), mapped, ) assertEquals(2, mapped.map { it.id }.toSet().size) diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPersistedReceiptsTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPersistedReceiptsTest.kt new file mode 100644 index 00000000..969222c0 --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatPersistedReceiptsTest.kt @@ -0,0 +1,179 @@ +package com.m57.hermescontrol.ui.chat + +import com.m57.hermescontrol.data.ws.WsEvent +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** #1285: persisted-turn receipts are addresses, never negative acknowledgements. */ +class ChatPersistedReceiptsTest { + @Test + fun parsesCompleteReceiptWithMixedNumberTypes() { + val turn = + parsePersistedTurn( + mapOf( + "persisted_turn" to + mapOf( + "row_ids" to listOf(41, 42L, 43.0), + "complete" to true, + "user_row_id" to 41, + "final_assistant_row_id" to 43L, + ), + ), + ) + + assertEquals(PersistedTurn(true, 41L, 43L), turn) + assertEquals( + 43L, + parsePersistedTurn(mapOf("persisted_turn" to mapOf("final_assistant_row_id" to 43.0)))?.finalAssistantRowId, + ) + } + + @Test + fun missingIdsStayUnprovenAndMalformedValuesAreIgnored() { + val turn = + parsePersistedTurn( + mapOf( + "persisted_turn" to + mapOf("complete" to "true", "user_row_id" to true, "final_assistant_row_id" to 1.5), + ), + )!! + + assertFalse(turn.complete) + assertNull(positiveRowId("8")) + assertNull(positiveRowId(0)) + assertNull(positiveRowId(-1L)) + assertNull(turn.userRowId) + assertNull(turn.finalAssistantRowId) + assertNull(parsePersistedTurn(mapOf("text" to "done"))) + assertNull(parsePersistedTurn(null)) + } + + @Test + fun rowIdMatchesUserEchoEvenWhenServerTextDiffers() { + val local = ChatMessage(id = "local", role = MessageRole.USER, content = "hi", serverRowId = 42) + val rest = + ChatMessage(id = "rest-s-42", role = MessageRole.USER, content = "hi\n\n--- rewritten", serverRowId = 42) + + val merged = mergeTranscriptWithLive(listOf(rest), listOf(local), preserveLiveIds = true) + + assertEquals(listOf("local"), merged.map { it.id }) + assertEquals("rest-s-42", merged.single().restId) + assertEquals(42L, merged.single().serverRowId) + } + + @Test + fun differentRowIdsKeepRepeatedTextAsSeparateTurns() { + val local = ChatMessage(id = "local", role = MessageRole.USER, content = "again", serverRowId = 43) + val older = ChatMessage(id = "rest-s-42", role = MessageRole.USER, content = "again", serverRowId = 42) + + val merged = mergeTranscriptWithLive(listOf(older), listOf(local), preserveLiveIds = true) + + assertEquals(setOf("local", "rest-s-42"), merged.map { it.id }.toSet()) + } + + @Test + fun missingRowIdFallsBackToContentMatching() { + val local = ChatMessage(id = "local", role = MessageRole.USER, content = "hello") + val rest = ChatMessage(id = "rest-s-9", role = MessageRole.USER, content = "hello", serverRowId = 9) + + val merged = mergeTranscriptWithLive(listOf(rest), listOf(local), preserveLiveIds = true) + + assertEquals(listOf("local"), merged.map { it.id }) + assertEquals(9L, merged.single().serverRowId) + } + + @Test + fun completeReceiptBindsFinalAssistantRow() { + val result = + complete( + "answer", + mapOf( + "row_ids" to listOf(1, 2), + "complete" to true, + "final_assistant_row_id" to 2, + ), + ) + + assertEquals( + 2L, + result.state.messages + .single() + .serverRowId, + ) + } + + @Test + fun partialReceiptStillBindsProvenFinalRowButKeepsLocalReply() { + val result = + complete("answer", mapOf("row_ids" to listOf(2), "complete" to false, "final_assistant_row_id" to 2)) + + assertEquals( + "answer", + result.state.messages + .single() + .content, + ) + assertEquals( + 2L, + result.state.messages + .single() + .serverRowId, + ) + } + + @Test + fun receiptWithoutFinalIdLeavesReplyUnbound() { + val result = complete("answer", mapOf("row_ids" to listOf(1), "complete" to false, "user_row_id" to 1)) + + assertEquals( + "answer", + result.state.messages + .single() + .content, + ) + assertNull( + result.state.messages + .single() + .serverRowId, + ) + } + + @Test + fun strippedCommentaryPrefixDoesNotClaimFinalRow() { + val sealed = ChatMessage(id = "sealed", role = MessageRole.ASSISTANT, content = "Looking. ") + val result = + ChatWsEventReducer.reduce( + state = ChatUiState(currentSessionId = "session-1", messages = listOf(sealed)), + streamingState = StreamingState(sealedOrphanIds = listOf("sealed")), + event = + WsEvent.MessageComplete( + "Looking. Done", + "session-1", + rawPayload = + mapOf( + "persisted_turn" to + mapOf("row_ids" to listOf(5), "complete" to true, "final_assistant_row_id" to 5), + ), + ), + currentSessionId = "session-1", + ) + + val reply = result.state.messages.single { it.id != "sealed" } + assertEquals("Done", reply.content) + assertNull(reply.serverRowId) + assertTrue(result.state.messages.none { it.serverRowId == 5L }) + } + + private fun complete( + text: String, + receipt: Map, + ) = ChatWsEventReducer.reduce( + state = ChatUiState(currentSessionId = "session-1"), + streamingState = StreamingState(), + event = WsEvent.MessageComplete(text, "session-1", rawPayload = mapOf("persisted_turn" to receipt)), + currentSessionId = "session-1", + ) +} diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt index c2a9f6f5..4048d080 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt @@ -3516,6 +3516,112 @@ class ChatViewModelTest { assertEquals(PendingSendState.UNKNOWN, sendStore.all().single().state) } + @Test + fun completePersistedTurnRetiresOnlyTheReceiptBoundToItsUserRow() = + runTest { + val sendStore = ChatSendStore() + val (viewModel, sessionId) = createViewModelWithSession(sendStore = sendStore) + every { HermesWsClient.sendMessage(sessionId, "Durable", any(), any()) } answers { + arg<((String) -> Unit)?>(2)?.invoke("durable-submit") + "durable-submit" + } + viewModel.sendMessage("Durable") + advanceUntilIdle() + mockEventsFlow.emit( + WsEvent.RpcResult("durable-submit", mapOf("status" to "streaming", "user_row_id" to 77)), + ) + advanceUntilIdle() + + val receipt = sendStore.all().single() + assertEquals(77L, receipt.userRowId) + assertEquals( + 77L, + viewModel.uiState.value.messages + .single { it.id == receipt.id } + .serverRowId, + ) + + // Partial and foreign receipts never retire the local turn. + mockEventsFlow.emit(WsEvent.MessageStart(sessionId)) + mockEventsFlow.emit( + WsEvent.MessageComplete( + "Reply", + sessionId, + rawPayload = + mapOf( + "persisted_turn" to + mapOf( + "row_ids" to listOf(77), + "complete" to false, + "user_row_id" to 77, + ), + ), + ), + ) + advanceUntilIdle() + assertEquals(listOf(receipt.id), sendStore.all().map { it.id }) + mockEventsFlow.emit(WsEvent.MessageStart(sessionId)) + mockEventsFlow.emit( + WsEvent.MessageComplete( + "Reply", + sessionId, + rawPayload = + mapOf( + "persisted_turn" to mapOf("row_ids" to listOf(78), "complete" to true, "user_row_id" to 78), + ), + ), + ) + advanceUntilIdle() + assertEquals(listOf(receipt.id), sendStore.all().map { it.id }) + + mockEventsFlow.emit(WsEvent.MessageStart(sessionId)) + mockEventsFlow.emit( + WsEvent.MessageComplete( + "Reply", + sessionId, + rawPayload = + mapOf( + "persisted_turn" to + mapOf( + "row_ids" to listOf(77, 79), + "complete" to true, + "user_row_id" to 77, + "final_assistant_row_id" to 79, + ), + ), + ), + ) + advanceUntilIdle() + assertTrue(sendStore.all().isEmpty()) + assertTrue( + viewModel.uiState.value.messages + .any { it.id == receipt.id }, + ) + } + + @Test + fun submitAckWithoutUserRowIdLeavesReceiptUnproven() = + runTest { + val sendStore = ChatSendStore() + val (viewModel, sessionId) = createViewModelWithSession(sendStore = sendStore) + every { HermesWsClient.sendMessage(sessionId, "Legacy", any(), any()) } answers { + arg<((String) -> Unit)?>(2)?.invoke("legacy-submit") + "legacy-submit" + } + viewModel.sendMessage("Legacy") + advanceUntilIdle() + mockEventsFlow.emit(WsEvent.RpcResult("legacy-submit", mapOf("status" to "streaming"))) + advanceUntilIdle() + + val receipt = sendStore.all().single() + assertNull(receipt.userRowId) + assertEquals(PendingSendState.ACCEPTED, receipt.state) + assertTrue( + viewModel.uiState.value.messages + .any { it.id == receipt.id && it.serverRowId == null }, + ) + } + @Test fun busyGuideUsesSteerAndUnsupportedFallsBackToQueueWithoutInterrupt() = runTest {