feat: auto-dismiss reply notifications when read in chat (#1195)

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,

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

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
+    }
+}

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<TurnBoundary>
+
+    fun write(boundaries: List<TurnBoundary>)
+}
+
+/**
+ * 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<TurnBoundary> =
+        prefs.all.mapNotNull { (key, value) -> decodeBoundary(key, value as? String) }
+
+    override fun write(boundaries: List<TurnBoundary>) {
+        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<TurnBoundary> = 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<TurnBoundary>,
+        now: Long = System.currentTimeMillis(),
+    ): List<TurnBoundary> =
+        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<TurnBoundary>) {
+        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<SessionMessage>? {
+    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) })

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<SessionMessage>?,
+    /** Attempt delays; the first attempt runs immediately. */
+    private val retryDelaysMs: List<Long> = 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<SessionMessage>? =
+        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<String>,
+) {
+    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 },
+    )
+}