diff --git a/app/schemas/com.m57.hermescontrol.data.local.HermesDatabase/8.json b/app/schemas/com.m57.hermescontrol.data.local.HermesDatabase/8.json new file mode 100644 index 00000000..099208f0 --- /dev/null +++ b/app/schemas/com.m57.hermescontrol.data.local.HermesDatabase/8.json @@ -0,0 +1,115 @@ +{ + "formatVersion": 1, + "database": { + "version": 8, + "identityHash": "4f44d92f9b8db2b02a1b156b37d64651", + "entities": [ + { + "tableName": "chat_messages", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `session_id` TEXT NOT NULL, `role` TEXT NOT NULL, `content` TEXT NOT NULL, `reasoning_text` TEXT NOT NULL, `timestamp` INTEGER NOT NULL, `tool_name` TEXT, `tool_call_id` TEXT NOT NULL, `tool_status` TEXT, `is_streaming` INTEGER NOT NULL, `display_kind` TEXT, `token_count` INTEGER, `tps` REAL, `completion_id` TEXT, PRIMARY KEY(`id`))", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "sessionId", + "columnName": "session_id", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "role", + "columnName": "role", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "content", + "columnName": "content", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "reasoningText", + "columnName": "reasoning_text", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "timestamp", + "columnName": "timestamp", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "toolName", + "columnName": "tool_name", + "affinity": "TEXT" + }, + { + "fieldPath": "toolCallId", + "columnName": "tool_call_id", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "toolStatus", + "columnName": "tool_status", + "affinity": "TEXT" + }, + { + "fieldPath": "isStreaming", + "columnName": "is_streaming", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "displayKind", + "columnName": "display_kind", + "affinity": "TEXT" + }, + { + "fieldPath": "tokenCount", + "columnName": "token_count", + "affinity": "INTEGER" + }, + { + "fieldPath": "tps", + "columnName": "tps", + "affinity": "REAL" + }, + { + "fieldPath": "completionId", + "columnName": "completion_id", + "affinity": "TEXT" + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "id" + ] + }, + "indices": [ + { + "name": "index_chat_messages_session_id_timestamp", + "unique": false, + "columnNames": [ + "session_id", + "timestamp" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_chat_messages_session_id_timestamp` ON `${TABLE_NAME}` (`session_id`, `timestamp`)" + } + ] + } + ], + "setupQueries": [ + "CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)", + "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '4f44d92f9b8db2b02a1b156b37d64651')" + ] + } +} \ No newline at end of file diff --git a/app/src/androidTest/java/com/m57/hermescontrol/ui/chat/ChatReadNotificationTest.kt b/app/src/androidTest/java/com/m57/hermescontrol/ui/chat/ChatReadNotificationTest.kt new file mode 100644 index 00000000..946125fc --- /dev/null +++ b/app/src/androidTest/java/com/m57/hermescontrol/ui/chat/ChatReadNotificationTest.kt @@ -0,0 +1,500 @@ +package com.m57.hermescontrol.ui.chat + +import android.app.NotificationChannel +import android.app.NotificationManager +import android.content.Context +import android.os.Build +import android.os.Bundle +import android.service.notification.StatusBarNotification +import androidx.compose.foundation.layout.size +import androidx.compose.foundation.lazy.LazyColumn +import androidx.compose.foundation.lazy.items +import androidx.compose.foundation.lazy.rememberLazyListState +import androidx.compose.material3.SnackbarHostState +import androidx.compose.material3.Text +import androidx.compose.runtime.remember +import androidx.compose.ui.Modifier +import androidx.compose.ui.test.junit4.createComposeRule +import androidx.compose.ui.unit.dp +import androidx.core.app.NotificationCompat +import androidx.test.core.app.ApplicationProvider +import androidx.test.ext.junit.runners.AndroidJUnit4 +import androidx.test.filters.MediumTest +import androidx.test.platform.app.InstrumentationRegistry +import com.m57.hermescontrol.data.local.AuthManager +import com.m57.hermescontrol.data.model.SessionMessage +import com.m57.hermescontrol.data.ws.ConnectionStatus +import com.m57.hermescontrol.notification.ChatNotificationService +import com.m57.hermescontrol.notification.ReplyNotificationTracker +import com.m57.hermescontrol.ui.chat.components.ChatLifecycleEffects +import com.m57.hermescontrol.ui.chat.components.rememberChatScrollController +import io.mockk.mockk +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.serialization.json.JsonPrimitive +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.junit.runner.RunWith + +@RunWith(AndroidJUnit4::class) +@MediumTest +class ChatReadNotificationTest { + @get:Rule + val composeTestRule = createComposeRule() + + private lateinit var context: Context + private lateinit var notificationManager: NotificationManager + private val testChannelId = "test_chat_channel" + + @Before + fun setUp() { + context = ApplicationProvider.getApplicationContext() + notificationManager = context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.TIRAMISU) { + InstrumentationRegistry + .getInstrumentation() + .uiAutomation + .grantRuntimePermission( + context.packageName, + android.Manifest.permission.POST_NOTIFICATIONS, + ) + } + val channel = + NotificationChannel( + testChannelId, + "Test Chat Notifications", + NotificationManager.IMPORTANCE_HIGH, + ) + notificationManager.createNotificationChannel(channel) + notificationManager.cancelAll() + ReplyNotificationTracker.resetForTest() + } + + @After + fun tearDown() { + notificationManager.cancelAll() + ReplyNotificationTracker.resetForTest() + } + + private fun awaitNotification( + timeoutMs: Long = 3000L, + predicate: (Array) -> Boolean, + ): Boolean { + val start = System.currentTimeMillis() + while (System.currentTimeMillis() - start < timeoutMs) { + if (predicate(notificationManager.activeNotifications)) return true + Thread.sleep(50) + } + return predicate(notificationManager.activeNotifications) + } + + private fun postRealNotification( + kind: String, + scopeId: String = "default", + sessionId: String = "session-1", + completionId: String = "comp-1", + textSnippet: String = "Here is the reply", + serverMessageId: Int? = null, + ): Long { + val gen = + if (kind == ReplyNotificationTracker.KIND_REPLY) { + ReplyNotificationTracker.registerPendingReply( + scopeId = scopeId, + sessionId = sessionId, + completionId = completionId, + textSnippet = textSnippet, + ) + } else { + ReplyNotificationTracker.nextGeneration() + } + + val notification = + NotificationCompat + .Builder(context, testChannelId) + .setSmallIcon(android.R.drawable.ic_dialog_info) + .setContentTitle("Agent") + .setContentText(textSnippet) + .addExtras( + Bundle().apply { + putString(ReplyNotificationTracker.EXTRA_NOTIF_KIND, kind) + putString(ReplyNotificationTracker.EXTRA_SCOPE_ID, scopeId) + putString(ReplyNotificationTracker.EXTRA_SESSION_ID, sessionId) + putString(ReplyNotificationTracker.EXTRA_COMPLETION_ID, completionId) + putString(ReplyNotificationTracker.EXTRA_TEXT_SNIPPET, textSnippet) + putLong(ReplyNotificationTracker.EXTRA_GENERATION, gen) + serverMessageId?.let { + putInt(ReplyNotificationTracker.EXTRA_SERVER_MESSAGE_ID, it) + } + }, + ).build() + + when (kind) { + ReplyNotificationTracker.KIND_REPLY -> { + ReplyNotificationTracker.postReplyNotification(context, notification, gen) + } + + ReplyNotificationTracker.KIND_ACTION -> { + ReplyNotificationTracker.postActionNotification(context, notification) + } + + ReplyNotificationTracker.KIND_REPLIED -> { + ReplyNotificationTracker.postRepliedNotification(context, notification) + } + } + + val active = awaitNotification { sbns -> sbns.any { it.id == ChatNotificationService.PENDING_NOTIFICATION_ID } } + assertTrue("Notification ID 2 must be active in system after posting", active) + return gen + } + + @Test + fun chatLifecycleEffects_dismissesRealSystemNotification_whenAssistantMessageDisplayed() { + val sessionId = "session-read-test" + val completionId = "comp-12345" + val replyText = "Here is the completed response." + + postRealNotification( + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = AuthManager.activeProfileId.value.orEmpty(), + sessionId = sessionId, + completionId = completionId, + textSnippet = replyText, + ) + + val assistantMessage = + ChatMessage( + id = "msg-1", + role = MessageRole.ASSISTANT, + content = replyText, + completionId = completionId, + ) + val messages = listOf(assistantMessage) + val mockVm = mockk(relaxed = true) + + composeTestRule.setContent { + val listState = rememberLazyListState() + val scrollScope = remember { CoroutineScope(Dispatchers.Main) } + val scrollController = rememberChatScrollController(listState, scrollScope) + val snackbarHostState = remember { SnackbarHostState() } + + LazyColumn( + state = listState, + modifier = Modifier.size(300.dp, 600.dp), + ) { + items(messages, key = { "prose-${it.id}" }) { + Text(it.content) + } + } + + ChatLifecycleEffects( + sessionId = sessionId, + connectionStatus = ConnectionStatus.CONNECTED, + currentSessionId = sessionId, + messages = messages, + errorMessage = null, + backgroundCompleteMessage = null, + openError = null, + clarifyRequest = null, + sudoPrompt = null, + secretPrompt = null, + listState = listState, + scrollController = scrollController, + snackbarHostState = snackbarHostState, + viewModel = mockVm, + ) + } + + composeTestRule.waitForIdle() + + val dismissed = + awaitNotification { sbns -> sbns.none { it.id == ChatNotificationService.PENDING_NOTIFICATION_ID } } + assertTrue( + "Notification ID 2 must be dismissed from real system notifications when message is viewed", + dismissed, + ) + } + + /** + * The mobile-produced durable row id must survive in the real posted + * notification; after a process death it is the only thing that can name the + * row whose visibility dismisses this notification. + */ + @Test + fun serverRowIdSurvivesProcessDeathThroughRealNotificationExtras() { + val sessionId = "session-exact-row-test" + val replyText = "Done" + + postRealNotification( + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = AuthManager.activeProfileId.value.orEmpty(), + sessionId = sessionId, + completionId = "comp-exact-row", + textSnippet = replyText, + serverMessageId = 52, + ) + + // Simulate process death: nothing in memory, only the posted extras. + ReplyNotificationTracker.resetForTest() + val recovered = ReplyNotificationTracker.getActiveTarget(context) + assertEquals( + "Recovered target must carry the durable row id from the notification extras", + 52, + recovered?.serverMessageId, + ) + + val history = + listOf( + SessionMessage(id = 10, role = "assistant", content = JsonPrimitive("Done")), + SessionMessage(id = 52, role = "assistant", content = JsonPrimitive("Done")), + ) + val messages = mapServerMessages(sessionId, history, 0, true, emptyList(), context = context) + assertEquals("Older duplicate must stay unmapped", null, messages[0].completionId) + assertEquals("Exact row must receive the completion id", "comp-exact-row", messages[1].completionId) + + val mockVm = mockk(relaxed = true) + composeTestRule.setContent { + val listState = rememberLazyListState() + val scrollScope = remember { CoroutineScope(Dispatchers.Main) } + val scrollController = rememberChatScrollController(listState, scrollScope) + val snackbarHostState = remember { SnackbarHostState() } + + LazyColumn( + state = listState, + modifier = Modifier.size(300.dp, 600.dp), + ) { + items(messages, key = { "prose-${it.id}" }) { + Text(it.content) + } + } + + ChatLifecycleEffects( + sessionId = sessionId, + connectionStatus = ConnectionStatus.CONNECTED, + currentSessionId = sessionId, + messages = messages, + errorMessage = null, + backgroundCompleteMessage = null, + openError = null, + clarifyRequest = null, + sudoPrompt = null, + secretPrompt = null, + listState = listState, + scrollController = scrollController, + snackbarHostState = snackbarHostState, + viewModel = mockVm, + ) + } + + composeTestRule.waitForIdle() + + val dismissed = + awaitNotification { sbns -> sbns.none { it.id == ChatNotificationService.PENDING_NOTIFICATION_ID } } + assertTrue( + "Viewing the exact hydrated row must dismiss the real system notification", + dismissed, + ) + } + + @Test + fun chatLifecycleEffects_doesNotDismissRealSystemNotification_forActionRequest() { + val sessionId = "session-action-test" + val completionId = "comp-action" + val promptText = "Approval required" + + postRealNotification( + kind = ReplyNotificationTracker.KIND_ACTION, + scopeId = AuthManager.activeProfileId.value.orEmpty(), + sessionId = sessionId, + completionId = completionId, + textSnippet = promptText, + ) + + val assistantMessage = + ChatMessage( + id = "msg-1", + role = MessageRole.ASSISTANT, + content = promptText, + completionId = completionId, + ) + val messages = listOf(assistantMessage) + val mockVm = mockk(relaxed = true) + + composeTestRule.setContent { + val listState = rememberLazyListState() + val scrollScope = remember { CoroutineScope(Dispatchers.Main) } + val scrollController = rememberChatScrollController(listState, scrollScope) + val snackbarHostState = remember { SnackbarHostState() } + + LazyColumn( + state = listState, + modifier = Modifier.size(300.dp, 600.dp), + ) { + items(messages, key = { "prose-${it.id}" }) { + Text(it.content) + } + } + + ChatLifecycleEffects( + sessionId = sessionId, + connectionStatus = ConnectionStatus.CONNECTED, + currentSessionId = sessionId, + messages = messages, + errorMessage = null, + backgroundCompleteMessage = null, + openError = null, + clarifyRequest = null, + sudoPrompt = null, + secretPrompt = null, + listState = listState, + scrollController = scrollController, + snackbarHostState = snackbarHostState, + viewModel = mockVm, + ) + } + + composeTestRule.waitForIdle() + Thread.sleep(200) + + assertTrue( + "Action notifications must NOT be auto-dismissed when viewing chat", + notificationManager.activeNotifications.any { it.id == ChatNotificationService.PENDING_NOTIFICATION_ID }, + ) + } + + @Test + fun chatLifecycleEffects_doesNotDismissRealSystemNotification_forRepliedNotification() { + val sessionId = "session-replied-test" + val completionId = "comp-replied" + val replyStatusText = "Replied: done" + + postRealNotification( + kind = ReplyNotificationTracker.KIND_REPLIED, + scopeId = AuthManager.activeProfileId.value.orEmpty(), + sessionId = sessionId, + completionId = completionId, + textSnippet = replyStatusText, + ) + + val assistantMessage = + ChatMessage( + id = "msg-1", + role = MessageRole.ASSISTANT, + content = replyStatusText, + completionId = completionId, + ) + val messages = listOf(assistantMessage) + val mockVm = mockk(relaxed = true) + + composeTestRule.setContent { + val listState = rememberLazyListState() + val scrollScope = remember { CoroutineScope(Dispatchers.Main) } + val scrollController = rememberChatScrollController(listState, scrollScope) + val snackbarHostState = remember { SnackbarHostState() } + + LazyColumn( + state = listState, + modifier = Modifier.size(300.dp, 600.dp), + ) { + items(messages, key = { "prose-${it.id}" }) { + Text(it.content) + } + } + + ChatLifecycleEffects( + sessionId = sessionId, + connectionStatus = ConnectionStatus.CONNECTED, + currentSessionId = sessionId, + messages = messages, + errorMessage = null, + backgroundCompleteMessage = null, + openError = null, + clarifyRequest = null, + sudoPrompt = null, + secretPrompt = null, + listState = listState, + scrollController = scrollController, + snackbarHostState = snackbarHostState, + viewModel = mockVm, + ) + } + + composeTestRule.waitForIdle() + Thread.sleep(200) + + assertTrue( + "Replied status notifications must NOT be auto-dismissed when viewing chat", + notificationManager.activeNotifications.any { it.id == ChatNotificationService.PENDING_NOTIFICATION_ID }, + ) + } + + @Test + fun chatLifecycleEffects_doesNotDismissNotification_forDifferentSession() { + val sessionIdWithNotif = "session-with-notif" + val differentSessionId = "other-session" + val completionId = "comp-12345" + val replyText = "Here is the completed response." + + postRealNotification( + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = AuthManager.activeProfileId.value.orEmpty(), + sessionId = sessionIdWithNotif, + completionId = completionId, + textSnippet = replyText, + ) + + val assistantMessage = + ChatMessage( + id = "msg-1", + role = MessageRole.ASSISTANT, + content = replyText, + completionId = completionId, + ) + val messages = listOf(assistantMessage) + val mockVm = mockk(relaxed = true) + + composeTestRule.setContent { + val listState = rememberLazyListState() + val scrollScope = remember { CoroutineScope(Dispatchers.Main) } + val scrollController = rememberChatScrollController(listState, scrollScope) + val snackbarHostState = remember { SnackbarHostState() } + + LazyColumn( + state = listState, + modifier = Modifier.size(300.dp, 600.dp), + ) { + items(messages, key = { "prose-${it.id}" }) { + Text(it.content) + } + } + + ChatLifecycleEffects( + sessionId = differentSessionId, + connectionStatus = ConnectionStatus.CONNECTED, + currentSessionId = differentSessionId, + messages = messages, + errorMessage = null, + backgroundCompleteMessage = null, + openError = null, + clarifyRequest = null, + sudoPrompt = null, + secretPrompt = null, + listState = listState, + scrollController = scrollController, + snackbarHostState = snackbarHostState, + viewModel = mockVm, + ) + } + + composeTestRule.waitForIdle() + Thread.sleep(200) + + assertTrue( + "Notification ID 2 must remain active for a different session", + notificationManager.activeNotifications.any { it.id == ChatNotificationService.PENDING_NOTIFICATION_ID }, + ) + } +} diff --git a/app/src/main/java/com/m57/hermescontrol/HermesControlApp.kt b/app/src/main/java/com/m57/hermescontrol/HermesControlApp.kt index 2289e292..5e75e555 100644 --- a/app/src/main/java/com/m57/hermescontrol/HermesControlApp.kt +++ b/app/src/main/java/com/m57/hermescontrol/HermesControlApp.kt @@ -14,6 +14,7 @@ import com.m57.hermescontrol.data.local.SessionListCacheStore import com.m57.hermescontrol.data.remote.NetworkMonitor import com.m57.hermescontrol.data.remote.OkHttpProvider import com.m57.hermescontrol.data.update.UpdateNoticeManager +import com.m57.hermescontrol.notification.TurnCorrelationTracker import com.m57.hermescontrol.ui.analytics.AnalyticsPreloader import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -31,6 +32,9 @@ class HermesControlApp : AuthManager.init(this) NetworkMonitor.init(this) SessionListCacheStore.init(this) + // Reply-notification turn boundaries survive process death (armed before + // each mobile prompt, consumed by the matching completion). + TurnCorrelationTracker.attach(this) appScope.launch { AuthManager.initializationState.first { it == AuthManager.InitializationState.Ready } // Issue #537 follow-up (A): preload analytics in the background after launch diff --git a/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageEntity.kt b/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageEntity.kt index 548e45b5..3a4d6f18 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageEntity.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageEntity.kt @@ -46,4 +46,6 @@ data class ChatMessageEntity( val tokenCount: Int? = null, @ColumnInfo(name = "tps") val tps: Double? = null, + @ColumnInfo(name = "completion_id") + val completionId: String? = null, ) diff --git a/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageMapper.kt b/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageMapper.kt index 2600557b..953c093d 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageMapper.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/local/ChatMessageMapper.kt @@ -37,6 +37,7 @@ fun ChatMessageEntity.toUiModel(): ChatMessage = displayKind = displayKind, tokenCount = tokenCount, tps = tps, + completionId = completionId, ) fun ChatMessage.toEntity(sessionId: String): ChatMessageEntity = @@ -54,4 +55,5 @@ fun ChatMessage.toEntity(sessionId: String): ChatMessageEntity = displayKind = displayKind, tokenCount = tokenCount, tps = tps, + completionId = completionId, ) diff --git a/app/src/main/java/com/m57/hermescontrol/data/local/HermesDatabase.kt b/app/src/main/java/com/m57/hermescontrol/data/local/HermesDatabase.kt index 16741b0d..f3525f14 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/local/HermesDatabase.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/local/HermesDatabase.kt @@ -14,7 +14,7 @@ import java.io.File @Database( entities = [ChatMessageEntity::class], - version = 7, + version = 8, exportSchema = true, ) abstract class HermesDatabase : RoomDatabase() { @@ -81,6 +81,15 @@ abstract class HermesDatabase : RoomDatabase() { } } + val MIGRATION_7_8: Migration = + object : Migration(7, 8) { + override suspend fun migrate(connection: SQLiteConnection) { + connection.execSQL( + "ALTER TABLE `chat_messages` ADD COLUMN `completion_id` TEXT", + ) + } + } + suspend fun get(context: Context): HermesDatabase = withContext(Dispatchers.IO) { instance?.let { return@withContext it } @@ -116,6 +125,7 @@ abstract class HermesDatabase : RoomDatabase() { MIGRATION_4_5, MIGRATION_5_6, MIGRATION_6_7, + MIGRATION_7_8, ).fallbackToDestructiveMigration(false) .build() .also { instance = it } diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt index 14c064ce..bbad8a29 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt @@ -110,7 +110,18 @@ object EventParser { "message.complete" -> { val text = payload?.get("text") as? String ?: "" val reasoning = payload?.get("reasoning") as? String - WsEvent.MessageComplete(text, sessionId, reasoning, rawPayload = payload) + val completionId = + payload?.get("completion_id") as? String + ?: java.util.UUID + .randomUUID() + .toString() + WsEvent.MessageComplete( + text, + sessionId, + reasoning, + rawPayload = payload, + completionId = completionId, + ) } "message.done" -> { diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt index cea239f9..712d4bd5 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/WsEvent.kt @@ -63,6 +63,11 @@ sealed class WsEvent { val storedSessionId: String? = null, /** Full raw payload map (including usage/avg_tps) emitted with message.complete. */ val rawPayload: Map? = null, + /** Stable completion identity for notification and read-tracking correlation. */ + val completionId: String = + java.util.UUID + .randomUUID() + .toString(), ) : WsEvent() data class MessageDone( diff --git a/app/src/main/java/com/m57/hermescontrol/notification/ChatNotificationService.kt b/app/src/main/java/com/m57/hermescontrol/notification/ChatNotificationService.kt index 1feba748..831aa535 100644 --- a/app/src/main/java/com/m57/hermescontrol/notification/ChatNotificationService.kt +++ b/app/src/main/java/com/m57/hermescontrol/notification/ChatNotificationService.kt @@ -13,7 +13,6 @@ import androidx.core.app.NotificationCompat import androidx.core.app.RemoteInput import com.m57.hermescontrol.MainActivity import com.m57.hermescontrol.R -import com.m57.hermescontrol.data.local.AuthManager import com.m57.hermescontrol.data.remote.NetworkMonitor import com.m57.hermescontrol.data.session.ActiveSessionHolder import com.m57.hermescontrol.data.ws.HermesWsClient @@ -69,6 +68,12 @@ class ChatNotificationService : Service() { private val isAppInForeground = AtomicBoolean(false) internal val lifecycle = ForegroundServiceLifecycle() + /** + * Turn-row correlation used when a reply completes in the background. + * Injectable so the resolution path can be exercised without REST. + */ + internal var turnRowResolver: TurnRowResolver = defaultTurnRowResolver + @Volatile internal var activeServiceInstance: ChatNotificationService? = null @@ -111,10 +116,23 @@ class ChatNotificationService : Service() { .take(100) .replace("\n", " ") .ifBlank { getString(R.string.notif_new_message) } - showReplyNotification( - preview, + val targetSessionId = event.storedSessionId - ?: ActiveSessionHolder.resolveStoredSessionId(event.sessionId), + ?: ActiveSessionHolder.resolveStoredSessionId(event.sessionId) + showReplyNotification( + text = preview, + sessionId = targetSessionId, + isReplyMessage = true, + completionId = event.completionId, + // The durable REST row for this turn, when the + // boundary armed before the prompt was submitted + // still lets us name it unambiguously. Null is a + // normal, safe outcome: the notification is then + // never auto-dismissed from REST hydration, which + // is strictly better than dismissing the wrong + // duplicate reply. + serverMessageId = + coalesceTurnRow(targetSessionId, event.text), ) // The wait is over — retire the foreground // service. The reply notification above @@ -167,9 +185,37 @@ class ChatNotificationService : Service() { } } + /** + * Names the exact REST row this completion produced, using the turn + * boundary armed before the prompt was submitted. Any failure to prove that + * identity returns null — the notification is still posted, it just will not + * be auto-dismissed from REST hydration. + */ + private suspend fun coalesceTurnRow( + sessionId: String?, + completionText: String, + ): Int? { + if (sessionId.isNullOrBlank()) return null + return try { + correlateCompletedTurnRow( + scopeId = correlationScopeId(), + sessionId = sessionId, + completionText = completionText, + resolver = turnRowResolver, + ) + } catch (e: kotlinx.coroutines.CancellationException) { + throw e + } catch (e: Exception) { + null + } + } + private fun showReplyNotification( text: String, sessionId: String?, + isReplyMessage: Boolean = false, + completionId: String? = null, + serverMessageId: Int? = null, ) { val builder = NotificationCompat @@ -183,6 +229,39 @@ class ChatNotificationService : Service() { .setAutoCancel(true) .setContentIntent(buildContentIntent(sessionId)) + var replyGeneration: Long? = null + if (isReplyMessage && !sessionId.isNullOrBlank() && !completionId.isNullOrBlank()) { + val scopeId = correlationScopeId() + val generation = + ReplyNotificationTracker.registerPendingReply( + scopeId = scopeId, + sessionId = sessionId, + completionId = completionId, + textSnippet = text, + serverMessageId = serverMessageId, + ) + replyGeneration = generation + builder.addExtras( + android.os.Bundle().apply { + putString(ReplyNotificationTracker.EXTRA_NOTIF_KIND, ReplyNotificationTracker.KIND_REPLY) + putString(ReplyNotificationTracker.EXTRA_SCOPE_ID, scopeId) + putString(ReplyNotificationTracker.EXTRA_SESSION_ID, sessionId) + putString(ReplyNotificationTracker.EXTRA_COMPLETION_ID, completionId) + putString(ReplyNotificationTracker.EXTRA_TEXT_SNIPPET, text) + putLong(ReplyNotificationTracker.EXTRA_GENERATION, generation) + serverMessageId?.let { + putInt(ReplyNotificationTracker.EXTRA_SERVER_MESSAGE_ID, it) + } + }, + ) + } else { + builder.addExtras( + android.os.Bundle().apply { + putString(ReplyNotificationTracker.EXTRA_NOTIF_KIND, ReplyNotificationTracker.KIND_ACTION) + }, + ) + } + if (!sessionId.isNullOrBlank()) { val replyLabel = getString(R.string.notif_reply_placeholder) val remoteInput = @@ -194,6 +273,11 @@ class ChatNotificationService : Service() { val replyIntent = Intent(this, NotificationReplyReceiver::class.java).apply { action = "$packageName.ACTION_NOTIFICATION_REPLY" + component = + android.content.ComponentName( + this@ChatNotificationService, + NotificationReplyReceiver::class.java, + ) setPackage(packageName) putExtra(NotificationReplyReceiver.EXTRA_SESSION_ID, sessionId) } @@ -218,19 +302,32 @@ class ChatNotificationService : Service() { builder.addAction(action) } - val manager = getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager - manager.notify(PENDING_NOTIFICATION_ID, builder.build()) + val notification = builder.build() + if (replyGeneration != null) { + ReplyNotificationTracker.postReplyNotification( + context = this, + notification = notification, + generation = replyGeneration, + ) + } else { + ReplyNotificationTracker.postActionNotification( + context = this, + notification = notification, + ) + } } private fun buildContentIntent(sessionId: String?): PendingIntent { val intent = Intent(this, MainActivity::class.java).apply { action = MainActivity.ACTION_OPEN_CHAT_FROM_NOTIFICATION + component = android.content.ComponentName(this@ChatNotificationService, MainActivity::class.java) + setPackage(packageName) flags = Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_CLEAR_TOP + if (!sessionId.isNullOrBlank()) { + putExtra(NotificationReplyReceiver.EXTRA_SESSION_ID, sessionId) + } } - if (!sessionId.isNullOrBlank()) { - intent.putExtra(NotificationReplyReceiver.EXTRA_SESSION_ID, sessionId) - } return PendingIntent.getActivity( this, sessionId?.hashCode() ?: 0, diff --git a/app/src/main/java/com/m57/hermescontrol/notification/NotificationReplyReceiver.kt b/app/src/main/java/com/m57/hermescontrol/notification/NotificationReplyReceiver.kt index dde82529..b776b055 100644 --- a/app/src/main/java/com/m57/hermescontrol/notification/NotificationReplyReceiver.kt +++ b/app/src/main/java/com/m57/hermescontrol/notification/NotificationReplyReceiver.kt @@ -24,6 +24,13 @@ open class NotificationReplyReceiver : BroadcastReceiver() { const val KEY_TEXT_REPLY = "key_text_reply" const val EXTRA_SESSION_ID = "extra_session_id" private const val REPLY_TIMEOUT_MS = 5_000L + + /** + * Turn-boundary read budget inside the reply deadline. Deliberately tight: + * the probe sits on the reply path, so a slow or unreachable gateway must + * cost the user's reply only a sliver of its 5s deadline. + */ + private const val BOUNDARY_TIMEOUT_MS = 250L } // Reusable scope for async reply processing — avoids creating a new @@ -50,7 +57,11 @@ open class NotificationReplyReceiver : BroadcastReceiver() { .setContentText("Replied") .setPriority(NotificationCompat.PRIORITY_HIGH) .setAutoCancel(true) - .build() + .addExtras( + android.os.Bundle().apply { + putString(ReplyNotificationTracker.EXTRA_NOTIF_KIND, ReplyNotificationTracker.KIND_REPLIED) + }, + ).build() override fun onReceive( context: Context, @@ -78,6 +89,17 @@ open class NotificationReplyReceiver : BroadcastReceiver() { return@withContext } + // Mobile-originated follow-up turn: arm its durable + // boundary BEFORE prompt.submit. Budgeted tighter + // than the chat composer's — this receiver has its + // own 5s deadline and a failed capture must only + // make the reply uncorrelatable, never lose it. + captureTurnBoundary( + scopeId = correlationScopeId(), + sessionId = sessionId, + timeoutMs = BOUNDARY_TIMEOUT_MS, + ) + val runtimeSessionId = ActiveSessionHolder.resolveRuntimeSessionId(sessionId) ?: resumeSession(sessionId) @@ -102,8 +124,7 @@ open class NotificationReplyReceiver : BroadcastReceiver() { dao.upsert(entity) val repliedNotification = buildReplyNotification(context) - val manager = context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager - manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, repliedNotification) + ReplyNotificationTracker.postRepliedNotification(context, repliedNotification) // The follow-up turn is now pending — keep the // foreground service alive if it retired after the diff --git a/app/src/main/java/com/m57/hermescontrol/notification/ReplyNotificationTracker.kt b/app/src/main/java/com/m57/hermescontrol/notification/ReplyNotificationTracker.kt new file mode 100644 index 00000000..b6009916 --- /dev/null +++ b/app/src/main/java/com/m57/hermescontrol/notification/ReplyNotificationTracker.kt @@ -0,0 +1,302 @@ +package com.m57.hermescontrol.notification + +import android.app.Notification +import android.app.NotificationManager +import android.content.Context +import androidx.annotation.VisibleForTesting +import java.util.concurrent.atomic.AtomicLong + +data class ReplyNotificationTarget( + val scopeId: String, + val sessionId: String, + val completionId: String, + val generation: Long, + val textSnippet: String = "", + val timestamp: Long = System.currentTimeMillis(), + /** + * Durable REST `messages.id` of the row this reply produced. Resolved by + * [TurnCorrelationTracker] from the boundary armed before the prompt was + * submitted — never taken from the completion event, text, or a clock. Null + * means the turn could not be correlated, and then hydration must not touch + * this notification at all (fail closed). + */ + val serverMessageId: Int? = null, +) { + fun matches( + candidateScopeId: String?, + candidateSessionId: String?, + candidateCompletionId: String?, + ): Boolean { + if (scopeId.isNotBlank() && scopeId != candidateScopeId) { + return false + } + if (candidateSessionId.isNullOrBlank() || sessionId != candidateSessionId) { + return false + } + if (candidateCompletionId.isNullOrBlank() || completionId.isBlank()) { + return false + } + return candidateCompletionId == completionId + } +} + +internal data class ActiveReplyInfo( + val id: Int, + val kind: String?, + val scopeId: String?, + val sessionId: String?, + val completionId: String?, + val textSnippet: String?, + val generation: Long, + val timestamp: Long, + val serverMessageId: Int? = null, +) + +object ReplyNotificationTracker { + const val EXTRA_NOTIF_KIND = "hermes_notif_kind" + const val EXTRA_SCOPE_ID = "hermes_scope_id" + const val EXTRA_SESSION_ID = "hermes_session_id" + const val EXTRA_COMPLETION_ID = "hermes_completion_id" + const val EXTRA_TEXT_SNIPPET = "hermes_text_snippet" + const val EXTRA_GENERATION = "hermes_generation" + const val EXTRA_SERVER_MESSAGE_ID = "hermes_server_message_id" + + const val KIND_REPLY = "reply" + const val KIND_ACTION = "action" + const val KIND_REPLIED = "replied" + + private val generationCounter = AtomicLong(0L) + private val tombstoneGeneration = AtomicLong(0L) + + @Volatile + private var activeTarget: ReplyNotificationTarget? = null + + private val defaultActiveNotificationProvider: (Context) -> ActiveReplyInfo? = { context -> + runCatching { + val manager = + context.getSystemService(Context.NOTIFICATION_SERVICE) as? NotificationManager + val sbn = + manager?.activeNotifications?.firstOrNull { + it.id == ChatNotificationService.PENDING_NOTIFICATION_ID + } + if (sbn != null) { + val extras = sbn.notification?.extras + ActiveReplyInfo( + id = sbn.id, + kind = extras?.getString(EXTRA_NOTIF_KIND), + scopeId = extras?.getString(EXTRA_SCOPE_ID), + sessionId = extras?.getString(EXTRA_SESSION_ID), + completionId = extras?.getString(EXTRA_COMPLETION_ID), + textSnippet = extras?.getString(EXTRA_TEXT_SNIPPET), + generation = extras?.getLong(EXTRA_GENERATION, 0L) ?: 0L, + timestamp = sbn.postTime, + serverMessageId = extras?.getInt(EXTRA_SERVER_MESSAGE_ID, -1)?.takeIf { it >= 0 }, + ) + } else { + null + } + }.getOrNull() + } + + internal var activeNotificationProvider: (Context) -> ActiveReplyInfo? = defaultActiveNotificationProvider + + fun nextGeneration(): Long { + val floor = maxOf(generationCounter.get(), tombstoneGeneration.get()) + return generationCounter.updateAndGet { maxOf(it, floor) + 1 } + } + + @Synchronized + fun registerPendingReply( + scopeId: String, + sessionId: String, + completionId: String, + textSnippet: String, + timestamp: Long = System.currentTimeMillis(), + serverMessageId: Int? = null, + ): Long { + val generation = nextGeneration() + activeTarget = + ReplyNotificationTarget( + scopeId = scopeId, + sessionId = sessionId, + completionId = completionId, + generation = generation, + textSnippet = textSnippet, + timestamp = timestamp, + serverMessageId = serverMessageId, + ) + return generation + } + + @Synchronized + fun postReplyNotification( + context: Context, + notification: Notification, + generation: Long, + ): Boolean { + if (generation <= tombstoneGeneration.get()) { + return false + } + val current = activeTarget + if (current != null && current.generation > generation) { + return false + } + val manager = + context.getSystemService(Context.NOTIFICATION_SERVICE) as? NotificationManager + ?: return false + manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, notification) + return true + } + + @Synchronized + fun postActionNotification( + context: Context, + notification: Notification, + ): Boolean { + val nextGen = nextGeneration() + tombstoneGeneration.updateAndGet { maxOf(it, nextGen) } + activeTarget = null + val manager = + context.getSystemService(Context.NOTIFICATION_SERVICE) as? NotificationManager + ?: return false + manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, notification) + return true + } + + @Synchronized + fun postRepliedNotification( + context: Context, + notification: Notification, + ): Boolean { + val nextGen = nextGeneration() + tombstoneGeneration.updateAndGet { maxOf(it, nextGen) } + activeTarget = null + val manager = + context.getSystemService(Context.NOTIFICATION_SERVICE) as? NotificationManager + ?: return false + manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, notification) + return true + } + + @Synchronized + fun onReplyNotificationPosted( + scopeId: String, + sessionId: String, + completionId: String, + textSnippet: String, + generation: Long, + timestamp: Long = System.currentTimeMillis(), + serverMessageId: Int? = null, + ) { + if (generation <= tombstoneGeneration.get()) return + generationCounter.updateAndGet { maxOf(it, generation) } + activeTarget = + ReplyNotificationTarget( + scopeId = scopeId, + sessionId = sessionId, + completionId = completionId, + generation = generation, + textSnippet = textSnippet, + timestamp = timestamp, + serverMessageId = serverMessageId, + ) + } + + @Synchronized + fun onNonReplyNotificationPosted() { + val nextGen = nextGeneration() + tombstoneGeneration.updateAndGet { maxOf(it, nextGen) } + activeTarget = null + } + + @Synchronized + fun getActiveTarget(context: Context? = null): ReplyNotificationTarget? { + val inMemory = activeTarget + if (inMemory != null) return inMemory + return if (context != null) resolveTarget(context) else null + } + + @Synchronized + fun onMessageVisible( + context: Context, + scopeId: String?, + sessionId: String?, + completionId: String?, + ): Boolean { + if (sessionId.isNullOrBlank()) return false + val target = resolveTarget(context) ?: return false + if (target.matches(scopeId, sessionId, completionId)) { + return cancelReplyNotification(context, target.generation) + } + return false + } + + @Synchronized + fun cancelReplyNotification( + context: Context, + expectedGeneration: Long, + ): Boolean { + val target = resolveTarget(context) + if (target != null && target.generation > expectedGeneration) { + return false + } + val manager = + context.getSystemService(Context.NOTIFICATION_SERVICE) as? NotificationManager + ?: return false + + val activeInfo = activeNotificationProvider(context) + if (activeInfo != null) { + if (activeInfo.kind != KIND_REPLY) { + activeTarget = null + return false + } + if (activeInfo.generation <= tombstoneGeneration.get()) { + activeTarget = null + return false + } + if (activeInfo.generation > expectedGeneration) { + return false + } + } + + tombstoneGeneration.updateAndGet { maxOf(it, expectedGeneration) } + manager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) + activeTarget = null + return true + } + + private fun resolveTarget(context: Context): ReplyNotificationTarget? { + val inMemory = activeTarget + if (inMemory != null) return inMemory + + val activeInfo = activeNotificationProvider(context) ?: return null + if (activeInfo.kind != KIND_REPLY) return null + generationCounter.updateAndGet { maxOf(it, activeInfo.generation) } + if (activeInfo.generation <= tombstoneGeneration.get()) return null + val sessionId = activeInfo.sessionId.orEmpty() + if (sessionId.isBlank()) return null + + return ReplyNotificationTarget( + scopeId = activeInfo.scopeId.orEmpty(), + sessionId = sessionId, + completionId = activeInfo.completionId.orEmpty(), + generation = activeInfo.generation, + textSnippet = activeInfo.textSnippet.orEmpty(), + timestamp = activeInfo.timestamp, + serverMessageId = activeInfo.serverMessageId, + ).also { activeTarget = it } + } + + @VisibleForTesting + fun resetForTest() { + activeTarget = null + generationCounter.set(0L) + tombstoneGeneration.set(0L) + activeNotificationProvider = defaultActiveNotificationProvider + } + + @VisibleForTesting + fun setTargetForTest(target: ReplyNotificationTarget?) { + activeTarget = target + } +} diff --git a/app/src/main/java/com/m57/hermescontrol/notification/TurnCorrelationTracker.kt b/app/src/main/java/com/m57/hermescontrol/notification/TurnCorrelationTracker.kt new file mode 100644 index 00000000..8c3d8656 --- /dev/null +++ b/app/src/main/java/com/m57/hermescontrol/notification/TurnCorrelationTracker.kt @@ -0,0 +1,384 @@ +package com.m57.hermescontrol.notification + +import android.content.Context +import androidx.annotation.VisibleForTesting +import com.m57.hermescontrol.data.local.AuthManager +import com.m57.hermescontrol.data.model.SessionMessage +import com.m57.hermescontrol.data.remote.ApiClient +import com.m57.hermescontrol.data.remote.NetworkResult +import com.m57.hermescontrol.data.remote.safeApiCall +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.withTimeoutOrNull +import java.util.concurrent.atomic.AtomicLong + +/** + * Durable lower bound for one mobile-originated turn: the highest REST + * transcript row id that existed immediately BEFORE `prompt.submit` left the + * device. + * + * The id is the gateway's AUTOINCREMENT `messages.id`, so every row persisted + * by the turn itself is strictly greater than [beforeMessageId]. That gives the + * notification layer a turn-scoped window it can search without text + * heuristics, row ordering, or cross-machine clocks. + * + * Captured before the prompt is submitted, never after: a fast turn can persist + * its assistant row before a post-submit snapshot request finishes, which would + * place the reply inside the "before" snapshot (and a lower bound can then never + * exclude it). + */ +data class TurnBoundary( + val scopeId: String, + val sessionId: String, + val beforeMessageId: Int, + /** Monotonic arm order — a stale clear must not remove a newer turn's boundary. */ + val generation: Long, + val armedAt: Long, +) { + fun matchesKey( + otherScopeId: String, + otherSessionId: String, + ): Boolean = scopeId == otherScopeId && sessionId == otherSessionId +} + +/** Persistence seam so the tracker can be unit-tested without an Android context. */ +internal interface TurnBoundaryStore { + fun read(): List + + fun write(boundaries: List) +} + +/** + * SharedPreferences-backed store — a handful of tiny `scope|session -> state` + * rows, so a process death during a running turn does not lose the boundary. + */ +internal class PrefsTurnBoundaryStore( + context: Context, +) : TurnBoundaryStore { + private val prefs = + context.applicationContext.getSharedPreferences(PREFS_NAME, Context.MODE_PRIVATE) + + override fun read(): List = + prefs.all.mapNotNull { (key, value) -> decodeBoundary(key, value as? String) } + + override fun write(boundaries: List) { + val editor = prefs.edit().clear() + boundaries.forEach { editor.putString(encodeKey(it.scopeId, it.sessionId), encodeValue(it)) } + editor.apply() + } + + internal companion object { + const val PREFS_NAME = "hermes_turn_boundaries" + private const val KEY_SEPARATOR = '|' + + fun encodeKey( + scopeId: String, + sessionId: String, + ): String = "$scopeId$KEY_SEPARATOR$sessionId" + + fun encodeValue(boundary: TurnBoundary): String = + "${boundary.beforeMessageId}:${boundary.generation}:${boundary.armedAt}" + + fun decodeBoundary( + key: String, + value: String?, + ): TurnBoundary? { + val separator = key.indexOf(KEY_SEPARATOR) + if (separator <= 0 || value == null) return null + val parts = value.split(':') + if (parts.size != 3) return null + val beforeMessageId = parts[0].toIntOrNull() ?: return null + val generation = parts[1].toLongOrNull() ?: return null + val armedAt = parts[2].toLongOrNull() ?: return null + return TurnBoundary( + scopeId = key.substring(0, separator), + sessionId = key.substring(separator + 1), + beforeMessageId = beforeMessageId, + generation = generation, + armedAt = armedAt, + ) + } + } +} + +/** + * Turn-boundary registry for reply-notification correlation. + * + * A boundary is armed per (profile scope, stored session id) immediately before + * a mobile-originated prompt is submitted, and consumed when a completion for + * that session is handled. Sessions that never had a boundary (desktop/TUI + * turns, another connected client, a failed capture, queued or group-chat + * turns) simply fail closed: the reply notification is posted without a durable + * row id and therefore is never auto-dismissed from REST hydration alone. + * + * Scope isolation is structural — a boundary is only visible to the exact + * profile that armed it, so switching profiles makes the old boundary + * unreachable instead of resolvable against the wrong namespace. + */ +object TurnCorrelationTracker { + /** Bound the registry: stale turns must not accumulate across app restarts. */ + private const val MAX_ENTRIES = 32 + + /** A boundary older than this belongs to a turn nobody is waiting on anymore. */ + private const val MAX_AGE_MS = 6 * 60 * 60 * 1000L + + private val generationCounter = AtomicLong(0L) + private val lock = Any() + + @Volatile + private var cache: List = emptyList() + + @Volatile + private var store: TurnBoundaryStore? = null + + /** Idempotent — called once from [com.m57.hermescontrol.HermesControlApp]. */ + fun attach(context: Context) { + synchronized(lock) { + if (store != null) return + // Never let persistence break app startup: an unusable context + // degrades to in-memory boundaries. + val prefsStore = runCatching { PrefsTurnBoundaryStore(context) }.getOrNull() ?: return + store = prefsStore + val restored = prune(runCatching { prefsStore.read() }.getOrDefault(emptyList())) + cache = restored + reserveGenerationFloor(restored) + } + } + + /** Test seam: swap the persistent store (or detach with `null`). */ + @VisibleForTesting + internal fun attachStore(store: TurnBoundaryStore?) { + synchronized(lock) { + this.store = store + val restored = store?.let { prune(it.read()) } ?: emptyList() + cache = restored + reserveGenerationFloor(restored) + } + } + + /** + * Records the pre-submit high-watermark. Returns the armed boundary, or null + * when the identity is unusable (blank scope/session, no rows and no + * confirmed read). + */ + fun armBoundary( + scopeId: String, + sessionId: String, + beforeMessageId: Int, + now: Long = System.currentTimeMillis(), + ): TurnBoundary? { + if (scopeId.isBlank() || sessionId.isBlank() || beforeMessageId < 0) return null + val boundary = + TurnBoundary( + scopeId = scopeId, + sessionId = sessionId, + beforeMessageId = beforeMessageId, + generation = generationCounter.incrementAndGet(), + armedAt = now, + ) + synchronized(lock) { + val next = prune(cache.filterNot { it.matchesKey(scopeId, sessionId) } + boundary, now = now) + cache = next + runCatching { store?.write(next) } + } + return boundary + } + + fun boundaryFor( + scopeId: String, + sessionId: String, + now: Long = System.currentTimeMillis(), + ): TurnBoundary? { + if (scopeId.isBlank() || sessionId.isBlank()) return null + return synchronized(lock) { + cache.firstOrNull { it.matchesKey(scopeId, sessionId) && isFresh(it, now) } + } + } + + /** + * Consumes a boundary. [generation] guards the race where a newer turn armed + * a replacement while an older completion was still being correlated. + */ + fun clearBoundary( + scopeId: String, + sessionId: String, + generation: Long? = null, + ): Boolean { + synchronized(lock) { + val existing = cache.firstOrNull { it.matchesKey(scopeId, sessionId) } ?: return false + if (generation != null && existing.generation != generation) return false + val next = cache.filterNot { it === existing } + cache = next + runCatching { store?.write(next) } + return true + } + } + + private fun prune( + boundaries: List, + now: Long = System.currentTimeMillis(), + ): List = + boundaries + .filter { isFresh(it, now) } + .sortedByDescending { it.armedAt } + .take(MAX_ENTRIES) + + /** + * Keeps generations unique across process death. + * + * The counter lives in memory only, so after a restart a restored boundary + * (generation N) and the next armed boundary would both be N — and the + * stale-clear guard, which exists precisely to stop an old completion's + * clear from removing a newer turn's boundary, would stop protecting + * anything. Lifting the counter above the recovered generations keeps the + * guard meaningful. + */ + private fun reserveGenerationFloor(restored: List) { + val maxRestored = restored.maxOfOrNull { it.generation } ?: 0L + generationCounter.updateAndGet { maxOf(it, maxRestored) } + } + + private fun isFresh( + boundary: TurnBoundary, + now: Long, + ): Boolean = now - boundary.armedAt <= MAX_AGE_MS + + @VisibleForTesting + internal fun resetForTest() { + synchronized(lock) { + cache = emptyList() + store = null + generationCounter.set(0L) + } + } +} + +/** How long the pre-submit boundary read may take before the turn goes uncorrelatable. */ +private const val TURN_BOUNDARY_TIMEOUT_MS = 1_500L + +/** + * Correlation scope for the active connection profile. + * + * `AuthManager.activeProfileId` stays null until a server profile is explicitly + * selected, and the rest of AuthManager treats that as + * [AuthManager.DEFAULT_PROFILE_ID] (see `currentDataScope()`). Collapsing + * null/blank here gives this feature ONE scope identity — a raw `.orEmpty()` + * would reject the default profile, so a normal install would arm no boundary, + * resolve no row, and silently never auto-dismiss a reply notification. + * + * The arming side (prompt submit), the resolution side (completion), the + * notification extras, and the chat viewport observer must ALL use this, or the + * two halves disagree on the scope and dismissal stops working. + */ +internal fun correlationScopeId(): String = + AuthManager.activeProfileId.value?.takeIf { it.isNotBlank() } + ?: AuthManager.DEFAULT_PROFILE_ID + +/** Rows scanned when correlating a completed turn (newest-first page). */ +private const val TURN_ROW_SCAN_LIMIT = 20 + +/** + * Reads the durable REST transcript high-watermark for [sessionId] and arms a + * boundary for it. Call immediately BEFORE submitting the prompt. + * + * Deliberately does not send a `profile` query param: the boundary ids are later + * compared against rows hydrated by the chat screen, which fetches this + * endpoint without a profile. Keeping both reads in the same namespace is what + * makes the id comparison meaningful. + * + * Returns false on any failure (including a gateway that does not confirm + * `order=latest`) — the caller sends its prompt anyway and the turn simply + * becomes uncorrelatable. + */ +internal suspend fun captureTurnBoundary( + scopeId: String, + sessionId: String, + timeoutMs: Long = TURN_BOUNDARY_TIMEOUT_MS, +): Boolean { + if (scopeId.isBlank() || sessionId.isBlank()) return false + val maxMessageId = + try { + withTimeoutOrNull(timeoutMs) { fetchBoundaryTailId(sessionId) } + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + // Total by contract: this probe is best-effort identity for a + // notification convenience and must NEVER fail the caller's turn. + // Swallowing Throwable (not just Exception) is deliberate — a + // gateway-layer Error (uninitialised client, NoClassDefFound) used to + // escape past the caller's `catch (Exception)` and get thrown inside + // a background scope, which aborted the send and leaked as an + // uncaught exception attributed to an unrelated later test. + null + } ?: return false + return TurnCorrelationTracker.armBoundary(scopeId, sessionId, maxMessageId) != null +} + +/** + * Newest row id of the transcript, or null when the read cannot be trusted. + * + * `pagination` is only returned by gateways that honoured `order=latest`; a + * backend that ignores the param would answer a `limit=1` probe with the OLDEST + * row, producing a boundary that is too low — a lower bound that is too low is + * exactly what lets a historical duplicate win. Absent proof, fail closed. + */ +private suspend fun fetchBoundaryTailId(sessionId: String): Int? { + val result = + safeApiCall { + ApiClient.hermesApi.getSessionMessages( + sessionId = sessionId, + limit = 1, + offset = 0, + includeCompacted = true, + order = "latest", + ) + } + val body = (result as? NetworkResult.Success)?.data ?: return null + // The whole safety property rests on this single row being the HIGHEST id in + // the transcript. Require the gateway to confirm it honoured order=latest + // (same proof the chat hydration uses) — a backend that ignores the param + // answers with the OLDEST row, and a lower bound that is too low is exactly + // what lets a historical duplicate win. + if (body.pagination?.order != "latest") return null + return body.messages.mapNotNull { it.id }.maxOrNull() ?: 0 +} + +/** Newest-first REST page used to correlate a completed turn. */ +internal suspend fun fetchTurnRows(sessionId: String): List? { + val result = + safeApiCall { + ApiClient.hermesApi.getSessionMessages( + sessionId = sessionId, + limit = TURN_ROW_SCAN_LIMIT, + offset = 0, + includeCompacted = true, + order = "latest", + ) + } + return (result as? NetworkResult.Success)?.data?.messages +} + +/** + * Resolves the exact REST row persisted by the turn that just completed, given + * the boundary armed before that turn was submitted. + * + * The boundary is consumed either way in `finally`: the turn it describes is + * over, and leaving it armed would let the next completion bind to a row above + * a stale lower bound. + */ +internal suspend fun correlateCompletedTurnRow( + scopeId: String, + sessionId: String, + completionText: String, + resolver: TurnRowResolver = defaultTurnRowResolver, +): Int? { + val boundary = TurnCorrelationTracker.boundaryFor(scopeId, sessionId) ?: return null + return try { + resolver.resolve(sessionId, boundary, completionText) + } finally { + TurnCorrelationTracker.clearBoundary(scopeId, sessionId, boundary.generation) + } +} + +/** Production resolver — reads the newest REST page for the session. */ +internal val defaultTurnRowResolver = + TurnRowResolver(fetchLatestRows = { sessionId -> fetchTurnRows(sessionId) }) diff --git a/app/src/main/java/com/m57/hermescontrol/notification/TurnRowResolver.kt b/app/src/main/java/com/m57/hermescontrol/notification/TurnRowResolver.kt new file mode 100644 index 00000000..075ccef7 --- /dev/null +++ b/app/src/main/java/com/m57/hermescontrol/notification/TurnRowResolver.kt @@ -0,0 +1,93 @@ +package com.m57.hermescontrol.notification + +import com.m57.hermescontrol.data.model.SessionMessage +import com.m57.hermescontrol.ui.chat.HostMediaExtractor +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.delay + +/** + * Resolves the one persisted REST assistant row produced by a completed + * mobile-originated turn. + * + * Design rules (all of them are load-bearing): + * - Only rows strictly above [TurnBoundary.beforeMessageId] are candidates, so + * historical duplicates can never win. + * - The FULL completion text is compared, never the 100-character notification + * preview. + * - Exactly one candidate resolves. Zero candidates retry (the row may not be + * persisted yet). More than one candidate is ambiguous and fails closed — + * `firstOrNull`, `lastOrNull`, `maxBy` and "newest match" are all guessing. + * - Never compares Android notification time with server timestamps. + */ +internal class TurnRowResolver( + private val fetchLatestRows: suspend (String) -> List?, + /** Attempt delays; the first attempt runs immediately. */ + private val retryDelaysMs: List = listOf(0L, 200L, 400L), +) { + suspend fun resolve( + sessionId: String, + boundary: TurnBoundary, + completionText: String, + ): Int? { + val wanted = assistantTextKey(completionText) + if (wanted.isEmpty) return null + if (sessionId.isBlank()) return null + + for ((attempt, delayMs) in retryDelaysMs.withIndex()) { + if (attempt > 0) delay(delayMs) + val rows = fetchRows(sessionId) ?: continue + val candidates = + rows.mapNotNull { row -> + val id = row.id ?: return@mapNotNull null + if (id <= boundary.beforeMessageId) return@mapNotNull null + if (!row.role.equals("assistant", ignoreCase = true)) return@mapNotNull null + if (assistantTextKey(row.contentText) != wanted) return@mapNotNull null + id + } + if (candidates.size == 1) return candidates.single() + // Two rows above the boundary match this completion: the turn's own + // row cannot be told apart from the other. Leave the notification + // active rather than bind it to a coin flip. + if (candidates.size > 1) return null + } + return null + } + + private suspend fun fetchRows(sessionId: String): List? = + try { + fetchLatestRows(sessionId) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + null + } +} + +/** + * Normalized comparison key for an assistant row. + * + * Text replies compare their trimmed full text. `MEDIA:` replies compare the + * directive-stripped text PLUS the extracted host paths, using the same + * canonicalization the chat mapper applies before rendering — so a MEDIA row + * and the matching completion agree on both halves, and a MEDIA-only reply + * (which strips to an empty string) still compares by its paths instead of + * degenerating into "any assistant row". + */ +internal data class AssistantTextKey( + val text: String, + val media: List, +) { + val isEmpty: Boolean + get() = text.isBlank() && media.isEmpty() +} + +private const val MEDIA_DIRECTIVE = "MEDIA:" + +internal fun assistantTextKey(raw: String): AssistantTextKey { + val trimmed = raw.trim() + if (!trimmed.contains(MEDIA_DIRECTIVE)) return AssistantTextKey(trimmed, emptyList()) + return AssistantTextKey( + text = HostMediaExtractor.strip(trimmed).trim(), + media = HostMediaExtractor.extract(trimmed).map { it.path }, + ) +} diff --git a/app/src/main/java/com/m57/hermescontrol/ui/bots/group/GroupChatViewModel.kt b/app/src/main/java/com/m57/hermescontrol/ui/bots/group/GroupChatViewModel.kt index eb10325e..8d6e7fd5 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/bots/group/GroupChatViewModel.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/bots/group/GroupChatViewModel.kt @@ -696,6 +696,11 @@ class GroupChatViewModel( storedId?.let { inFlightTurns[it] = turnDeferred } try { + // Bot-group turns are orchestrated across per-bot sessions that + // are created and swapped mid-flight, so no clean pre-submit + // turn boundary can be proven for them. They stay uncorrelated + // (reply notifications from group turns are never auto-dismissed + // from REST hydration) instead of borrowing another turn's bound. HermesWsClient .request( WsMethods.PROMPT_SUBMIT, 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 783ddba0..3f48a358 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 @@ -85,6 +85,8 @@ data class ChatMessage( val tokenCount: Int? = null, /** Tokens per second generation speed (persisted to SQLite). */ val tps: Double? = null, + /** Stable completion identity correlating with reply notifications for read tracking. */ + val completionId: String? = null, ) /** diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatScreen.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatScreen.kt index 2c6d0adf..287ce2fb 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatScreen.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/ChatScreen.kt @@ -155,6 +155,7 @@ fun ChatScreen( val credentialWarning by HermesWsClient.credentialWarning.collectAsStateWithLifecycle() val connectorsViewModel: ChatConnectorsViewModel = viewModel() val connectorsState by connectorsViewModel.uiState.collectAsStateWithLifecycle() + val actionProgressState by viewModel.actionProgress.state.collectAsStateWithLifecycle() // Snapshot-backed search state — read directly so only the scopes that // read its fields recompose on search changes (bar, matched bubbles). val searchState = viewModel.searchState @@ -426,6 +427,20 @@ fun ChatScreen( context = context, ) + val isChatContentReadable = + !showReloginDialog && + !state.updateConfirmOpen && + !actionProgressState.visible && + !state.showModelPicker && + state.modelSwitchConfirmMessage == null && + state.sudoPrompt == null && + state.secretPrompt == null && + !(showContextSheet && state.contextBreakdown != null) && + !showSubagentInspectionSheet && + state.btwState == null && + viewingImage == null && + !connectorsState.isVisible + // Lifecycle effects, permissions, session switching, auto-scroll, errors ChatLifecycleEffects( sessionId = sessionId, @@ -442,6 +457,7 @@ fun ChatScreen( scrollController = scrollController, snackbarHostState = snackbarHostState, viewModel = viewModel, + isOverlayActive = !isChatContentReadable, ) HermesScaffold( 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 2ef7269b..b536fdae 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 @@ -5,6 +5,7 @@ import com.m57.hermescontrol.data.model.Attachment import com.m57.hermescontrol.data.model.AttachmentSource import com.m57.hermescontrol.data.model.SessionMessage import com.m57.hermescontrol.data.remote.GatewayFileClient +import com.m57.hermescontrol.notification.ReplyNotificationTracker /** * Maps REST transcript rows ([SessionMessage]) into UI [ChatMessage]s. @@ -20,12 +21,93 @@ internal fun mapServerMessages( offset: Int, latestPaging: Boolean, liveMessages: List, + isPagingOlder: Boolean = false, + context: android.content.Context? = null, ): List { val existingReasoningMap = liveMessages .filter { it.reasoningText.isNotBlank() } .associateBy { it.content } + val liveByExactId = + liveMessages + .filter { !it.completionId.isNullOrBlank() } + .associateBy { it.id } + + val unmappedLiveWsAssistants = + if (isPagingOlder) { + emptyList() + } else { + liveMessages + .filter { + it.role == MessageRole.ASSISTANT && !it.id.startsWith("rest-") && + !it.completionId.isNullOrBlank() + } + } + + val wsCompletionIdByRestIndex = mutableMapOf() + 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 + } + } + } + } + + // 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 exactIndex = + messages.indexOfFirst { message -> + message.id == activeTarget.serverMessageId && + message.role.equals("assistant", ignoreCase = true) + } + if (exactIndex >= 0 && !liveByExactId.containsKey("rest-$sessionId-${activeTarget.serverMessageId}")) { + wsCompletionIdByRestIndex[exactIndex] = activeTarget.completionId + } + } + } + // 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 @@ -171,6 +253,12 @@ internal fun mapServerMessages( } } + val completionId = + if (role == MessageRole.ASSISTANT) { + liveByExactId[restId]?.completionId ?: wsCompletionIdByRestIndex[index] + } else { + null + } val tokenCount = msg.tokenCount ?: TokenEstimator.estimate(finalContent).takeIf { it > 0 } mapped.add( ChatMessage( @@ -184,6 +272,7 @@ internal fun mapServerMessages( isStreaming = false, displayKind = msg.display_kind, tokenCount = tokenCount, + completionId = completionId, ), ) } 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 b4a87a99..f0029c79 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 @@ -28,6 +28,8 @@ import com.m57.hermescontrol.data.ws.HermesWsClient import com.m57.hermescontrol.data.ws.WsEvent import com.m57.hermescontrol.data.ws.WsMethods import com.m57.hermescontrol.data.ws.toJsonElement +import com.m57.hermescontrol.notification.captureTurnBoundary +import com.m57.hermescontrol.notification.correlationScopeId import com.m57.hermescontrol.ui.common.ActionProgressController import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers @@ -1879,6 +1881,15 @@ class ChatViewModel( }, ) } else { + // Arm the durable turn boundary BEFORE prompt.submit leaves + // the device: the REST high-watermark has to be read while + // the reply row cannot exist yet. A failed capture is + // non-fatal — the prompt is still sent, the turn just + // becomes uncorrelatable and its reply notification is never + // auto-dismissed from hydration instead of being bound to a + // guessed row. + prepareTurnCorrelation(storageSessionId) + if (dispatchGeneration != sessionGeneration) return@launch wsClient.sendMessage( agentSessionId, fullText, @@ -1898,6 +1909,17 @@ class ChatViewModel( } } + /** + * Arms the reply-notification turn boundary for a prompt about to be + * submitted. Best-effort by design: sending chat never depends on REST + * health, so a failed read only makes this turn's reply notification + * uncorrelatable — it stays active instead of being dismissed by a guess. + */ + private suspend fun prepareTurnCorrelation(storageSessionId: String?) { + if (storageSessionId.isNullOrBlank()) return + captureTurnBoundary(scopeId = correlationScopeId(), sessionId = storageSessionId) + } + private fun captureTurnUsageBaselineIfNeeded() { if (_streamingState.value.turnUsageBaselineCaptured) return _streamingState.update { @@ -2354,8 +2376,15 @@ class ChatViewModel( ) { if (text.isBlank()) return val sessionId = runtimeSessionId ?: return + val storageSessionId = _uiState.value.currentSessionId _uiState.update { it.copy(isAgentTyping = true) } viewModelScope.launch(ioDispatcher) { + // A queued prompt runs behind a turn that is already in flight, so + // no clean lower bound exists for it (see /queue). It stays + // uncorrelated rather than borrowing the running turn's boundary. + if (!queued) { + prepareTurnCorrelation(storageSessionId) + } wsClient.sendMessage( sessionId, text, @@ -2778,6 +2807,7 @@ class ChatViewModel( serverOffset, latestPaging, _uiState.value.messages, + context = getApplication(), ) loadedMessageOffset = serverOffset withContext(ioDispatcher) { @@ -3211,6 +3241,7 @@ class ChatViewModel( returnedOffset, latestPaging, _uiState.value.messages, + isPagingOlder = true, ) loadedMessageOffset = returnedOffset withContext(ioDispatcher) { repo.persistMessages(older, sessionId) } @@ -3295,6 +3326,7 @@ class ChatViewModel( nextOffset, latestPaging, _uiState.value.messages, + isPagingOlder = true, ) if (incoming.isEmpty()) return@launch withContext(ioDispatcher) { repo.persistMessages(incoming, sessionId) } 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 1fe44662..53c1b1da 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 @@ -429,9 +429,38 @@ object ChatWsEventReducer { // above before this return because the final snapshot is still needed // as the next turn's cumulative baseline. if (text.isBlank()) { + val effects = mutableListOf() + val sid = state.currentSessionId + var updatedTargetMsg: ChatMessage? = null + val updatedMessages = + if (!event.completionId.isNullOrBlank()) { + val lastAssistantIdx = + usageState.messages + .indexOfLast { + it.role == MessageRole.ASSISTANT && + (streamingState.sealedOrphanIds.contains(it.id) || it.id == streaming?.id) + }.takeIf { it >= 0 } + if (lastAssistantIdx != null) { + usageState.messages.toMutableList().also { list -> + val updated = list[lastAssistantIdx].copy(completionId = event.completionId) + list[lastAssistantIdx] = updated + updatedTargetMsg = updated + } + } else { + usageState.messages + } + } else { + usageState.messages + } + if (sid != null && updatedTargetMsg != null) { + effects.add(ReducerEffect.PersistMessage(updatedTargetMsg, sid)) + } + effects.add(ReducerEffect.RefreshSessions) + effects.add(ReducerEffect.RefreshContextUsage) return ReducerResult( - state = usageState.copy(isAgentTyping = false), + state = usageState.copy(messages = updatedMessages, isAgentTyping = false), streamingState = StreamingState(), + effects = effects, ) } val tps = turnUsage?.avgTps.validTpsOrNull() ?: usageState.latestTps.validTpsOrNull() @@ -449,6 +478,7 @@ object ChatWsEventReducer { finishTimestamp = System.currentTimeMillis(), tokenCount = tokenCount, tps = tps, + completionId = event.completionId, ) ?: ChatMessage( role = MessageRole.ASSISTANT, content = text, @@ -456,6 +486,7 @@ object ChatWsEventReducer { finishTimestamp = System.currentTimeMillis(), tokenCount = tokenCount, tps = tps, + completionId = event.completionId, ) val effects = mutableListOf() val sid = state.currentSessionId diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatLifecycleEffects.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatLifecycleEffects.kt index a45f6b12..e8cfd3bc 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatLifecycleEffects.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatLifecycleEffects.kt @@ -10,18 +10,24 @@ import androidx.compose.material3.SnackbarHostState import androidx.compose.runtime.Composable import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.LaunchedEffect +import androidx.compose.runtime.remember +import androidx.compose.runtime.snapshotFlow import androidx.compose.ui.platform.LocalContext import androidx.core.content.ContextCompat import androidx.lifecycle.Lifecycle import androidx.lifecycle.LifecycleEventObserver import androidx.lifecycle.compose.LocalLifecycleOwner +import androidx.lifecycle.repeatOnLifecycle import com.m57.hermescontrol.NavigationController import com.m57.hermescontrol.data.ws.ConnectionStatus +import com.m57.hermescontrol.notification.ReplyNotificationTracker +import com.m57.hermescontrol.notification.correlationScopeId import com.m57.hermescontrol.ui.chat.ChatMessage import com.m57.hermescontrol.ui.chat.ChatViewModel import com.m57.hermescontrol.ui.chat.ClarifyUi import com.m57.hermescontrol.ui.chat.SecretPromptUi import com.m57.hermescontrol.ui.chat.SudoPromptUi +import kotlinx.coroutines.flow.distinctUntilChanged @Composable fun ChatLifecycleEffects( @@ -39,6 +45,7 @@ fun ChatLifecycleEffects( scrollController: ChatScrollController, snackbarHostState: SnackbarHostState, viewModel: ChatViewModel, + isOverlayActive: Boolean = false, ) { val context = LocalContext.current val lifecycleOwner = LocalLifecycleOwner.current @@ -87,6 +94,32 @@ fun ChatLifecycleEffects( onDispose { lifecycleOwner.lifecycle.removeObserver(observer) } } + val messageMap = remember(messages) { messages.associateBy { it.id } } + + // Auto-dismiss reply notifications when their message is displayed in the viewport + LaunchedEffect(lifecycleOwner, currentSessionId, messageMap, listState, isOverlayActive) { + lifecycleOwner.lifecycle.repeatOnLifecycle(Lifecycle.State.RESUMED) { + snapshotFlow> { + if (currentSessionId.isNullOrBlank() || isOverlayActive) { + emptyList() + } else { + ChatReadObserver.findVisibleAssistantMessages(listState.layoutInfo, messageMap) + } + }.distinctUntilChanged() + .collect { visibleAssistantMsgs -> + val scopeId = correlationScopeId() + for (msg in visibleAssistantMsgs) { + ReplyNotificationTracker.onMessageVisible( + context = context, + scopeId = scopeId, + sessionId = currentSessionId, + completionId = msg.completionId, + ) + } + } + } + } + // Request POST_NOTIFICATIONS permission on Android 13+ val requestNotificationPermission = rememberLauncherForActivityResult( diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatReadObserver.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatReadObserver.kt new file mode 100644 index 00000000..99e3c68b --- /dev/null +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatReadObserver.kt @@ -0,0 +1,49 @@ +package com.m57.hermescontrol.ui.chat.components + +import androidx.compose.foundation.lazy.LazyListLayoutInfo +import com.m57.hermescontrol.ui.chat.ChatMessage +import com.m57.hermescontrol.ui.chat.MessageRole + +object ChatReadObserver { + /** + * Inspects the LazyColumn's visible items to determine which completed assistant + * messages currently have any portion visible within the viewport. + * + * In [com.m57.hermescontrol.ui.chat.fullbleed.FullBleedChatList], assistant prose + * bubbles are keyed with `"prose-${message.id}"`. An item is visible when its top + * edge is before the viewport end offset AND its bottom edge (offset + size) is past + * the viewport start offset. + */ + fun findVisibleAssistantMessages( + layoutInfo: LazyListLayoutInfo, + messageMap: Map, + ): List { + val visible = layoutInfo.visibleItemsInfo + if (visible.isEmpty() || messageMap.isEmpty()) return emptyList() + + val start = layoutInfo.viewportStartOffset + val end = layoutInfo.viewportEndOffset + + return visible + .filter { it.offset + it.size > start && it.offset < end } + .mapNotNull { item -> + val key = item.key as? String ?: return@mapNotNull null + if (key.startsWith("prose-")) { + val msgId = key.removePrefix("prose-") + val msg = messageMap[msgId] + if (msg != null && msg.role == MessageRole.ASSISTANT && !msg.isStreaming) { + msg + } else { + null + } + } else { + null + } + } + } + + fun findVisibleAssistantMessages( + layoutInfo: LazyListLayoutInfo, + messages: List, + ): List = findVisibleAssistantMessages(layoutInfo, messages.associateBy { it.id }) +} diff --git a/app/src/test/java/com/m57/hermescontrol/data/local/HermesDatabaseMigrationsTest.kt b/app/src/test/java/com/m57/hermescontrol/data/local/HermesDatabaseMigrationsTest.kt index 63a290fa..f57f20ac 100644 --- a/app/src/test/java/com/m57/hermescontrol/data/local/HermesDatabaseMigrationsTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/data/local/HermesDatabaseMigrationsTest.kt @@ -85,4 +85,16 @@ class HermesDatabaseMigrationsTest { ) } } + + @Test + fun migration7to8_executesCompletionIdColumnAdditionOnSQLiteConnection() = + runBlocking { + val connection = mockk(relaxed = true) + HermesDatabase.MIGRATION_7_8.migrate(connection) + verify(exactly = 1) { + connection.execSQL( + "ALTER TABLE `chat_messages` ADD COLUMN `completion_id` TEXT", + ) + } + } } diff --git a/app/src/test/java/com/m57/hermescontrol/notification/NotificationReplyReceiverTest.kt b/app/src/test/java/com/m57/hermescontrol/notification/NotificationReplyReceiverTest.kt index 744d1ac9..c287e05f 100644 --- a/app/src/test/java/com/m57/hermescontrol/notification/NotificationReplyReceiverTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/notification/NotificationReplyReceiverTest.kt @@ -25,6 +25,7 @@ import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test @@ -63,6 +64,7 @@ class NotificationReplyReceiverTest { fun setUp() { upsertCallCount = 0 ActiveSessionHolder.clear() + ReplyNotificationTracker.resetForTest() // Mock Android framework statics (same pattern as HermesWsClientTest) mockkStatic(android.util.Log::class) @@ -128,6 +130,7 @@ class NotificationReplyReceiverTest { @After fun tearDown() { ActiveSessionHolder.clear() + ReplyNotificationTracker.resetForTest() HermesDatabase.setForTest(null) unmockkAll() } @@ -186,6 +189,24 @@ class NotificationReplyReceiverTest { } } + @Test + fun `valid reply clears active reply notification target`() { + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = "prof-1", + sessionId = "session-abc", + completionId = "comp-1", + textSnippet = "old reply", + generation = 1L, + ) + assertNotNull(ReplyNotificationTracker.getActiveTarget()) + + givenValidReply("session-abc", "Hello") + receiver.onReceive(mockContext, mockIntent) + Thread.sleep(500) + + assertNull(ReplyNotificationTracker.getActiveTarget()) + } + @Test fun `valid reply finishes pending result after Room save`() { givenValidReply("session-abc", "Hello") diff --git a/app/src/test/java/com/m57/hermescontrol/notification/ReplyNotificationTrackerTest.kt b/app/src/test/java/com/m57/hermescontrol/notification/ReplyNotificationTrackerTest.kt new file mode 100644 index 00000000..3219a17d --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/notification/ReplyNotificationTrackerTest.kt @@ -0,0 +1,253 @@ +package com.m57.hermescontrol.notification + +import android.app.NotificationManager +import android.content.Context +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +class ReplyNotificationTrackerTest { + private lateinit var mockContext: Context + private lateinit var mockNotificationManager: NotificationManager + + @Before + fun setUp() { + mockContext = mockk(relaxed = true) + mockNotificationManager = mockk(relaxed = true) + every { mockContext.getSystemService(Context.NOTIFICATION_SERVICE) } returns mockNotificationManager + ReplyNotificationTracker.resetForTest() + } + + @After + fun tearDown() { + ReplyNotificationTracker.resetForTest() + } + + @Test + fun `onReplyNotificationPosted sets target correctly`() { + val gen = ReplyNotificationTracker.nextGeneration() + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-456", + textSnippet = "Hello world reply", + generation = gen, + ) + + val target = ReplyNotificationTracker.getActiveTarget() + assertEquals("prof-1", target?.scopeId) + assertEquals("sess-123", target?.sessionId) + assertEquals("comp-456", target?.completionId) + assertEquals("Hello world reply", target?.textSnippet) + assertEquals(gen, target?.generation) + } + + @Test + fun `matches verifies scope, session and completionId`() { + val target = + ReplyNotificationTarget( + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-456", + generation = 1L, + textSnippet = "Answer text", + ) + + assertTrue(target.matches("prof-1", "sess-123", "comp-456")) + // Scoped target rejects blank or null scope + assertFalse(target.matches("", "sess-123", "comp-456")) + assertFalse(target.matches(null, "sess-123", "comp-456")) + + // Target with blank scope matches unspecified scope + val unscopedTarget = target.copy(scopeId = "") + assertTrue(unscopedTarget.matches("", "sess-123", "comp-456")) + assertTrue(unscopedTarget.matches(null, "sess-123", "comp-456")) + + // Wrong session fails + assertFalse(target.matches("prof-1", "other-sess", "comp-456")) + // Wrong scope fails + assertFalse(target.matches("prof-2", "sess-123", "comp-456")) + // Wrong completionId fails + assertFalse(target.matches("prof-1", "sess-123", "other-comp")) + } + + @Test + fun `matches requires exact completionId match and rejects null or blank`() { + val target = + ReplyNotificationTarget( + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-456", + generation = 1L, + textSnippet = "Here is the full summary of the changes", + ) + + assertTrue( + target.matches( + candidateScopeId = "prof-1", + candidateSessionId = "sess-123", + candidateCompletionId = "comp-456", + ), + ) + + assertFalse( + "Missing candidate completionId must not match target", + target.matches( + candidateScopeId = "prof-1", + candidateSessionId = "sess-123", + candidateCompletionId = null, + ), + ) + + assertFalse( + "Different candidate completionId must not match target", + target.matches( + candidateScopeId = "prof-1", + candidateSessionId = "sess-123", + candidateCompletionId = "other-comp", + ), + ) + } + + @Test + fun `onNonReplyNotificationPosted clears target`() { + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-456", + textSnippet = "Reply", + generation = 1L, + ) + assertTrue(ReplyNotificationTracker.getActiveTarget() != null) + + ReplyNotificationTracker.onNonReplyNotificationPosted() + assertNull(ReplyNotificationTracker.getActiveTarget()) + } + + @Test + fun `onMessageVisible cancels notification on match and clears target`() { + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-456", + textSnippet = "Reply text", + generation = 1L, + ) + + val cancelled = + ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-456", + ) + + assertTrue(cancelled) + verify { mockNotificationManager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + assertNull(ReplyNotificationTracker.getActiveTarget()) + } + + @Test + fun `onMessageVisible does not cancel if session does not match`() { + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-456", + textSnippet = "Reply text", + generation = 1L, + ) + + val cancelled = + ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = "prof-1", + sessionId = "different-session", + completionId = "comp-456", + ) + + assertFalse(cancelled) + verify(inverse = true) { mockNotificationManager.cancel(any()) } + assertTrue(ReplyNotificationTracker.getActiveTarget() != null) + } + + @Test + fun `cancelReplyNotification refuses to cancel if newer generation was posted`() { + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-newer", + textSnippet = "Newer reply", + generation = 5L, + ) + + val cancelled = + ReplyNotificationTracker.cancelReplyNotification( + context = mockContext, + expectedGeneration = 4L, // older generation + ) + + assertFalse(cancelled) + verify(inverse = true) { mockNotificationManager.cancel(any()) } + } + + @Test + fun `cancelReplyNotification refuses to cancel if active notification is not a reply`() { + ReplyNotificationTracker.setTargetForTest(null) + ReplyNotificationTracker.activeNotificationProvider = { + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_ACTION, + scopeId = "prof-1", + sessionId = "sess-123", + completionId = "comp-1", + textSnippet = "Approval needed", + generation = 1L, + timestamp = 1000L, + ) + } + + val cancelled = + ReplyNotificationTracker.cancelReplyNotification( + context = mockContext, + expectedGeneration = 1L, + ) + + assertFalse(cancelled) + verify(inverse = true) { mockNotificationManager.cancel(any()) } + } + + @Test + fun `cold start recovery restores target from active notification extras`() { + ReplyNotificationTracker.setTargetForTest(null) + ReplyNotificationTracker.activeNotificationProvider = { + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "prof-recovered", + sessionId = "sess-recovered", + completionId = "comp-recovered", + textSnippet = "recovered snippet", + generation = 3L, + timestamp = 1000L, + ) + } + + val cancelled = + ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = "prof-recovered", + sessionId = "sess-recovered", + completionId = "comp-recovered", + ) + + assertTrue(cancelled) + verify { mockNotificationManager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + } +} diff --git a/app/src/test/java/com/m57/hermescontrol/notification/TurnCorrelationTrackerTest.kt b/app/src/test/java/com/m57/hermescontrol/notification/TurnCorrelationTrackerTest.kt new file mode 100644 index 00000000..f5f18794 --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/notification/TurnCorrelationTrackerTest.kt @@ -0,0 +1,411 @@ +package com.m57.hermescontrol.notification + +import com.m57.hermescontrol.data.model.SessionMessage +import com.m57.hermescontrol.data.remote.ApiClient +import io.mockk.every +import io.mockk.mockkObject +import io.mockk.unmockkObject +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.json.JsonPrimitive +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import java.io.IOException + +/** + * Turn-boundary correlation contract for reply notifications. + * + * The invariant under test: a completion may only be bound to a REST row that + * is provably inside its own turn — strictly above the pre-submit high-watermark + * — and only when exactly one such row matches. Everything else fails closed, + * because a notification that stays active is strictly better than one that is + * dismissed by the wrong duplicate reply. + * + * There is deliberately no clock anywhere in this file: Android notification + * time and Hermes server time come from different machines. + */ +class TurnCorrelationTrackerTest { + private val writeLog = mutableListOf>() + private var storedBoundaries: List = emptyList() + + @Before + fun setUp() { + writeLog.clear() + storedBoundaries = emptyList() + TurnCorrelationTracker.resetForTest() + TurnCorrelationTracker.attachStore(recordingStore()) + } + + @After + fun tearDown() { + TurnCorrelationTracker.resetForTest() + } + + // ── Harness ───────────────────────────────────────────────────────────── + + private fun recordingStore(): TurnBoundaryStore = + object : TurnBoundaryStore { + override fun read(): List = storedBoundaries + + override fun write(boundaries: List) { + storedBoundaries = boundaries + writeLog.add(boundaries) + } + } + + private fun row( + id: Int, + text: String, + role: String = "assistant", + ) = SessionMessage(id = id, role = role, content = JsonPrimitive(text)) + + private fun boundary( + beforeMessageId: Int, + scopeId: String = "default", + sessionId: String = "session", + ) = TurnBoundary( + scopeId = scopeId, + sessionId = sessionId, + beforeMessageId = beforeMessageId, + generation = 1L, + armedAt = 1L, + ) + + /** Resolver over a scripted sequence of REST pages; records how often it fetched. */ + private fun scriptedResolver( + pages: List>, + fetches: MutableList = mutableListOf(), + ): TurnRowResolver { + var index = 0 + return TurnRowResolver( + fetchLatestRows = { + fetches.add(index) + pages.getOrNull(index++) + }, + retryDelaysMs = listOf(0L, 0L, 0L), + ) + } + + private fun resolve( + pages: List>, + completionText: String, + beforeMessageId: Int = 0, + ): Int? = + runBlocking { + scriptedResolver(pages).resolve("session", boundary(beforeMessageId), completionText) + } + + // ── A/B: historical duplicates can never win ───────────────────────────── + + @Test + fun historicalDuplicateBelowBoundaryIsExcluded() { + val rows = listOf(row(5, "Done"), row(11, "Done")) + assertEquals(11, resolve(listOf(rows), "Done", beforeMessageId = 10)) + } + + @Test + fun oldDuplicateVisibleBeforeOwnRowAppearsIsNeverAccepted() { + // Attempt 1 sees only the historical duplicate; the turn's own row has + // not been persisted yet. Resolving on the first attempt would bind the + // notification to the wrong reply. + val fetches = mutableListOf() + val pages = + listOf( + listOf(row(5, "Done")), + listOf(row(5, "Done"), row(11, "Done")), + ) + val resolved = + runBlocking { + scriptedResolver(pages, fetches) + .resolve("session", boundary(beforeMessageId = 10), "Done") + } + assertEquals(11, resolved) + assertEquals(2, fetches.size) + } + + @Test + fun rowPersistedOnlyOnFinalAttemptStillResolves() { + val pages = + listOf( + listOf(row(5, "Done")), + listOf(row(5, "Done")), + listOf(row(5, "Done"), row(11, "Done")), + ) + assertEquals(11, resolve(pages, "Done", beforeMessageId = 10)) + } + + // ── C: ambiguity fails closed ──────────────────────────────────────────── + + @Test + fun twoMatchingRowsAboveBoundaryFailClosedWithoutRetrying() { + val fetches = mutableListOf() + val pages = listOf(listOf(row(11, "Done"), row(12, "Done")), listOf(row(11, "Done"), row(12, "Done"))) + val resolved = + runBlocking { + scriptedResolver(pages, fetches) + .resolve("session", boundary(beforeMessageId = 10), "Done") + } + assertNull("Ambiguous turn must not be guessed", resolved) + assertEquals("Ambiguity is terminal — no pointless retries", 1, fetches.size) + } + + @Test + fun nonMatchingContentAboveBoundaryFailsClosed() { + assertNull(resolve(listOf(listOf(row(11, "Other"))), "Done", beforeMessageId = 10)) + } + + @Test + fun userRowIsNotACandidate() { + assertNull(resolve(listOf(listOf(row(11, "Done", role = "user"))), "Done", beforeMessageId = 10)) + } + + @Test + fun blankCompletionTextNeverResolves() { + assertNull(resolve(listOf(listOf(row(11, "Done"))), " ", beforeMessageId = 10)) + } + + // ── Truncation / MEDIA ─────────────────────────────────────────────────── + + @Test + fun identicalHundredCharacterPrefixStillDisambiguates() { + val prefix = "x".repeat(100) + val rows = + listOf( + row(11, prefix + "A"), + row(12, prefix + "B"), + ) + // The full completion text is compared, never the 100-char preview, so + // the shared prefix cannot make the two rows interchangeable. + assertEquals(12, resolve(listOf(rows), prefix + "B", beforeMessageId = 10)) + } + + @Test + fun mediaReplyResolvesByCanonicalTextAndPath() { + val completion = "Here is the image MEDIA:/opt/hermes/image.png" + val rows = + listOf( + row(11, "Here is another image MEDIA:/opt/hermes/other.png"), + row(12, "Here is the image MEDIA:/opt/hermes/image.png"), + ) + assertEquals(12, resolve(listOf(rows), completion, beforeMessageId = 10)) + } + + @Test + fun mediaOnlyReplyResolvesByPathInsteadOfMatchingAnyAssistantRow() { + val rows = + listOf( + row(11, "something else entirely"), + row(12, "MEDIA:/opt/hermes/image.png"), + ) + assertEquals(12, resolve(listOf(rows), "MEDIA:/opt/hermes/image.png", beforeMessageId = 10)) + } + + @Test + fun duplicateMediaRowsAfterBoundaryFailClosed() { + val rows = + listOf( + row(11, "MEDIA:/opt/hermes/image.png"), + row(12, "MEDIA:/opt/hermes/image.png"), + ) + assertNull(resolve(listOf(rows), "MEDIA:/opt/hermes/image.png", beforeMessageId = 10)) + } + + // ── F: REST failure ────────────────────────────────────────────────────── + + @Test + fun restFailureFailsClosed() { + val resolver = + TurnRowResolver( + fetchLatestRows = { throw IOException("gateway unreachable") }, + retryDelaysMs = listOf(0L, 0L, 0L), + ) + assertNull(runBlocking { resolver.resolve("session", boundary(10), "Done") }) + } + + @Test + fun emptyPagesFailClosed() { + assertNull(resolve(listOf(emptyList(), emptyList()), "Done", beforeMessageId = 10)) + } + + // ── G/H/I: missing, cross-profile and cross-session boundaries ─────────── + + @Test + fun noBoundaryMeansNoLookupAtAll() { + val fetches = mutableListOf() + val resolved = + runBlocking { + correlateCompletedTurnRow( + scopeId = "default", + sessionId = "session", + completionText = "Done", + resolver = scriptedResolver(listOf(listOf(row(11, "Done"))), fetches), + ) + } + assertNull(resolved) + assertTrue("Uncorrelatable turns must not even hit REST", fetches.isEmpty()) + } + + @Test + fun boundaryFromAnotherProfileIsInvisible() { + TurnCorrelationTracker.armBoundary("other-profile", "session", 10) + val fetches = mutableListOf() + val resolved = + runBlocking { + correlateCompletedTurnRow( + scopeId = "default", + sessionId = "session", + completionText = "Done", + resolver = scriptedResolver(listOf(listOf(row(11, "Done"))), fetches), + ) + } + assertNull(resolved) + assertTrue(fetches.isEmpty()) + } + + @Test + fun boundaryFromAnotherSessionIsInvisible() { + TurnCorrelationTracker.armBoundary("default", "other-session", 10) + val fetches = mutableListOf() + val resolved = + runBlocking { + correlateCompletedTurnRow( + scopeId = "default", + sessionId = "session", + completionText = "Done", + resolver = scriptedResolver(listOf(listOf(row(11, "Done"))), fetches), + ) + } + assertNull(resolved) + assertTrue(fetches.isEmpty()) + } + + // ── Boundary lifecycle ─────────────────────────────────────────────────── + + @Test + fun correlateResolvesThenConsumesTheBoundary() { + TurnCorrelationTracker.armBoundary("default", "session", 10) + val pages = listOf(listOf(row(11, "Done"))) + val first = + runBlocking { + correlateCompletedTurnRow("default", "session", "Done", scriptedResolver(pages)) + } + assertEquals(11, first) + assertNull( + "The turn is over — its boundary must be spent", + TurnCorrelationTracker.boundaryFor("default", "session"), + ) + + val second = + runBlocking { + correlateCompletedTurnRow("default", "session", "Done", scriptedResolver(pages)) + } + assertNull("A second completion must not reuse the consumed boundary", second) + } + + @Test + fun armingPersistsAndSurvivesProcessRestart() { + TurnCorrelationTracker.armBoundary("default", "session", 41) + assertEquals(41, storedBoundaries.single().beforeMessageId) + + // Simulate a fresh process: same store content, empty in-memory cache. + TurnCorrelationTracker.resetForTest() + TurnCorrelationTracker.attachStore(recordingStore()) + assertEquals(41, TurnCorrelationTracker.boundaryFor("default", "session")?.beforeMessageId) + } + + /** + * The generation counter is in-memory only, so a restart must lift it above + * whatever the restored boundaries already used. Otherwise an old + * completion's clear (generation N) matches the NEW boundary (also N) and the + * stale-clear guard stops protecting the newer turn. + */ + @Test + fun restoredBoundaryGenerationAdvancesTheCounterPastIt() { + storedBoundaries = + listOf( + TurnBoundary( + scopeId = "default", + sessionId = "session", + beforeMessageId = 10, + generation = 37L, + armedAt = System.currentTimeMillis(), + ), + ) + TurnCorrelationTracker.resetForTest() + TurnCorrelationTracker.attachStore(recordingStore()) + assertEquals(37L, TurnCorrelationTracker.boundaryFor("default", "session")?.generation) + + val next = TurnCorrelationTracker.armBoundary("default", "session", 25) + assertNotNull(next) + assertTrue( + "A post-restart boundary must not reuse a recovered generation (got ${next?.generation})", + (next?.generation ?: 0L) > 37L, + ) + + // The old completion, still holding generation 37, must not clear it. + assertFalse(TurnCorrelationTracker.clearBoundary("default", "session", 37L)) + assertEquals(25, TurnCorrelationTracker.boundaryFor("default", "session")?.beforeMessageId) + } + + @Test + fun reArmingForTheSameSessionReplacesTheBoundary() { + TurnCorrelationTracker.armBoundary("default", "session", 10) + TurnCorrelationTracker.armBoundary("default", "session", 25) + assertEquals(25, TurnCorrelationTracker.boundaryFor("default", "session")?.beforeMessageId) + assertEquals(1, storedBoundaries.size) + } + + @Test + fun staleGenerationCannotClearANewerBoundary() { + val stale = TurnCorrelationTracker.armBoundary("default", "session", 10) + TurnCorrelationTracker.armBoundary("default", "session", 25) + assertNotNull(stale) + assertFalse(TurnCorrelationTracker.clearBoundary("default", "session", stale?.generation)) + assertEquals(25, TurnCorrelationTracker.boundaryFor("default", "session")?.beforeMessageId) + } + + @Test + fun blankIdentityOrNegativeBoundaryNeverArms() { + assertNull(TurnCorrelationTracker.armBoundary("", "session", 10)) + assertNull(TurnCorrelationTracker.armBoundary("default", " ", 10)) + assertNull(TurnCorrelationTracker.armBoundary("default", "session", -1)) + assertTrue(writeLog.isEmpty()) + } + + @Test + fun zeroBoundaryIsValidForAnEmptyTranscript() { + // A brand-new session has no rows yet: the next turn's rows are all + // above 0, and there are no historical duplicates to exclude. + assertNotNull(TurnCorrelationTracker.armBoundary("default", "session", 0)) + assertEquals(0, TurnCorrelationTracker.boundaryFor("default", "session")?.beforeMessageId) + } + + /** + * The probe is best-effort by contract and must never fail the caller's turn. + * + * A gateway-layer `Error` (uninitialised client, NoClassDefFound) is not an + * `Exception`, so it used to escape the caller's `catch (Exception)`, abort + * the notification-reply send, and leak out of a background scope — which CI + * then reported as `UncaughtExceptionsBeforeTest` in an unrelated later test + * class. Encode that as a regression so it cannot come back. + */ + @Test + fun boundaryProbeSwallowsEvenGatewayLayerErrors() { + mockkObject(ApiClient) + try { + every { ApiClient.hermesApi } throws ExceptionInInitializerError("ApiClient unavailable") + assertFalse( + "A broken gateway layer must not fail the turn", + runBlocking { captureTurnBoundary("default", "session") }, + ) + assertNull(TurnCorrelationTracker.boundaryFor("default", "session")) + } finally { + unmockkObject(ApiClient) + } + } +} diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatReadNotificationIntegrationTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatReadNotificationIntegrationTest.kt new file mode 100644 index 00000000..3baa2d43 --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatReadNotificationIntegrationTest.kt @@ -0,0 +1,296 @@ +package com.m57.hermescontrol.ui.chat + +import android.app.NotificationManager +import android.content.Context +import androidx.compose.foundation.lazy.LazyListItemInfo +import androidx.compose.foundation.lazy.LazyListLayoutInfo +import com.m57.hermescontrol.notification.ChatNotificationService +import com.m57.hermescontrol.notification.ReplyNotificationTracker +import com.m57.hermescontrol.ui.chat.components.ChatReadObserver +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import org.junit.After +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +class ChatReadNotificationIntegrationTest { + private lateinit var mockContext: Context + private lateinit var mockNotificationManager: NotificationManager + + @Before + fun setUp() { + mockContext = mockk(relaxed = true) + mockNotificationManager = mockk(relaxed = true) + every { mockContext.getSystemService(Context.NOTIFICATION_SERVICE) } returns mockNotificationManager + ReplyNotificationTracker.resetForTest() + } + + @After + fun tearDown() { + ReplyNotificationTracker.resetForTest() + } + + private fun createMockItem( + key: Any, + offset: Int, + size: Int, + ): LazyListItemInfo { + val item = mockk() + every { item.key } returns key + every { item.offset } returns offset + every { item.size } returns size + return item + } + + private fun createMockLayoutInfo( + items: List, + startOffset: Int = 0, + endOffset: Int = 1000, + ): LazyListLayoutInfo { + val layout = mockk() + every { layout.visibleItemsInfo } returns items + every { layout.viewportStartOffset } returns startOffset + every { layout.viewportEndOffset } returns endOffset + return layout + } + + @Test + fun `end-to-end flow - visible target message cancels notification`() { + val scopeId = "profile-main" + val sessionId = "session-42" + val completionId = "completion-abc" + val textSnippet = "Here is the result of your task" + + val generation = ReplyNotificationTracker.nextGeneration() + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = scopeId, + sessionId = sessionId, + completionId = completionId, + textSnippet = textSnippet, + generation = generation, + ) + + val targetMessage = + ChatMessage( + id = "msg-target", + role = MessageRole.ASSISTANT, + content = textSnippet, + completionId = completionId, + ) + + val layout = + createMockLayoutInfo( + listOf( + createMockItem("prose-msg-target", offset = 200, size = 150), + ), + ) + + val visibleMessages = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(targetMessage)) + var wasCancelled = false + for (msg in visibleMessages) { + if (ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = scopeId, + sessionId = sessionId, + completionId = msg.completionId, + ) + ) { + wasCancelled = true + } + } + + assertTrue(wasCancelled) + verify { mockNotificationManager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + } + + @Test + fun `scrolled above target - notification stays until scrolled to target`() { + val scopeId = "profile-main" + val sessionId = "session-42" + val completionId = "completion-abc" + val textSnippet = "Newest assistant reply" + + val generation = ReplyNotificationTracker.nextGeneration() + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = scopeId, + sessionId = sessionId, + completionId = completionId, + textSnippet = textSnippet, + generation = generation, + ) + + val olderMessage = + ChatMessage( + id = "msg-older", + role = MessageRole.ASSISTANT, + content = "Old assistant message", + completionId = "comp-old", + ) + val targetMessage = + ChatMessage( + id = "msg-target", + role = MessageRole.ASSISTANT, + content = textSnippet, + completionId = completionId, + ) + val allMessages = listOf(olderMessage, targetMessage) + + // Viewport 0..1000: user is at the top, only older message is visible; target is at offset 1500 (below viewport) + val scrolledUpLayout = + createMockLayoutInfo( + listOf( + createMockItem("prose-msg-older", offset = 100, size = 200), + createMockItem("prose-msg-target", offset = 1500, size = 300), + ), + startOffset = 0, + endOffset = 1000, + ) + + val visibleWhenScrolledUp = ChatReadObserver.findVisibleAssistantMessages(scrolledUpLayout, allMessages) + var wasCancelledWhenScrolledUp = false + for (msg in visibleWhenScrolledUp) { + if (ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = scopeId, + sessionId = sessionId, + completionId = msg.completionId, + ) + ) { + wasCancelledWhenScrolledUp = true + } + } + + assertFalse(wasCancelledWhenScrolledUp) + verify(inverse = true) { mockNotificationManager.cancel(any()) } + + // Now user scrolls down: target message enters viewport (offset 400, size 300) + val scrolledDownLayout = + createMockLayoutInfo( + listOf( + createMockItem("prose-msg-target", offset = 400, size = 300), + ), + startOffset = 0, + endOffset = 1000, + ) + + val visibleWhenScrolledDown = ChatReadObserver.findVisibleAssistantMessages(scrolledDownLayout, allMessages) + var wasCancelledWhenScrolledDown = false + for (msg in visibleWhenScrolledDown) { + if (ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = scopeId, + sessionId = sessionId, + completionId = msg.completionId, + ) + ) { + wasCancelledWhenScrolledDown = true + } + } + + assertTrue(wasCancelledWhenScrolledDown) + verify { mockNotificationManager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + } + + @Test + fun `viewing a different session does not dismiss notification`() { + val scopeId = "profile-main" + val sessionIdWithNotif = "session-notif" + val completionId = "completion-abc" + val textSnippet = "Important reply" + + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = scopeId, + sessionId = sessionIdWithNotif, + completionId = completionId, + textSnippet = textSnippet, + generation = 1L, + ) + + val messageInOtherSession = + ChatMessage( + id = "msg-other", + role = MessageRole.ASSISTANT, + content = "Some other reply", + completionId = "other-comp", + ) + + val layout = + createMockLayoutInfo( + listOf( + createMockItem("prose-msg-other", offset = 200, size = 150), + ), + ) + + val visibleMessages = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(messageInOtherSession)) + var wasCancelled = false + for (msg in visibleMessages) { + if (ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = scopeId, + sessionId = "different-session-id", + completionId = msg.completionId, + ) + ) { + wasCancelled = true + } + } + + assertFalse(wasCancelled) + verify(inverse = true) { mockNotificationManager.cancel(any()) } + } + + @Test + fun `action notification replacement prevents dismissal by message read`() { + val scopeId = "profile-main" + val sessionId = "session-42" + val completionId = "completion-abc" + val textSnippet = "Reply before action" + + // 1. Reply notification posted + ReplyNotificationTracker.onReplyNotificationPosted( + scopeId = scopeId, + sessionId = sessionId, + completionId = completionId, + textSnippet = textSnippet, + generation = 1L, + ) + + // 2. Later, an approval request arrives and replaces notification ID 2 + ReplyNotificationTracker.onNonReplyNotificationPosted() + + // 3. User opens chat and views the previous reply message + val targetMessage = + ChatMessage( + id = "msg-target", + role = MessageRole.ASSISTANT, + content = textSnippet, + completionId = completionId, + ) + val layout = + createMockLayoutInfo( + listOf( + createMockItem("prose-msg-target", offset = 200, size = 150), + ), + ) + + val visibleMessages = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(targetMessage)) + var wasCancelled = false + for (msg in visibleMessages) { + if (ReplyNotificationTracker.onMessageVisible( + context = mockContext, + scopeId = scopeId, + sessionId = sessionId, + completionId = msg.completionId, + ) + ) { + wasCancelled = true + } + } + + assertFalse("Action notification must NOT be dismissed by reading chat message", wasCancelled) + verify(inverse = true) { mockNotificationManager.cancel(any()) } + } +} 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 2874b7c2..7c072688 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 @@ -10,6 +10,9 @@ import com.m57.hermescontrol.data.local.AuthManager import com.m57.hermescontrol.data.local.HermesDatabase import com.m57.hermescontrol.data.model.Attachment import com.m57.hermescontrol.data.model.AttachmentSource +import com.m57.hermescontrol.data.model.PaginationInfo +import com.m57.hermescontrol.data.model.SessionMessage +import com.m57.hermescontrol.data.model.SessionMessagesResponse import com.m57.hermescontrol.data.remote.ApiClient import com.m57.hermescontrol.data.remote.GatewayFile import com.m57.hermescontrol.data.remote.GatewayFileClient @@ -21,6 +24,7 @@ import com.m57.hermescontrol.data.ws.HermesWsClient import com.m57.hermescontrol.data.ws.JsonRpcError import com.m57.hermescontrol.data.ws.WsEvent import com.m57.hermescontrol.data.ws.WsMethods +import com.m57.hermescontrol.notification.TurnCorrelationTracker import com.m57.hermescontrol.ui.chat.fakes.FakeChatPersistenceRepository import com.m57.hermescontrol.ui.chat.fakes.FakeSlashUsageStore import io.mockk.* @@ -52,6 +56,7 @@ import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.jsonObject import kotlinx.serialization.json.jsonPrimitive import kotlinx.serialization.json.put +import okhttp3.ResponseBody.Companion.toResponseBody import org.junit.After import org.junit.Assert.assertArrayEquals import org.junit.Assert.assertEquals @@ -5590,4 +5595,234 @@ class ChatViewModelTest { io.mockk.verify { AuthManager.setLastOpenedSessionId("session-other-456") } } + + // ── Turn-boundary capture ordering (reply-notification correlation) ─────── + + /** + * Turn boundaries are keyed by the active profile, so these tests need one. + * Scoped per test rather than added to setUp: a non-blank profile is also + * appended to `session.resume` params, which the older resume/switch tests + * assert exactly. + */ + private fun stubActiveProfile() { + every { AuthManager.activeProfileId } returns MutableStateFlow("default") + } + + /** + * The REST high-watermark must be read BEFORE prompt.submit leaves the + * device. Capturing it afterwards lets a fast turn persist its assistant row + * first, and a lower bound that already contains the reply can never exclude + * a historical duplicate. + */ + @Test + fun sendMessage_readsTurnBoundaryBeforeSubmittingPrompt() = + runTest { + TurnCorrelationTracker.resetForTest() + stubActiveProfile() + val (viewModel, sessionId) = createViewModelWithSession() + val events = mutableListOf() + val api = mockk(relaxed = true) + every { ApiClient.hermesApi } returns api + coEvery { api.getSessionMessages(any(), any(), any(), any(), any(), any()) } answers { + // Only the boundary probe asks for a single newest row. + if (arg(1) == 1 && arg(3) == "latest") events += "boundary" + retrofit2.Response.success( + SessionMessagesResponse( + messages = listOf(SessionMessage(id = 41, role = "user", content = JsonPrimitive("hi"))), + pagination = PaginationInfo(limit = 1, offset = 0, order = "latest", returned = 1), + ), + ) + } + every { HermesWsClient.sendMessage(any(), any(), any(), any()) } answers { + events += "submit" + reqCount++ + val id = "req-msg-$reqCount" + arg<((String) -> Unit)?>(2)?.invoke(id) + id + } + + viewModel.sendMessage("hello there") + advanceUntilIdle() + + assertEquals(listOf("boundary", "submit"), events) + assertEquals(41, TurnCorrelationTracker.boundaryFor("default", sessionId)?.beforeMessageId) + } + + @Test + fun sendMessage_stillSubmitsWhenBoundaryReadFails() = + runTest { + TurnCorrelationTracker.resetForTest() + stubActiveProfile() + val (viewModel, sessionId) = createViewModelWithSession() + val api = mockk(relaxed = true) + every { ApiClient.hermesApi } returns api + coEvery { api.getSessionMessages(any(), any(), any(), any(), any(), any()) } returns + retrofit2.Response.error(500, "boom".toResponseBody()) + var submits = 0 + every { HermesWsClient.sendMessage(any(), any(), any(), any()) } answers { + submits++ + reqCount++ + val id = "req-msg-$reqCount" + arg<((String) -> Unit)?>(2)?.invoke(id) + id + } + + viewModel.sendMessage("hello there") + advanceUntilIdle() + + assertEquals("REST health must never block sending chat", 1, submits) + assertNull(TurnCorrelationTracker.boundaryFor("default", sessionId)) + } + + /** + * A gateway that ignores `order=latest` answers a `limit=1` probe with the + * OLDEST row; trusting that number would arm a boundary that is far too low + * and let historical duplicates win. Without the pagination proof the turn is + * left uncorrelatable, and the prompt still goes out. + */ + @Test + fun sendMessage_leavesTurnUncorrelatableWhenGatewayCannotConfirmOrder() = + runTest { + TurnCorrelationTracker.resetForTest() + stubActiveProfile() + val (viewModel, sessionId) = createViewModelWithSession() + val api = mockk(relaxed = true) + every { ApiClient.hermesApi } returns api + coEvery { api.getSessionMessages(any(), any(), any(), any(), any(), any()) } answers { + retrofit2.Response.success( + SessionMessagesResponse( + messages = listOf(SessionMessage(id = 1, role = "user", content = JsonPrimitive("oldest"))), + ), + ) + } + var submits = 0 + every { HermesWsClient.sendMessage(any(), any(), any(), any()) } answers { + submits++ + reqCount++ + val id = "req-msg-$reqCount" + arg<((String) -> Unit)?>(2)?.invoke(id) + id + } + + viewModel.sendMessage("hello there") + advanceUntilIdle() + + assertEquals(1, submits) + assertNull( + "Unconfirmed row order must not produce a boundary", + TurnCorrelationTracker.boundaryFor("default", sessionId), + ) + } + + /** + * `AuthManager.activeProfileId` is null until a server profile is explicitly + * selected, and the rest of AuthManager treats that as + * [AuthManager.DEFAULT_PROFILE_ID]. A raw `.orEmpty()` made the whole feature + * a no-op on a normal install: no boundary armed, so no reply notification + * could ever be REST-auto-dismissed. + */ + @Test + fun sendMessage_nullActiveProfileArmsBoundaryUnderTheDefaultScope() = + runTest { + TurnCorrelationTracker.resetForTest() + every { AuthManager.activeProfileId } returns MutableStateFlow(null) + val (viewModel, sessionId) = createViewModelWithSession() + val api = mockk(relaxed = true) + every { ApiClient.hermesApi } returns api + coEvery { api.getSessionMessages(any(), any(), any(), any(), any(), any()) } answers { + retrofit2.Response.success( + SessionMessagesResponse( + messages = listOf(SessionMessage(id = 41, role = "user", content = JsonPrimitive("hi"))), + pagination = PaginationInfo(limit = 1, offset = 0, order = "latest", returned = 1), + ), + ) + } + var submits = 0 + every { HermesWsClient.sendMessage(any(), any(), any(), any()) } answers { + submits++ + reqCount++ + val id = "req-msg-$reqCount" + arg<((String) -> Unit)?>(2)?.invoke(id) + id + } + + viewModel.sendMessage("hello there") + advanceUntilIdle() + + assertEquals(1, submits) + assertEquals( + "A default-profile install must still be correlatable", + 41, + TurnCorrelationTracker.boundaryFor(AuthManager.DEFAULT_PROFILE_ID, sessionId)?.beforeMessageId, + ) + } + + /** + * `pagination.order` is the only proof the gateway honoured `order=latest`. + * A gateway that reports `oldest` answered the `limit=1` probe with the + * OLDEST row, and that as a lower bound would sit below the whole transcript + * — exactly what lets a historical duplicate win. + */ + @Test + fun sendMessage_leavesTurnUncorrelatableWhenGatewayReportsOldestOrder() = + runTest { + TurnCorrelationTracker.resetForTest() + stubActiveProfile() + val (viewModel, sessionId) = createViewModelWithSession() + val api = mockk(relaxed = true) + every { ApiClient.hermesApi } returns api + coEvery { api.getSessionMessages(any(), any(), any(), any(), any(), any()) } answers { + retrofit2.Response.success( + SessionMessagesResponse( + messages = listOf(SessionMessage(id = 1, role = "user", content = JsonPrimitive("oldest"))), + pagination = PaginationInfo(limit = 1, offset = 0, order = "oldest", returned = 1), + ), + ) + } + var submits = 0 + every { HermesWsClient.sendMessage(any(), any(), any(), any()) } answers { + submits++ + reqCount++ + val id = "req-msg-$reqCount" + arg<((String) -> Unit)?>(2)?.invoke(id) + id + } + + viewModel.sendMessage("hello there") + advanceUntilIdle() + + assertEquals(1, submits) + assertNull( + "An unproven row order must not produce a boundary", + TurnCorrelationTracker.boundaryFor("default", sessionId), + ) + } + + /** + * A queued prompt runs behind a turn that is already in flight, so no clean + * lower bound exists for it — it must not borrow the running turn's boundary. + */ + @Test + fun queuedPromptNeverArmsATurnBoundary() = + runTest { + TurnCorrelationTracker.resetForTest() + stubActiveProfile() + val (viewModel, sessionId) = createViewModelWithSession() + val api = mockk(relaxed = true) + every { ApiClient.hermesApi } returns api + every { HermesWsClient.sendMessage(any(), any(), any(), any()) } answers { + reqCount++ + val id = "req-msg-$reqCount" + arg<((String) -> Unit)?>(2)?.invoke(id) + id + } + + viewModel.sendMessage("/queue do the thing") + advanceUntilIdle() + + io.mockk.coVerify(exactly = 0) { + api.getSessionMessages(any(), any(), any(), any(), any(), any()) + } + assertNull(TurnCorrelationTracker.boundaryFor("default", sessionId)) + } } diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt index 2b9543c1..7b5814ae 100644 --- a/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ChatWsEventReducerTest.kt @@ -450,6 +450,25 @@ class ChatWsEventReducerTest { ) } + @Test + fun testMessageComplete_setsCompletionIdOnAssistantMessage() { + val event = + WsEvent.MessageComplete( + text = "hello from assistant", + sessionId = "session-1", + completionId = "comp-123", + ) + val result = + ChatWsEventReducer.reduce( + state = ChatUiState(), + streamingState = StreamingState(), + event = event, + currentSessionId = "session-1", + ) + val msg = result.state.messages.single() + assertEquals("comp-123", msg.completionId) + } + @Test fun testMessageStart_prunesCompletedSubagents() { val completedSubagent = diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/ReadNotificationReviewRegressionTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/ReadNotificationReviewRegressionTest.kt new file mode 100644 index 00000000..12fe98ca --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/ReadNotificationReviewRegressionTest.kt @@ -0,0 +1,807 @@ +package com.m57.hermescontrol.ui.chat + +import android.app.Notification +import android.app.NotificationManager +import android.content.Context +import com.m57.hermescontrol.data.config.ServerStore +import com.m57.hermescontrol.data.config.ServerStoreState +import com.m57.hermescontrol.data.local.AuthManager +import com.m57.hermescontrol.data.local.toEntity +import com.m57.hermescontrol.data.local.toUiModel +import com.m57.hermescontrol.data.model.SessionMessage +import com.m57.hermescontrol.data.ws.WsEvent +import com.m57.hermescontrol.notification.ActiveReplyInfo +import com.m57.hermescontrol.notification.ChatNotificationService +import com.m57.hermescontrol.notification.ReplyNotificationTarget +import com.m57.hermescontrol.notification.ReplyNotificationTracker +import com.m57.hermescontrol.notification.TurnCorrelationTracker +import com.m57.hermescontrol.notification.TurnRowResolver +import com.m57.hermescontrol.notification.correlateCompletedTurnRow +import io.mockk.every +import io.mockk.mockk +import io.mockk.mockkObject +import io.mockk.unmockkObject +import io.mockk.verify +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonPrimitive +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference + +/** + * Regression matrix for reply-notification read dismissal. + * + * Split by design: + * - The `alreadyResolvedServerId*` cases start from a durable server row id that + * the notification already carries, and prove hydration binds it to that exact + * row only — the downstream half of the contract. + * - [mobileTurnBoundaryResolvesTheExactRowThenOnlyThatRowDismisses] covers the + * production path that PRODUCES that id (turn boundary → unique row). + * - The mobile-side resolution rules themselves live in + * `TurnCorrelationTrackerTest`, and the send-order guarantee in + * `ChatViewModelTest`. + * + * Nothing here compares Android notification time with Hermes server time; + * server row timestamps are not identity. + */ +class ReadNotificationReviewRegressionTest { + private lateinit var context: Context + private lateinit var manager: NotificationManager + + @Before + fun setUp() { + context = mockk(relaxed = true) + manager = mockk(relaxed = true) + every { context.getSystemService(Context.NOTIFICATION_SERVICE) } returns manager + mockkObject(AuthManager) + val mockStore = mockk(relaxed = true) + every { mockStore.getLatestState() } returns ServerStoreState(baseUrl = "http://localhost:8080") + every { AuthManager.serverStore } returns mockStore + every { AuthManager.getToken() } returns "token" + every { AuthManager.activeProfileId } returns MutableStateFlow("default") + ReplyNotificationTracker.resetForTest() + } + + @After + fun tearDown() { + ReplyNotificationTracker.resetForTest() + unmockkObject(AuthManager) + } + + private fun postTarget(text: String = "Done") { + ReplyNotificationTracker.onReplyNotificationPosted("default", "session", "new-completion", text, 1L) + } + + @Test + fun roomRoundTripMustRetainCompletionIdentity() { + val live = ChatMessage(role = MessageRole.ASSISTANT, content = "Done", completionId = "new-completion") + val restored = live.toEntity("session").toUiModel() + assertEquals("Room reload lost notification identity", "new-completion", restored.completionId) + } + + @Test + fun repeatedTextInHistoryMustNotCopyLatestIdentityOntoOlderMessage() { + val live = + listOf( + ChatMessage(role = MessageRole.ASSISTANT, content = "Done", completionId = "old-completion"), + ChatMessage(role = MessageRole.ASSISTANT, content = "Done", completionId = "new-completion"), + ) + val history = + listOf( + SessionMessage(id = 100, role = "assistant", content = JsonPrimitive("Done")), + SessionMessage(id = 200, role = "assistant", content = JsonPrimitive("Done")), + ) + val mapped = mapServerMessages("session", history, 0, true, live) + assertEquals(listOf("old-completion", "new-completion"), mapped.map { it.completionId }) + } + + @Test + fun viewingOldCachedTextMustNotClearNewReplyNotification() { + postTarget("Done") + val old = + ChatMessage(role = MessageRole.ASSISTANT, content = "Done yesterday; unrelated old response") + .toEntity("session") + .toUiModel() + val cancelled = + ReplyNotificationTracker.onMessageVisible( + context, + "default", + "session", + old.completionId, + ) + assertFalse("Old text incorrectly acknowledged the new reply", cancelled) + verify(exactly = 0) { manager.cancel(any()) } + } + + @Test + fun strippedCommentaryCompletionMustBindTerminalVisibleMessage() { + val orphan = ChatMessage(id = "orphan", role = MessageRole.ASSISTANT, content = "already shown") + val state = ChatUiState(currentSessionId = "session", messages = listOf(orphan)) + val streaming = StreamingState(streamingMessage = orphan, sealedOrphanIds = listOf("orphan")) + val result = + ChatWsEventReducer.reduce( + state, + streaming, + WsEvent.MessageComplete("already shown", "session", completionId = "new-completion"), + "session", + ) + assertEquals( + "Completion lost when no new final bubble is needed", + "new-completion", + result.state.messages + .single() + .completionId, + ) + val persistEffect = + result.effects.filterIsInstance().singleOrNull() + assertEquals( + "PersistMessage effect must be emitted for the sealed orphan bubble so Room persists it", + "new-completion", + persistEffect?.message?.completionId, + ) + } + + @Test + fun loadOlderMessagesMustNotStealCompletionIdFromLiveMessage() { + val live = + listOf( + ChatMessage( + id = "ws-1", + role = MessageRole.ASSISTANT, + content = "Done", + completionId = "new-completion", + ), + ) + val olderHistory = + listOf( + SessionMessage(id = 50, role = "assistant", content = JsonPrimitive("Done")), + ) + val mapped = mapServerMessages("session", olderHistory, 50, true, live, isPagingOlder = true) + assertEquals("Older paged message must not steal live completionId", null, mapped.single().completionId) + } + + @Test + fun duplicateTextOutsideFetchedPageMustNotShiftCompletionId() { + val live = + listOf( + ChatMessage( + id = "rest-session-10", + role = MessageRole.ASSISTANT, + content = "Done", + completionId = null, + ), + ChatMessage( + id = "rest-session-20", + role = MessageRole.ASSISTANT, + content = "Done", + completionId = "new-completion", + ), + ) + // Page only includes message 20 (message 10 is outside this page) + val history = + listOf( + SessionMessage(id = 20, role = "assistant", content = JsonPrimitive("Done")), + ) + val mapped = mapServerMessages("session", history, 0, true, live, isPagingOlder = false) + assertEquals( + "Exact REST id match must preserve identity regardless of older duplicates", + "new-completion", + mapped.single().completionId, + ) + } + + @Test + fun metadataFreeReplacementMustNotBeCancelled() { + postTarget() + ReplyNotificationTracker.activeNotificationProvider = { + ActiveReplyInfo(2, null, null, null, null, null, 0L, 1L) + } + assertFalse(ReplyNotificationTracker.onMessageVisible(context, "default", "session", "new-completion")) + verify(exactly = 0) { manager.cancel(any()) } + } + + @Test + fun actionPostingMustNotInterleaveBetweenInspectionAndCancellation() { + val reply = + ActiveReplyInfo( + 2, + ReplyNotificationTracker.KIND_REPLY, + "default", + "session", + "new-completion", + "Done", + 1L, + 1L, + ) + val actualSlot = AtomicReference("reply") + val ready = CountDownLatch(1) + val allowNotify = CountDownLatch(1) + val posted = CountDownLatch(1) + val action = mockk() + every { manager.notify(2, action) } answers { actualSlot.set("action") } + every { manager.cancel(2) } answers { actualSlot.set("empty") } + postTarget() + val writer = + Thread { + // Same split ownership/write sequence as showReplyNotification and NotificationReplyReceiver. + ReplyNotificationTracker.onNonReplyNotificationPosted() + ready.countDown() + check(allowNotify.await(5, TimeUnit.SECONDS)) + manager.notify(2, action) + posted.countDown() + } + writer.start() + assertTrue(ready.await(5, TimeUnit.SECONDS)) + var reads = 0 + ReplyNotificationTracker.activeNotificationProvider = { + val snapshot = reply + reads++ + if (reads == 2) { + // Pause after querying activeNotifications but before cancel(), then complete the action write. + allowNotify.countDown() + check(posted.await(5, TimeUnit.SECONDS)) + } + snapshot + } + try { + ReplyNotificationTracker.onMessageVisible(context, "default", "session", "new-completion") + } finally { + allowNotify.countDown() + writer.join(6000) + } + assertEquals("New action notification was cancelled by the old read", "action", actualSlot.get()) + } + + @Test + fun oldReplyRacingNewerActionMustNotPostOrOverwriteAction() { + val gen1 = ReplyNotificationTracker.registerPendingReply("default", "session", "comp-1", "Old Reply") + val actionNotif = mockk() + val replyNotif = mockk() + + // Action arrives and posts + assertTrue(ReplyNotificationTracker.postActionNotification(context, actionNotif)) + verify(exactly = 1) { manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, actionNotif) } + + // Stale reply 1 tries to post after action arrived + val posted = ReplyNotificationTracker.postReplyNotification(context, replyNotif, gen1) + assertFalse("Stale reply must be rejected after action notification was posted", posted) + verify(exactly = 0) { manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, replyNotif) } + assertNull("Active target must remain null for action", ReplyNotificationTracker.getActiveTarget()) + } + + @Test + fun oldReplyRacingNewerReplyMustNotOverwriteNewerReply() { + val gen1 = ReplyNotificationTracker.registerPendingReply("default", "session", "comp-1", "Old Reply") + val gen2 = ReplyNotificationTracker.registerPendingReply("default", "session", "comp-2", "New Reply") + val replyNotif1 = mockk() + val replyNotif2 = mockk() + + // Gen 2 posts successfully + assertTrue(ReplyNotificationTracker.postReplyNotification(context, replyNotif2, gen2)) + assertEquals("comp-2", ReplyNotificationTracker.getActiveTarget()?.completionId) + + // Stale Gen 1 tries to post + assertFalse(ReplyNotificationTracker.postReplyNotification(context, replyNotif1, gen1)) + assertEquals("comp-2", ReplyNotificationTracker.getActiveTarget()?.completionId) + } + + @Test + fun readCancelRacingReplyPublicationMustPreventZombieNotification() { + val gen1 = ReplyNotificationTracker.registerPendingReply("default", "session", "comp-1", "Pending Reply") + val replyNotif = mockk() + + // User views message in foreground chat before background service calls postReplyNotification + val cancelled = + ReplyNotificationTracker.onMessageVisible( + context, + "default", + "session", + "comp-1", + ) + assertTrue("Message visibility should cancel the pending reply target", cancelled) + + // Delayed background service finally tries to post reply + val posted = ReplyNotificationTracker.postReplyNotification(context, replyNotif, gen1) + assertFalse("Tombstoned/cancelled generation must not post notification ID 2", posted) + verify(exactly = 0) { manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, replyNotif) } + } + + @Test + fun mediaRepliesRetainCompletionIdAcrossRestRehydration() { + val live = + listOf( + ChatMessage( + id = "ws-media", + role = MessageRole.ASSISTANT, + content = "Here is your chart", + completionId = "comp-media-123", + ), + ) + val history = + listOf( + SessionMessage( + id = 1, + role = "assistant", + content = JsonPrimitive("Here is your chart MEDIA:/opt/hermes/chart.png"), + ), + ) + val mapped = mapServerMessages("session", history, 0, true, live, isPagingOlder = false) + assertEquals("comp-media-123", mapped.single().completionId) + } + + @Test + fun mediaOnlyReplyRetainsCompletionIdAcrossRestRehydration() { + val live = + listOf( + ChatMessage( + id = "ws-media-only", + role = MessageRole.ASSISTANT, + content = "", + completionId = "comp-media-only", + ), + ) + val history = + listOf( + SessionMessage( + id = 1, + role = "assistant", + content = JsonPrimitive("MEDIA:/opt/hermes/photo.jpg"), + ), + ) + val mapped = mapServerMessages("session", history, 0, true, live, isPagingOlder = false) + assertEquals("comp-media-only", mapped.single().completionId) + } + + @Test + fun mediaReplyWithReasoningOnlyRowBeforeItRetainsCompletionId() { + val live = + listOf( + ChatMessage( + id = "ws-media-reasoning", + role = MessageRole.ASSISTANT, + content = "The result", + completionId = "comp-reasoning-media", + ), + ) + val history = + listOf( + SessionMessage( + id = 1, + role = "assistant", + content = JsonPrimitive(""), + reasoning = JsonPrimitive("thinking step"), + ), + SessionMessage(id = 2, role = "assistant", content = JsonPrimitive("The result MEDIA:/opt/pic.png")), + ) + val mapped = mapServerMessages("session", history, 0, true, live, isPagingOlder = false) + assertEquals("comp-reasoning-media", mapped.single().completionId) + } + + @Test + fun blankMessageCompleteMustNotCorruptHistoricalAssistantMessage() { + val oldAssistant = + ChatMessage( + id = "old-assistant", + role = MessageRole.ASSISTANT, + content = "Old reply from yesterday", + completionId = "old-comp", + ) + val state = ChatUiState(currentSessionId = "session", messages = listOf(oldAssistant)) + val streaming = StreamingState() // No current streaming message or sealed orphans + + val result = + ChatWsEventReducer.reduce( + state, + streaming, + WsEvent.MessageComplete("", "session", completionId = "new-unrelated-completion"), + "session", + ) + + // Historical assistant message must be completely untouched + assertEquals( + "old-comp", + result.state.messages + .single() + .completionId, + ) + assertFalse( + "PersistMessage effect must not be emitted for unrelated historical messages", + result.effects.any { it is ReducerEffect.PersistMessage }, + ) + } + + @Test + fun coldStartRecoveredGenerationMustAdvanceCounterAndAllowFutureReplies() { + ReplyNotificationTracker.resetForTest() + // Active notification in system with generation 37 + val activeInfo = + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-37", + textSnippet = "Reply 37", + generation = 37L, + timestamp = System.currentTimeMillis(), + ) + ReplyNotificationTracker.activeNotificationProvider = { activeInfo } + + // User views message -> recovers and cancels generation 37 + val cancelled = ReplyNotificationTracker.onMessageVisible(context, "default", "session", "comp-37") + assertTrue("Generation 37 should be recovered and cancelled", cancelled) + + // New reply arrives + val newGen = ReplyNotificationTracker.registerPendingReply("default", "session", "comp-38", "Reply 38") + assertTrue("New generation must exceed recovered generation (was $newGen)", newGen > 37L) + + val replyNotif = mockk() + val posted = ReplyNotificationTracker.postReplyNotification(context, replyNotif, newGen) + assertTrue("New reply notification must be accepted and posted", posted) + verify { manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, replyNotif) } + } + + @Test + fun recoveredGenerationFollowedByActionAndReplyMustRemainStrictlyMonotonic() { + ReplyNotificationTracker.resetForTest() + val activeInfo = + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-37", + textSnippet = "Reply 37", + generation = 37L, + timestamp = System.currentTimeMillis(), + ) + ReplyNotificationTracker.activeNotificationProvider = { activeInfo } + + // Recover and cancel 37 + assertTrue(ReplyNotificationTracker.onMessageVisible(context, "default", "session", "comp-37")) + + // Post action + val actionNotif = mockk() + assertTrue(ReplyNotificationTracker.postActionNotification(context, actionNotif)) + + // Register new reply + val newGen = ReplyNotificationTracker.registerPendingReply("default", "session", "comp-39", "Reply 39") + assertTrue("New generation must strictly exceed 37 and action tombstone", newGen > 38L) + + val replyNotif = mockk() + assertTrue(ReplyNotificationTracker.postReplyNotification(context, replyNotif, newGen)) + } + + @Test + fun alreadyResolvedServerIdBindsOnlyTheExactRowAmongShortDuplicates() { + ReplyNotificationTracker.resetForTest() + val activeInfo = + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-newest-done", + serverMessageId = 2, + textSnippet = "Done", + generation = 5L, + timestamp = System.currentTimeMillis(), + ) + ReplyNotificationTracker.activeNotificationProvider = { activeInfo } + + // Two identical "Done" rows. The durable id (row 2) is what decides — + // there is no server row timestamp comparison anywhere in the mapper. + val history = + listOf( + SessionMessage( + id = 1, + role = "assistant", + content = JsonPrimitive("Done"), + ), + SessionMessage( + id = 2, + role = "assistant", + content = JsonPrimitive("Done"), + ), + ) + + val mapped = + mapServerMessages("session", history, 0, true, emptyList(), isPagingOlder = false, context = context) + assertEquals("Older message must have null completionId", null, mapped[0].completionId) + assertEquals("Newest message must receive target completionId", "comp-newest-done", mapped[1].completionId) + + // Showing ONLY the older Done must NOT cancel the notification + val cancelledOld = + ReplyNotificationTracker.onMessageVisible( + context = context, + scopeId = "default", + sessionId = "session", + completionId = mapped[0].completionId, + ) + assertFalse("Viewing older duplicate message must NOT cancel notification", cancelledOld) + verify(exactly = 0) { manager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + + // Showing the newest Done MUST cancel the notification + val cancelledNew = + ReplyNotificationTracker.onMessageVisible( + context = context, + scopeId = "default", + sessionId = "session", + completionId = mapped[1].completionId, + ) + assertTrue("Viewing newest message must cancel notification", cancelledNew) + verify(exactly = 1) { manager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + } + + @Test + fun alreadyResolvedServerIdIgnoresSharedHundredCharacterPrefix() { + ReplyNotificationTracker.resetForTest() + val textPrefix = "x".repeat(100) + val text1 = textPrefix + "A" + val text2 = textPrefix + "B" + val activeInfo = + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-long-2", + serverMessageId = 10, + textSnippet = text2.take(100), + generation = 5L, + timestamp = System.currentTimeMillis(), + ) + ReplyNotificationTracker.activeNotificationProvider = { activeInfo } + + val history = + listOf( + SessionMessage( + id = 10, + role = "assistant", + content = JsonPrimitive(text1), + ), + SessionMessage( + id = 20, + role = "assistant", + content = JsonPrimitive(text2), + ), + ) + + val mapped = + mapServerMessages("session", history, 0, true, emptyList(), isPagingOlder = false, context = context) + assertEquals("comp-long-2", mapped[0].completionId) + assertEquals(null, mapped[1].completionId) + + assertTrue( + ReplyNotificationTracker.onMessageVisible( + context, + "default", + "session", + mapped[0].completionId, + ), + ) + assertFalse( + ReplyNotificationTracker.onMessageVisible( + context, + "default", + "session", + mapped[1].completionId, + ), + ) + } + + @Test + fun alreadyResolvedServerIdBindsMediaRowAndDismissesOnVisible() { + ReplyNotificationTracker.resetForTest() + val activeInfo = + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-media-cold", + serverMessageId = 100, + textSnippet = "Here is the image MEDIA:/opt/hermes/image.png", + generation = 5L, + timestamp = System.currentTimeMillis(), + ) + ReplyNotificationTracker.activeNotificationProvider = { activeInfo } + + val history = + listOf( + SessionMessage( + id = 100, + role = "assistant", + content = JsonPrimitive("Here is the image MEDIA:/opt/hermes/image.png"), + ), + ) + + val mapped = + mapServerMessages("session", history, 0, true, emptyList(), isPagingOlder = false, context = context) + val row = mapped.single() + assertEquals("comp-media-cold", row.completionId) + assertEquals("Here is the image", row.content) + + val dismissed = + ReplyNotificationTracker.onMessageVisible( + context = context, + scopeId = "default", + sessionId = "session", + completionId = row.completionId, + ) + assertTrue("Viewing hydrated MEDIA row must dismiss notification", dismissed) + verify { manager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + } + + @Test + fun alreadyResolvedServerIdBindsMediaOnlyRowAndDismissesOnVisible() { + ReplyNotificationTracker.resetForTest() + val activeInfo = + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-media-only-cold", + serverMessageId = 100, + textSnippet = "MEDIA:/opt/hermes/image.png", + generation = 5L, + timestamp = System.currentTimeMillis(), + ) + ReplyNotificationTracker.activeNotificationProvider = { activeInfo } + + val history = + listOf( + SessionMessage( + id = 100, + role = "assistant", + content = JsonPrimitive("MEDIA:/opt/hermes/image.png"), + ), + ) + + val mapped = + mapServerMessages("session", history, 0, true, emptyList(), isPagingOlder = false, context = context) + val row = mapped.single() + assertEquals("comp-media-only-cold", row.completionId) + assertEquals("", row.content) + + val dismissed = + ReplyNotificationTracker.onMessageVisible( + context = context, + scopeId = "default", + sessionId = "session", + completionId = row.completionId, + ) + assertTrue("Viewing hydrated MEDIA-only row must dismiss notification", dismissed) + verify { manager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + } + + @Test + fun alreadyResolvedServerIdBindsOnlyTheExactMediaRowAmongDuplicates() { + ReplyNotificationTracker.resetForTest() + val activeInfo = + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-media-dup-2", + serverMessageId = 1, + textSnippet = "MEDIA:/opt/hermes/image.png", + generation = 5L, + timestamp = System.currentTimeMillis(), + ) + ReplyNotificationTracker.activeNotificationProvider = { activeInfo } + + val history = + listOf( + SessionMessage( + id = 1, + role = "assistant", + content = JsonPrimitive("MEDIA:/opt/hermes/image.png"), + ), + SessionMessage( + id = 2, + role = "assistant", + content = JsonPrimitive("MEDIA:/opt/hermes/image.png"), + ), + ) + + val mapped = + mapServerMessages("session", history, 0, true, emptyList(), isPagingOlder = false, context = context) + assertEquals("comp-media-dup-2", mapped[0].completionId) + assertEquals(null, mapped[1].completionId) + + assertTrue( + ReplyNotificationTracker.onMessageVisible( + context, + "default", + "session", + mapped[0].completionId, + ), + ) + assertFalse( + ReplyNotificationTracker.onMessageVisible( + context, + "default", + "session", + mapped[1].completionId, + ), + ) + } + + /** + * The path production actually runs, end to end: the turn boundary armed + * before the prompt names exactly one REST row above it, that row id travels + * in the notification, only that row is hydrated with the completion id, and + * only that row dismisses the notification. + */ + @Test + fun mobileTurnBoundaryResolvesTheExactRowThenOnlyThatRowDismisses() { + ReplyNotificationTracker.resetForTest() + TurnCorrelationTracker.resetForTest() + TurnCorrelationTracker.armBoundary("default", "session", 50) + + val history = + listOf( + SessionMessage(id = 10, role = "assistant", content = JsonPrimitive("Done")), + SessionMessage(id = 51, role = "user", content = JsonPrimitive("next")), + SessionMessage(id = 52, role = "assistant", content = JsonPrimitive("Done")), + ) + val resolved = + runBlocking { + correlateCompletedTurnRow( + scopeId = "default", + sessionId = "session", + completionText = "Done", + resolver = TurnRowResolver({ history }, listOf(0L)), + ) + } + assertEquals("The historical duplicate below the boundary must not win", 52, resolved) + + // The service stores that id in the notification extras; after process + // death the extras are all the tracker can recover from. + ReplyNotificationTracker.activeNotificationProvider = { + ActiveReplyInfo( + id = ChatNotificationService.PENDING_NOTIFICATION_ID, + kind = ReplyNotificationTracker.KIND_REPLY, + scopeId = "default", + sessionId = "session", + completionId = "comp-turn", + textSnippet = "Done", + generation = 5L, + timestamp = 0L, + serverMessageId = resolved, + ) + } + + val mapped = + mapServerMessages("session", history, 0, true, emptyList(), isPagingOlder = false, context = context) + assertEquals(listOf(null, null, "comp-turn"), mapped.map { it.completionId }) + + assertFalse( + "Viewing the older duplicate must not dismiss", + ReplyNotificationTracker.onMessageVisible(context, "default", "session", mapped[0].completionId), + ) + assertTrue( + "Viewing the turn's own row must dismiss", + ReplyNotificationTracker.onMessageVisible(context, "default", "session", mapped[2].completionId), + ) + verify(exactly = 1) { manager.cancel(ChatNotificationService.PENDING_NOTIFICATION_ID) } + } + + @Test + fun unknownScopeMustNotAcknowledgeScopedReply() { + val target = ReplyNotificationTarget("other-profile", "session", "completion", 1L, "Done") + assertFalse(target.matches(null, "session", "completion")) + } +} diff --git a/app/src/test/java/com/m57/hermescontrol/ui/chat/components/ChatReadObserverTest.kt b/app/src/test/java/com/m57/hermescontrol/ui/chat/components/ChatReadObserverTest.kt new file mode 100644 index 00000000..58b8ef72 --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/ui/chat/components/ChatReadObserverTest.kt @@ -0,0 +1,198 @@ +package com.m57.hermescontrol.ui.chat.components + +import androidx.compose.foundation.lazy.LazyListItemInfo +import androidx.compose.foundation.lazy.LazyListLayoutInfo +import com.m57.hermescontrol.ui.chat.ChatMessage +import com.m57.hermescontrol.ui.chat.MessageRole +import io.mockk.every +import io.mockk.mockk +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test + +class ChatReadObserverTest { + private fun createMockItem( + key: Any, + offset: Int, + size: Int, + ): LazyListItemInfo { + val item = mockk() + every { item.key } returns key + every { item.offset } returns offset + every { item.size } returns size + return item + } + + private fun createMockLayoutInfo( + items: List, + startOffset: Int = 0, + endOffset: Int = 1000, + ): LazyListLayoutInfo { + val layout = mockk() + every { layout.visibleItemsInfo } returns items + every { layout.viewportStartOffset } returns startOffset + every { layout.viewportEndOffset } returns endOffset + return layout + } + + @Test + fun `returns empty when visible items or messages are empty`() { + val layout = createMockLayoutInfo(emptyList()) + val msg = + ChatMessage( + id = "msg-1", + role = MessageRole.ASSISTANT, + content = "Hello", + ) + + assertTrue(ChatReadObserver.findVisibleAssistantMessages(layout, listOf(msg)).isEmpty()) + + val layoutWithItem = createMockLayoutInfo(listOf(createMockItem("prose-msg-1", 100, 50))) + assertTrue(ChatReadObserver.findVisibleAssistantMessages(layoutWithItem, emptyList()).isEmpty()) + } + + @Test + fun `identifies visible completed assistant messages`() { + val msg1 = + ChatMessage( + id = "msg-1", + role = MessageRole.ASSISTANT, + content = "Answer 1", + completionId = "comp-1", + ) + val msg2 = + ChatMessage( + id = "msg-2", + role = MessageRole.ASSISTANT, + content = "Answer 2", + completionId = "comp-2", + ) + + val items = + listOf( + createMockItem("prose-msg-1", 100, 200), + createMockItem("prose-msg-2", 400, 250), + ) + val layout = createMockLayoutInfo(items) + + val visible = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(msg1, msg2)) + assertEquals(2, visible.size) + assertEquals("msg-1", visible[0].id) + assertEquals("msg-2", visible[1].id) + } + + @Test + fun `ignores non-prose items like user bubbles, reasoning cards, and tool rows`() { + val assistant = + ChatMessage( + id = "ast-1", + role = MessageRole.ASSISTANT, + content = "Done", + ) + val user = + ChatMessage( + id = "usr-1", + role = MessageRole.USER, + content = "Question", + ) + + val items = + listOf( + createMockItem("user-usr-1", 50, 100), + createMockItem("reasoning-ast-1", 150, 100), + createMockItem("tool-tool-1", 250, 100), + createMockItem("prose-ast-1", 350, 100), + ) + val layout = createMockLayoutInfo(items) + + val visible = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(assistant, user)) + assertEquals(1, visible.size) + assertEquals("ast-1", visible.single().id) + } + + @Test + fun `ignores streaming assistant messages`() { + val streaming = + ChatMessage( + id = "stream-1", + role = MessageRole.ASSISTANT, + content = "typing...", + isStreaming = true, + ) + + val items = listOf(createMockItem("prose-stream-1", 100, 200)) + val layout = createMockLayoutInfo(items) + + val visible = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(streaming)) + assertTrue(visible.isEmpty()) + } + + @Test + fun `filters out items outside viewport bounds`() { + val msgAbove = + ChatMessage( + id = "above", + role = MessageRole.ASSISTANT, + content = "Above viewport", + ) + val msgInside = + ChatMessage( + id = "inside", + role = MessageRole.ASSISTANT, + content = "Inside viewport", + ) + val msgBelow = + ChatMessage( + id = "below", + role = MessageRole.ASSISTANT, + content = "Below viewport", + ) + + // Viewport is 200..800 + val items = + listOf( + // Fully above: offset -100, size 100 -> offset + size = 0 <= 200 + createMockItem("prose-above", -100, 100), + // Inside: offset 300, size 150 + createMockItem("prose-inside", 300, 150), + // Fully below: offset 900, size 100 >= 800 + createMockItem("prose-below", 900, 100), + ) + val layout = createMockLayoutInfo(items, startOffset = 200, endOffset = 800) + + val visible = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(msgAbove, msgInside, msgBelow)) + assertEquals(1, visible.size) + assertEquals("inside", visible.single().id) + } + + @Test + fun `includes partially visible messages at viewport edges`() { + val topPartial = + ChatMessage( + id = "top", + role = MessageRole.ASSISTANT, + content = "Top edge", + ) + val bottomPartial = + ChatMessage( + id = "bottom", + role = MessageRole.ASSISTANT, + content = "Bottom edge", + ) + + // Viewport 100..500 + val items = + listOf( + // Crosses top boundary: offset 50, size 100 (bottom is 150 > 100) + createMockItem("prose-top", 50, 100), + // Crosses bottom boundary: offset 450, size 100 (top is 450 < 500) + createMockItem("prose-bottom", 450, 100), + ) + val layout = createMockLayoutInfo(items, startOffset = 100, endOffset = 500) + + val visible = ChatReadObserver.findVisibleAssistantMessages(layout, listOf(topPartial, bottomPartial)) + assertEquals(2, visible.size) + assertEquals("top", visible[0].id) + assertEquals("bottom", visible[1].id) + } +}