The gateway replays the full transcript over a new socket on session.resume unless omit_messages is set; mobile hydrates history via REST in parallel, so the replay is redundant. When a reconnect lands mid-stream, the replayed message gets a fresh UUID and dedup can't collapse the 90% orphan with the 100% replay (different ids + content), leaving a duplicate bubble. Desktop already sends omit_messages: true; mirror that in both resume paths (ChatViewModel + NotificationReplyReceiver) and pin it in the exact-map test verifies. Co-authored-by: M57 <hy4ri@users.noreply.github.com>
141 lines
6.2 KiB
Kotlin
141 lines
6.2 KiB
Kotlin
package com.m57.hermescontrol.notification
|
|
|
|
import android.app.Notification
|
|
import android.app.NotificationManager
|
|
import android.content.BroadcastReceiver
|
|
import android.content.Context
|
|
import android.content.Intent
|
|
import androidx.core.app.NotificationCompat
|
|
import androidx.core.app.RemoteInput
|
|
import com.m57.hermescontrol.R
|
|
import com.m57.hermescontrol.data.session.ActiveSessionHolder
|
|
import com.m57.hermescontrol.data.ws.HermesWsClient
|
|
import com.m57.hermescontrol.data.ws.WsMethods
|
|
import kotlinx.coroutines.CoroutineScope
|
|
import kotlinx.coroutines.Dispatchers
|
|
import kotlinx.coroutines.SupervisorJob
|
|
import kotlinx.coroutines.launch
|
|
import kotlinx.coroutines.withContext
|
|
import kotlinx.coroutines.withTimeout
|
|
|
|
open class NotificationReplyReceiver : BroadcastReceiver() {
|
|
companion object {
|
|
const val KEY_TEXT_REPLY = "key_text_reply"
|
|
const val EXTRA_SESSION_ID = "extra_session_id"
|
|
private const val REPLY_TIMEOUT_MS = 5_000L
|
|
}
|
|
|
|
// Reusable scope for async reply processing — avoids creating a new
|
|
// unmanaged CoroutineScope per broadcast fire. (PERF-15)
|
|
private val replyScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
|
|
|
|
/**
|
|
* Test-friendly wrapper for [BroadcastReceiver.goAsync] which is `final`
|
|
* (Java) and cannot be mocked or overridden directly. Tests override this
|
|
* via anonymous subclass to inject a fake [PendingResult].
|
|
*/
|
|
internal open fun goAsyncCompat(): BroadcastReceiver.PendingResult = goAsync()
|
|
|
|
/**
|
|
* Test-friendly wrapper for notification creation. Override in tests to
|
|
* avoid [NotificationCompat.Builder.build()] calling Android framework
|
|
* methods that throw "not mocked" in unit tests.
|
|
*/
|
|
internal open fun buildReplyNotification(context: Context): Notification =
|
|
NotificationCompat
|
|
.Builder(context, ChatNotificationService.CHAT_CHANNEL_ID)
|
|
.setSmallIcon(R.drawable.ic_notification)
|
|
.setContentTitle("Hermes")
|
|
.setContentText("Replied")
|
|
.setPriority(NotificationCompat.PRIORITY_HIGH)
|
|
.setAutoCancel(true)
|
|
.build()
|
|
|
|
override fun onReceive(
|
|
context: Context,
|
|
intent: Intent,
|
|
) {
|
|
val remoteInput = RemoteInput.getResultsFromIntent(intent)
|
|
val replyText = remoteInput?.getCharSequence(KEY_TEXT_REPLY)?.toString()
|
|
val sessionId = intent.getStringExtra(EXTRA_SESSION_ID)
|
|
|
|
if (!replyText.isNullOrBlank() && !sessionId.isNullOrBlank()) {
|
|
val pendingResult = goAsyncCompat()
|
|
replyScope.launch {
|
|
try {
|
|
withTimeout(5000L) {
|
|
withContext(Dispatchers.IO) {
|
|
val db =
|
|
com.m57.hermescontrol.data.local.HermesDatabase
|
|
.get(context)
|
|
val dao = db.chatMessageDao()
|
|
if (!dao.sessionExists(sessionId)) {
|
|
android.util.Log.w(
|
|
"NotificationReply",
|
|
"Ignoring reply for unknown session: $sessionId",
|
|
)
|
|
return@withContext
|
|
}
|
|
|
|
val runtimeSessionId =
|
|
ActiveSessionHolder.resolveRuntimeSessionId(sessionId)
|
|
?: resumeSession(sessionId)
|
|
HermesWsClient
|
|
.request(
|
|
WsMethods.PROMPT_SUBMIT,
|
|
mapOf("session_id" to runtimeSessionId, "text" to replyText),
|
|
timeoutMs = REPLY_TIMEOUT_MS,
|
|
).await()
|
|
|
|
val entity =
|
|
com.m57.hermescontrol.data.local.ChatMessageEntity(
|
|
id =
|
|
java.util.UUID
|
|
.randomUUID()
|
|
.toString(),
|
|
sessionId = sessionId,
|
|
role = "USER",
|
|
content = replyText,
|
|
timestamp = System.currentTimeMillis(),
|
|
)
|
|
dao.upsert(entity)
|
|
|
|
val repliedNotification = buildReplyNotification(context)
|
|
val manager = context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager
|
|
manager.notify(ChatNotificationService.PENDING_NOTIFICATION_ID, repliedNotification)
|
|
|
|
// The follow-up turn is now pending — keep the
|
|
// foreground service alive if it retired after the
|
|
// previous reply completed, so the next
|
|
// MessageComplete still notifies while backgrounded
|
|
// (issue #794). No-op when the app is foreground.
|
|
if (!NotificationHelper.isAppInForeground()) {
|
|
NotificationHelper.start(context)
|
|
}
|
|
}
|
|
}
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("NotificationReply", "Failed to process reply", e)
|
|
} finally {
|
|
pendingResult.finish()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private suspend fun resumeSession(storedSessionId: String): String {
|
|
val result =
|
|
HermesWsClient
|
|
.request(
|
|
WsMethods.SESSION_RESUME,
|
|
mapOf("session_id" to storedSessionId, "omit_messages" to true),
|
|
timeoutMs = REPLY_TIMEOUT_MS,
|
|
).await() as? Map<*, *>
|
|
val runtimeSessionId =
|
|
(result?.get("session_id") as? String)?.takeIf { it.isNotBlank() }
|
|
?: error("Resume returned no runtime session id")
|
|
ActiveSessionHolder.set(runtimeSessionId, storedSessionId)
|
|
return runtimeSessionId
|
|
}
|
|
}
|