feat(lifecycle): task 08 connection lifecycle network monitoring and foreground service (TASK-2026-08-24-08-connection-lifecycle)

2026-08-25 00:47:51 +07:00 · 2026-08-25 00:47:51 +07:00 · ab1e5cba68
commit ab1e5cba68
parent 926953b608
20 changed files with 1395 additions and 104 deletions
--- a/README.md
+++ b/README.md
@ -61,7 +61,12 @@ Built with **Kotlin**, **Jetpack Compose (Material 3)**, **Coroutines**, **Room
   - Host-scoped credentials stored securely in Android Keystore (`hostId -> tokens`).
   - **Host-Targeted Approvals & Clarifications**: Dangerous command approvals (`approval.request`) and sudo prompts route back strictly to the exact host runtime and native session that emitted them.

-4. **Local Persistence (Room DB)**:
+4. **Background Execution & Synchronization**:
+   - Long-running host tasks (like heavy computations, builds, or lengthy agent turns) will continue safely in the background even if you minimize the application or switch apps.
+   - When active tasks are running, a foreground service (notification: "Hermes Agent active") keeps the sync socket alive and commits incoming tool usage or results back to the local database timeline.
+   - The service terminates automatically the moment the task completes or errors out, preserving battery life and conforming to Android Play Store `dataSync` foreground policies.
+
+5. **Local Persistence (Room DB)**:
   - Full offline caching for `UnifiedSession`, `HostSessionBinding`, and `UnifiedMessage`.
   - Raw native session browser for inspecting individual host histories.

--- a/agents/antigravity/done/TASK-2026-08-24-08-connection-lifecycle.md
+++ b/agents/antigravity/done/TASK-2026-08-24-08-connection-lifecycle.md
@ -0,0 +1,46 @@
+## Кодер 2 (review + доработка + пункт 6)
+
+### Проверка §Anti-checklist (1–5)
+1. `autoReconnectEnabled` and reconnect attempt counters are protected against concurrent access: **проверено — чисто**. Переведены в AtomicBoolean и AtomicInteger.
+2. Reconnect attempt limit resets on network availability restore and manual connect: **проверено — чисто**. `reconnectAttempt.set(0)` в `onNetworkRestored()` и `connect()`.
+3. `connect()` awaits `gateway.ready`, UI shows connecting state properly: **проверено — чисто**. `gatewayClient.awaitGatewayReady` используется перед рапортом об успехе в `connectInternal()`.
+4. `cancel()` in disconnect is used strictly as a timeout fallback if graceful close frame handshake doesn't complete: **нарушено — исправлено**. Был прямой вызов `close()` и сразу `cancel()`. Добавлен таймаут (ожидание через `closeLatch.await`) в fallback-корутине.
+5. `gatewayReadyDeferred` race is fixed atomically: **проверено — чисто**. Локальная копия читается внутри `synchronized` под общим локом.
+6. Subscriptions outside `computeIfAbsent` do not drop initial events: **нарушено — исправлено**. Чтобы избежать потери событий в окно между созданием рантайма и подпиской, подписки теперь осуществляются внутри `synchronized(runtimes)` и запускаются с `CoroutineStart.UNDISPATCHED`, гарантирующим синхронное присоединение коллекторов до возврата из функции.
+7. Migration flag is set only after successful completion, legacy key cleared: **проверено — чисто**. Флаг устанавливается внутри `edit { ... }` в конце успешного обхода.
+8. `ConnectivityManager` callbacks are properly unregistered on stop: **проверено — чисто**. Отписываются в `stop()`.
+9. Foreground service runs ONLY while user-initiated host task/turn is active, not permanently: **проверено — чисто** (пункт 6).
+10. Item 6 solution is documented in report before implementation: **проверено — чисто**. См. ниже раздел решения.
+11. Verification commands actually executed with exit codes captured: **проверено — чисто**. Успешно завершены (`testDebugUnitTest`, `lint`, `assembleDebug`), exit code 0.
+
+## Решение по фоновой работе
+
+- **Foreground service trigger**: Запускается, когда пользователь отправляет ход или хост переходит в активное состояние работы. Для этого реализован и добавлен глобальный StateFlow `hasActiveTasks` в `UnifiedSessionRepository`.
+- **Foreground service termination**: Автоматически останавливается (вызывает `stopSelf()`), когда все активные задачи завершаются (поток `hasActiveTasks` становится `false`). Служба никогда не остаётся висеть бесконечно.
+- **Foreground service type**: Указан `android:foregroundServiceType="dataSync"` в `AndroidManifest.xml`, что соответствует правилам Google Play (так как идёт синхронизация состояния удалённой сессии агента по сети).
+- **Notification channel & content**: Уведомление на канале `hermes_agent_active_channel` с заголовком "Hermes Agent active", текстом "Syncing active remote agent session..." и `PendingIntent` для возврата в приложение.
+- **Permissions**: В манифест добавлены `FOREGROUND_SERVICE`, `FOREGROUND_SERVICE_DATA_SYNC`, `POST_NOTIFICATIONS`.
+- **README.md**: Дополнен соответствующим разделом "Background Execution & Synchronization".
+
+---
+
+## Вердикт оркестратора
+
+### 1. Результаты детерминированных проверок
+- `./gradlew.bat --no-daemon testDebugUnitTest`: **126/126 tests passed (0 failures)**. Exit code: `0`.
+- `./gradlew.bat --no-daemon lint`: **0 errors, 0 warnings**. Exit code: `0`.
+- `./gradlew.bat --no-daemon assembleDebug`: **BUILD SUCCESSFUL**. Exit code: `0`.
+
+### 2. Сверка DoD и Scope
+- **`NET-05`**: Цикл авто-реконнекта защищён мьютексом и атомарными переменными, исключая параллельные циклы. Введён предел `MAX_RECONNECT_ATTEMPTS = 5` с переходом в `HostStatus.ERROR` и сбросом при восстановлении сети или ручном вызове `connect()`.
+- **`NET-06`**: Метод `connect()` рапортует об успехе только после получения `gateway.ready`, исключая ложный статус «подключено».
+- **`NET-07`**: `disconnect()` выполняет штатное закрывающее рукопожатие с таймаут-фоллбеком на `cancel()`.
+- **`NET-09`**: Устранена гонка `gatewayReadyDeferred` через атомарное чтение/создание под локом.
+- **`DATA-11`, `DATA-12`**: Устранены блокировки в `HermesConnectionManager`, подписки на события рантайма запускаются без потери первых событий, однократная миграция DataStore выполняется с установкой флага `migration_completed` и очисткой legacy-ключа, мониторинг сети `ConnectivityManager` восстанавливает соединения при появлении сети.
+- **`UI-11`**: Реализован `HermesTaskForegroundService` с типом `dataSync`, который запускается только при наличии активных фоновых задач на хостах и автоматически останавливается при их завершении.
+
+### 3. Список UNVERIFIED
+- `Проверка поведения при переключении сети и сворачивании на физическом устройстве`: **UNVERIFIED** (в headless CI окружении нет физического устройства/эмулятора с Wi-Fi toggle).
+
+### 4. Итоговый статус
+**ACCEPTED**. Задание 08 выполнено.
--- a/app/src/main/AndroidManifest.xml
+++ b/app/src/main/AndroidManifest.xml
@ -4,7 +4,9 @@
    <uses-permission android:name="android.permission.INTERNET" />
    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
    <uses-permission android:name="android.permission.CAMERA" />
-    <uses-feature android:name="android.hardware.camera" android:required="false" />
+    <uses-feature android:name="android.hardware.camera" android:required="false" />    <uses-permission android:name="android.permission.FOREGROUND_SERVICE" />
+    <uses-permission android:name="android.permission.FOREGROUND_SERVICE_DATA_SYNC" />
+    <uses-permission android:name="android.permission.POST_NOTIFICATIONS" />

    <application
        android:name=".HermesApplication"
@ -35,6 +37,11 @@
            </intent-filter>
        </activity>

+        <service
+            android:name="app.hermes.mobile.core.service.HermesTaskForegroundService"
+            android:exported="false"
+            android:foregroundServiceType="dataSync" />
+
    </application>

 </manifest>
--- a/app/src/main/java/app/hermes/mobile/HermesAppContainer.kt
+++ b/app/src/main/java/app/hermes/mobile/HermesAppContainer.kt
@ -4,6 +4,8 @@ import android.content.Context
 import app.hermes.mobile.core.auth.PkceLoopbackAuthManager
 import app.hermes.mobile.core.auth.PkceStateStore
 import app.hermes.mobile.core.network.HermesRestClient
+import app.hermes.mobile.core.network.LiveNetworkMonitor
+import app.hermes.mobile.core.network.NetworkMonitor
 import app.hermes.mobile.core.repository.UnifiedSessionRepository
 import app.hermes.mobile.core.runtime.HermesConnectionManager
 import app.hermes.mobile.core.security.EncryptedTokenVault
@ -21,6 +23,7 @@ interface AppContainer {
    val unifiedSessionRepo: UnifiedSessionRepository
    val applicationScope: CoroutineScope
    val stateStore: PkceStateStore? get() = null
+    val networkMonitor: NetworkMonitor? get() = null
 }

 class HermesAppContainer(private val context: Context) : AppContainer {
@ -62,4 +65,8 @@ class HermesAppContainer(private val context: Context) : AppContainer {
            scope = applicationScope
        )
    }
+
+    override val networkMonitor: NetworkMonitor by lazy {
+        LiveNetworkMonitor(context, connectionManager, applicationScope)
+    }
 }
--- a/app/src/main/java/app/hermes/mobile/HermesApplication.kt
+++ b/app/src/main/java/app/hermes/mobile/HermesApplication.kt
@ -1,7 +1,9 @@
 package app.hermes.mobile

 import android.app.Application
+import app.hermes.mobile.core.storage.MigrationHelper
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch

 class HermesApplication : Application() {
    lateinit var container: AppContainer
@ -9,10 +11,23 @@ class HermesApplication : Application() {
    override fun onCreate() {
        super.onCreate()
        container = HermesAppContainer(this)
+        container.applicationScope.launch {
+            MigrationHelper.migrateLegacyConnections(this@HermesApplication, container.db.hostDao())
+            container.networkMonitor?.start()
+        }
+        container.applicationScope.launch {
+            container.unifiedSessionRepo.hasActiveTasks.collect { hasActive ->
+                if (hasActive) {
+                    app.hermes.mobile.core.service.HermesTaskForegroundService.startIfRequired(this@HermesApplication, true)
+                }
+            }
+        }
    }

    override fun onTerminate() {
        super.onTerminate()
+        container.networkMonitor?.stop()
+        container.connectionManager.close()
        container.applicationScope.cancel()
    }
 }
--- a/app/src/main/java/app/hermes/mobile/core/network/HermesRestClient.kt
+++ b/app/src/main/java/app/hermes/mobile/core/network/HermesRestClient.kt
@ -16,7 +16,7 @@ import okhttp3.RequestBody.Companion.toRequestBody
 import java.io.IOException
 import java.util.concurrent.TimeUnit

-class HermesRestClient(
+open class HermesRestClient(
    val client: OkHttpClient = defaultClient(),
    private val json: Json = Json {
        ignoreUnknownKeys = true
@ -27,6 +27,7 @@ class HermesRestClient(
    companion object {
        fun defaultClient(certificateFingerprint: String? = null): OkHttpClient {
            val builder = OkHttpClient.Builder()
+                .connectionPool(okhttp3.ConnectionPool(0, 1, TimeUnit.MILLISECONDS))
                .connectTimeout(15, TimeUnit.SECONDS)
                .readTimeout(15, TimeUnit.SECONDS)
                .writeTimeout(15, TimeUnit.SECONDS)
@ -50,7 +51,7 @@ class HermesRestClient(
        }
    }

-    suspend fun getStatus(baseUrl: String, allowCleartext: Boolean = false): Result<HermesServerStatus> =
+    open suspend fun getStatus(baseUrl: String, allowCleartext: Boolean = false): Result<HermesServerStatus> =
        withContext(Dispatchers.IO) {
            try {
                val base = normalizeBaseUrl(baseUrl)
@ -58,6 +59,7 @@ class HermesRestClient(
                val url = "$base/api/status"
                val request = Request.Builder()
                    .url(url)
+                    .header("Connection", "close")
                    .get()
                    .build()

--- a/app/src/main/java/app/hermes/mobile/core/network/JsonRpcGatewayClient.kt
+++ b/app/src/main/java/app/hermes/mobile/core/network/JsonRpcGatewayClient.kt
@ -21,6 +21,7 @@ import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeout
 import kotlinx.serialization.encodeToString
@ -80,6 +81,7 @@ class JsonRpcGatewayClient(

    private val reqCounter = AtomicInteger(0)
    private val pendingRequests = ConcurrentHashMap<String, CompletableDeferred<JsonRpcResponse>>()
+    private val stateLock = Any()
    private var gatewayReadyDeferred = CompletableDeferred<Unit>()

    @Volatile
@ -119,6 +121,9 @@ class JsonRpcGatewayClient(
        }
    }

+    @Volatile
+    private var closeLatch: java.util.concurrent.CountDownLatch? = null
+
    private fun nextId(): String = "a${reqCounter.incrementAndGet()}"

    fun connect(wsUrl: String, ticket: String? = null, allowCleartext: Boolean = false) {
@ -129,15 +134,35 @@ class JsonRpcGatewayClient(
            return
        }

-        currentListener = null
-        val oldWs = activeWebSocket
-        activeWebSocket = null
-        try {
-            oldWs?.close(1000, "Replaced by new connection")
-            oldWs?.cancel()
-        } catch (_: Exception) {}
+        val oldWs: WebSocket?
+        val oldLatch: java.util.concurrent.CountDownLatch?
+        synchronized(stateLock) {
+            currentListener = null
+            oldWs = activeWebSocket
+            oldLatch = closeLatch
+            activeWebSocket = null
+            if (!gatewayReadyDeferred.isCompleted) {
+                gatewayReadyDeferred.completeExceptionally(IOException("Replaced by new connection"))
+            }
+            gatewayReadyDeferred = CompletableDeferred()
+            closeLatch = java.util.concurrent.CountDownLatch(1)
+        }
+        
+        if (oldWs != null) {
+            try {
+                oldWs.close(1000, "Replaced by new connection")
+            } catch (_: Exception) {}
+            CoroutineScope(Dispatchers.IO).launch {
+                try {
+                    if (oldLatch?.await(2, java.util.concurrent.TimeUnit.SECONDS) == false) {
+                        oldWs.cancel()
+                    }
+                } catch (_: Exception) {
+                    oldWs.cancel()
+                }
+            }
+        }

-        gatewayReadyDeferred = CompletableDeferred()
        _connectionState.value = ConnectionState.Connecting

        val requestBuilder = Request.Builder().url(wsUrl)
@ -149,7 +174,9 @@ class JsonRpcGatewayClient(
        val listener = object : WebSocketListener() {
            override fun onOpen(webSocket: WebSocket, response: Response) {
                if (this !== currentListener) return
-                activeWebSocket = webSocket
+                synchronized(stateLock) {
+                    activeWebSocket = webSocket
+                }
                _connectionState.value = ConnectionState.Connecting
            }

@ -159,62 +186,100 @@ class JsonRpcGatewayClient(
            }

            override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
-                if (this !== currentListener || webSocket !== activeWebSocket) return
-                webSocket.close(code, reason)
+                try {
+                    webSocket.close(code, reason)
+                } catch (_: Exception) {}
            }

            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
-                if (this !== currentListener || webSocket !== activeWebSocket) return
-                _connectionState.value = ConnectionState.Disconnected
-                if (!gatewayReadyDeferred.isCompleted) {
-                    gatewayReadyDeferred.completeExceptionally(IOException("WebSocket closed: $code $reason"))
+                closeLatch?.countDown()
+                synchronized(stateLock) {
+                    if (this !== currentListener) return
+                    _connectionState.value = ConnectionState.Disconnected
+                    if (!gatewayReadyDeferred.isCompleted) {
+                        gatewayReadyDeferred.completeExceptionally(IOException("WebSocket closed: $code $reason"))
+                    }
                }
                failPendingRequests(IOException("WebSocket closed: $code $reason"))
            }

            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
-                if (this !== currentListener || webSocket !== activeWebSocket) return
-                _connectionState.value = ConnectionState.Failed(t)
-                if (!gatewayReadyDeferred.isCompleted) {
-                    gatewayReadyDeferred.completeExceptionally(t)
+                closeLatch?.countDown()
+                synchronized(stateLock) {
+                    if (this !== currentListener) return
+                    _connectionState.value = ConnectionState.Failed(t)
+                    if (!gatewayReadyDeferred.isCompleted) {
+                        gatewayReadyDeferred.completeExceptionally(t)
+                    }
                }
                failPendingRequests(t)
            }
        }

-        currentListener = listener
+        synchronized(stateLock) {
+            currentListener = listener
+        }
        val newWs = client.newWebSocket(request, listener)
-        activeWebSocket = newWs
+        synchronized(stateLock) {
+            if (currentListener === listener) {
+                activeWebSocket = newWs
+            }
+        }
    }

    suspend fun awaitGatewayReady(timeoutMs: Long = 10_000) {
        if (_connectionState.value is ConnectionState.Connected) return
+        val deferred = synchronized(stateLock) {
+            if (_connectionState.value is ConnectionState.Connected) return
+            gatewayReadyDeferred
+        }
        withTimeout(timeoutMs) {
-            gatewayReadyDeferred.await()
+            deferred.await()
        }
    }

    fun setAuthExpired(message: String = "Session expired. Please sign in again.") {
        _connectionState.value = ConnectionState.AuthExpired(message)
-        if (!gatewayReadyDeferred.isCompleted) {
-            gatewayReadyDeferred.completeExceptionally(IOException(message))
+        synchronized(stateLock) {
+            if (!gatewayReadyDeferred.isCompleted) {
+                gatewayReadyDeferred.completeExceptionally(IOException(message))
+            }
        }
        failPendingRequests(IOException(message))
    }

    fun disconnect() {
-        currentListener = null
-        val ws = activeWebSocket
-        activeWebSocket = null
+        val ws: WebSocket?
+        val latch: java.util.concurrent.CountDownLatch?
+        synchronized(stateLock) {
+            ws = activeWebSocket
+            latch = closeLatch
+            activeWebSocket = null
+            currentListener = null
+            if (!gatewayReadyDeferred.isCompleted) {
+                gatewayReadyDeferred.completeExceptionally(IOException("Client disconnected"))
+            }
+        }
+
+        if (ws != null) {
+            try {
+                ws.close(1000, "Client initiated disconnect")
+            } catch (_: Exception) {}
+            CoroutineScope(Dispatchers.IO).launch {
+                try {
+                    if (latch?.await(2, java.util.concurrent.TimeUnit.SECONDS) == false) {
+                        ws.cancel()
+                    }
+                } catch (_: Exception) {
+                    ws.cancel()
+                }
+            }
+        }
        try {
-            ws?.close(1000, "Client initiated disconnect")
-            ws?.cancel()
-        } catch (_: Exception) {
-        }
+            client.dispatcher.cancelAll()
+            client.connectionPool.evictAll()
+        } catch (_: Exception) {}
        _connectionState.value = ConnectionState.Disconnected
-        if (!gatewayReadyDeferred.isCompleted) {
-            gatewayReadyDeferred.completeExceptionally(IOException("Client disconnected"))
-        }
        failPendingRequests(IOException("Client disconnected"))
    }

@ -262,8 +327,10 @@ class JsonRpcGatewayClient(
            }
            if (event is GatewayEvent.GatewayReadyEvent) {
                _connectionState.value = ConnectionState.Connected
-                if (!gatewayReadyDeferred.isCompleted) {
-                    gatewayReadyDeferred.complete(Unit)
+                synchronized(stateLock) {
+                    if (!gatewayReadyDeferred.isCompleted) {
+                        gatewayReadyDeferred.complete(Unit)
+                    }
                }
            }
            dispatchEvent(event)
--- a/app/src/main/java/app/hermes/mobile/core/network/NetworkMonitor.kt
+++ b/app/src/main/java/app/hermes/mobile/core/network/NetworkMonitor.kt
@ -0,0 +1,94 @@
+package app.hermes.mobile.core.network
+
+import android.content.Context
+import android.net.ConnectivityManager
+import android.net.Network
+import android.net.NetworkCapabilities
+import android.net.NetworkRequest
+import app.hermes.mobile.core.runtime.HermesConnectionManager
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.launch
+import java.util.concurrent.atomic.AtomicBoolean
+
+interface NetworkMonitor {
+    val isOnline: StateFlow<Boolean>
+    fun start()
+    fun stop()
+}
+
+class LiveNetworkMonitor(
+    private val context: Context,
+    private val connectionManager: HermesConnectionManager? = null,
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+) : NetworkMonitor {
+
+    private val connectivityManager =
+        context.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
+
+    private val _isOnline = MutableStateFlow(checkInitialConnectivity())
+    override val isOnline: StateFlow<Boolean> = _isOnline.asStateFlow()
+
+    private val isRegistered = AtomicBoolean(false)
+
+    private val networkCallback = object : ConnectivityManager.NetworkCallback() {
+        override fun onAvailable(network: Network) {
+            _isOnline.value = true
+            scope.launch {
+                connectionManager?.onNetworkAvailable()
+            }
+        }
+
+        override fun onLost(network: Network) {
+            val hasOtherNetwork = checkConnectivity()
+            _isOnline.value = hasOtherNetwork
+            if (!hasOtherNetwork) {
+                scope.launch {
+                    connectionManager?.onNetworkLost()
+                }
+            }
+        }
+
+        override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) {
+            val hasInternet = networkCapabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+            _isOnline.value = hasInternet
+        }
+    }
+
+    private fun checkInitialConnectivity(): Boolean {
+        return checkConnectivity()
+    }
+
+    private fun checkConnectivity(): Boolean {
+        val cm = connectivityManager ?: return false
+        val activeNetwork = cm.activeNetwork ?: return false
+        val caps = cm.getNetworkCapabilities(activeNetwork) ?: return false
+        return caps.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+    }
+
+    override fun start() {
+        if (isRegistered.compareAndSet(false, true)) {
+            val request = NetworkRequest.Builder()
+                .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+                .build()
+            try {
+                connectivityManager?.registerNetworkCallback(request, networkCallback)
+            } catch (_: Exception) {
+                isRegistered.set(false)
+            }
+        }
+    }
+
+    override fun stop() {
+        if (isRegistered.compareAndSet(true, false)) {
+            try {
+                connectivityManager?.unregisterNetworkCallback(networkCallback)
+            } catch (_: Exception) {
+            }
+        }
+    }
+}
--- a/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt
+++ b/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt
@ -69,6 +69,9 @@ class UnifiedSessionRepository(
    private val _activeClarify = MutableStateFlow<HostAttributedClarify?>(null)
    val activeClarify: StateFlow<HostAttributedClarify?> = _activeClarify.asStateFlow()

+    private val _hasActiveTasks = MutableStateFlow(false)
+    val hasActiveTasks: StateFlow<Boolean> = _hasActiveTasks.asStateFlow()
+
    // Mapping from (hostId, runtimeSessionId) to sessionId
    private val runtimeToSessionMap = ConcurrentHashMap<Pair<HermesHostId, String>, UnifiedSessionId>()

@ -216,6 +219,7 @@ class UnifiedSessionRepository(
        hostExecutingState.entries.removeIf { it.key.first == sessionId }
        runtimeToSessionMap.entries.removeIf { it.value == sessionId }
        sessionHostMutexes.entries.removeIf { it.key.first == sessionId }
+        _hasActiveTasks.update { hostExecutingState.values.any { it.value } }
    }

    fun registerRuntimeBinding(sessionId: UnifiedSessionId, hostId: HermesHostId, runtimeSessionId: RuntimeSessionId) {
@ -667,6 +671,8 @@ class UnifiedSessionRepository(
        sessionExecutingState.computeIfAbsent(sessionId) {
            MutableStateFlow(false)
        }.update { isAnyHostExecuting }
+        
+        _hasActiveTasks.update { hostExecutingState.values.any { it.value } }
    }

    private fun findSessionForEvent(
--- a/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt
+++ b/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt
@ -11,6 +11,7 @@ import app.hermes.mobile.core.storage.HostDao
 import app.hermes.mobile.core.storage.HostEntity
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
@ -29,7 +30,7 @@ class HermesConnectionManager(
    val restClient: HermesRestClient = HermesRestClient(),
    val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
    val runtimeFactory: (CoroutineScope, HermesHost) -> HermesHostRuntime = { parentScope, host ->
-        val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[kotlinx.coroutines.Job]) + Dispatchers.Default)
+        val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default)
        val hostRestClient = HermesRestClient.forHost(host.certificateFingerprint)
        val hostGatewayClient = JsonRpcGatewayClient(
            client = JsonRpcGatewayClient.defaultClient(host.certificateFingerprint),
@ -122,21 +123,32 @@ class HermesConnectionManager(
    }

    fun getOrCreateRuntime(host: HermesHost): HermesHostRuntime {
-        return runtimes.computeIfAbsent(host.id) {
-            val rt = runtimeFactory(scope, host)
-            // Forward events sequentially
-            scope.launch {
-                rt.events.collect { event ->
-                    dispatchEvent(event)
-                }
+        var rt = runtimes[host.id]
+        if (rt != null) return rt
+        
+        synchronized(runtimes) {
+            rt = runtimes[host.id]
+            if (rt != null) return rt
+            
+            val created = runtimeFactory(scope, host)
+            subscribeToRuntime(created, host.id)
+            runtimes[host.id] = created
+            return created
+        }
+    }
+
+    private fun subscribeToRuntime(rt: HermesHostRuntime, hostId: HermesHostId) {
+        // Forward events sequentially, start undispatched to attach collector immediately
+        scope.launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) {
+            rt.events.collect { event ->
+                dispatchEvent(event)
            }
-            // Update host status in DB on change
-            scope.launch {
-                rt.status.collect { st ->
-                    hostDao.updateHostStatus(host.id.value, st.name, System.currentTimeMillis())
-                }
+        }
+        // Update host status in DB on change
+        scope.launch {
+            rt.status.collect { st ->
+                hostDao.updateHostStatus(hostId.value, st.name, System.currentTimeMillis())
            }
-            rt
        }
    }

@ -179,8 +191,45 @@ class HermesConnectionManager(
    }

    suspend fun refreshAllHosts() {
-        val currentHosts = hostDao.getHosts()
-        _hosts.value = currentHosts.map { it.toDomain() }
+        val currentHosts = hostDao.getHosts().map { it.toDomain() }
+        _hosts.value = currentHosts
+
+        val validIds = currentHosts.map { it.id }.toSet()
+        for ((id, rt) in runtimes) {
+            if (id !in validIds) {
+                rt.close()
+                runtimes.remove(id)
+            }
+        }
+        for (h in currentHosts) {
+            val existingRt = runtimes[h.id]
+            if (existingRt != null) {
+                existingRt.updateHost(h)
+            } else {
+                getOrCreateRuntime(h)
+            }
+        }
+    }
+
+    fun onNetworkAvailable() {
+        for ((_, rt) in runtimes) {
+            if (rt.host.value.enabled) {
+                rt.onNetworkRestored()
+            }
+        }
+    }
+
+    fun onNetworkLost() {
+        for ((_, rt) in runtimes) {
+            rt.onNetworkLost()
+        }
+    }
+
+    fun close() {
+        for ((_, rt) in runtimes) {
+            rt.close()
+        }
+        runtimes.clear()
    }

    private fun HostEntity.toDomain(): HermesHost {
--- a/app/src/main/java/app/hermes/mobile/core/runtime/HermesHostRuntime.kt
+++ b/app/src/main/java/app/hermes/mobile/core/runtime/HermesHostRuntime.kt
@ -23,9 +23,12 @@ import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import java.io.IOException
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
 import kotlin.math.min
 import kotlin.random.Random

@ -81,23 +84,56 @@ class HermesHostRuntime(
        }
    }

-    private var reconnectJob: Job? = null
-    private var autoReconnectEnabled = false
-    private var reconnectAttempt = 0
+    companion object {
+        const val MAX_RECONNECT_ATTEMPTS = 5
+    }

-    init {
+    private val reconnectMutex = Mutex()
+    private var reconnectJob: Job? = null
+    private val autoReconnectEnabled = AtomicBoolean(false)
+    private val reconnectAttempt = AtomicInteger(0)
+
+    fun isAutoReconnectActive(): Boolean = autoReconnectEnabled.get()
+
+    fun getReconnectAttemptCount(): Int = reconnectAttempt.get()
+
+    fun onNetworkRestored() {
+        reconnectAttempt.set(0)
+        if (_host.value.enabled && (_status.value == HostStatus.ERROR || _status.value == HostStatus.OFFLINE)) {
+            autoReconnectEnabled.set(true)
+            scheduleReconnect()
+        }
+    }
+
+    fun onNetworkLost() {
        scope.launch {
-            gatewayClient.events.collect { event ->
-                dispatchEvent(HostGatewayEvent(hostId, event))
+            reconnectMutex.withLock {
+                reconnectJob?.cancel()
+                reconnectJob = null
            }
        }
+        if (_status.value == HostStatus.CONNECTING || _status.value == HostStatus.ONLINE) {
+            _status.value = HostStatus.OFFLINE
+        }
+    }

-        scope.launch {
+    init {
+        scope.launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) {
+            gatewayClient.events.collect { event ->
+                _events.emit(HostGatewayEvent(hostId, event))
+            }
+        }
+        scope.launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) {
            gatewayClient.connectionState.collect { state ->
                when (state) {
                    is ConnectionState.Connected -> {
-                        reconnectAttempt = 0
-                        reconnectJob?.cancel()
+                        reconnectAttempt.set(0)
+                        scope.launch {
+                            reconnectMutex.withLock {
+                                reconnectJob?.cancel()
+                                reconnectJob = null
+                            }
+                        }
                        _status.value = HostStatus.ONLINE
                        updateLastSeen()
                    }
@ -105,13 +141,18 @@ class HermesHostRuntime(
                        _status.value = HostStatus.CONNECTING
                    }
                    is ConnectionState.AuthExpired -> {
-                        autoReconnectEnabled = false
-                        reconnectJob?.cancel()
+                        autoReconnectEnabled.set(false)
+                        scope.launch {
+                            reconnectMutex.withLock {
+                                reconnectJob?.cancel()
+                                reconnectJob = null
+                            }
+                        }
                        _status.value = HostStatus.AUTH_EXPIRED
                    }
                    is ConnectionState.Failed -> {
                        _status.value = HostStatus.ERROR
-                        if (autoReconnectEnabled) {
+                        if (autoReconnectEnabled.get()) {
                            scheduleReconnect()
                        }
                    }
@ -119,7 +160,7 @@ class HermesHostRuntime(
                        if (_status.value != HostStatus.AUTH_EXPIRED && _status.value != HostStatus.AUTH_REQUIRED) {
                            _status.value = HostStatus.OFFLINE
                        }
-                        if (autoReconnectEnabled) {
+                        if (autoReconnectEnabled.get()) {
                            scheduleReconnect()
                        }
                    }
@ -151,7 +192,12 @@ class HermesHostRuntime(
    }

    suspend fun connect(): Result<Unit> {
-        autoReconnectEnabled = true
+        autoReconnectEnabled.set(true)
+        reconnectAttempt.set(0)
+        reconnectMutex.withLock {
+            reconnectJob?.cancel()
+            reconnectJob = null
+        }
        return connectInternal()
    }

@ -161,7 +207,11 @@ class HermesHostRuntime(

        return try {
            val statusResult = restClient.getStatus(currentHost.baseUrl, currentHost.allowCleartext)
-            val sStatus = statusResult.getOrNull() ?: HermesServerStatus()
+            if (statusResult.isFailure) {
+                _status.value = HostStatus.ERROR
+                return Result.failure(statusResult.exceptionOrNull() ?: IOException("Failed to check host status"))
+            }
+            val sStatus = statusResult.getOrThrow()
            _serverStatus.value = sStatus

            var ticket: String? = null
@ -259,6 +309,10 @@ class HermesHostRuntime(
                ticket = ticket,
                allowCleartext = currentHost.allowCleartext
            )
+            gatewayClient.awaitGatewayReady(10_000)
+            reconnectAttempt.set(0)
+            _status.value = HostStatus.ONLINE
+            updateLastSeen()
            Result.success(Unit)
        } catch (e: Exception) {
            _status.value = HostStatus.ERROR
@ -266,26 +320,61 @@ class HermesHostRuntime(
        }
    }

-    private fun scheduleReconnect() {
-        if (reconnectJob?.isActive == true) return
-        reconnectJob = scope.launch {
-            val baseDelay = min(30_000L, (1000L * (1 shl min(reconnectAttempt, 5))))
-            val jitter = Random.nextLong(0, 1000)
-            val totalDelay = baseDelay + jitter
-            reconnectAttempt++
+    internal fun scheduleReconnect() {
+        if (!autoReconnectEnabled.get() || !_host.value.enabled) return
+        if (reconnectAttempt.get() >= MAX_RECONNECT_ATTEMPTS) {
+            autoReconnectEnabled.set(false)
+            _status.value = HostStatus.ERROR
+            return
+        }

-            delay(totalDelay)
-            try {
-                connectInternal()
-            } catch (_: Exception) {
+        scope.launch {
+            reconnectMutex.withLock {
+                if (reconnectJob?.isActive == true) return@withLock
+                if (!autoReconnectEnabled.get() || !_host.value.enabled) return@withLock
+                val currentAttempt = reconnectAttempt.get()
+                if (currentAttempt >= MAX_RECONNECT_ATTEMPTS) {
+                    autoReconnectEnabled.set(false)
+                    _status.value = HostStatus.ERROR
+                    return@withLock
+                }
+
+                reconnectJob = scope.launch {
+                    val attempt = reconnectAttempt.getAndIncrement()
+                    val baseDelay = min(30_000L, 1000L * (1 shl min(attempt, 5)))
+                    val jitter = Random.nextLong(0, 1000)
+                    val totalDelay = baseDelay + jitter
+
+                    delay(totalDelay)
+                    if (!autoReconnectEnabled.get() || !_host.value.enabled) return@launch
+                    val res = connectInternal()
+                    if (res.isFailure && autoReconnectEnabled.get()) {
+                        if (reconnectAttempt.get() >= MAX_RECONNECT_ATTEMPTS) {
+                            autoReconnectEnabled.set(false)
+                            _status.value = HostStatus.ERROR
+                        } else {
+                            scheduleReconnect()
+                        }
+                    }
+                }
            }
        }
    }

    fun disconnect() {
-        autoReconnectEnabled = false
-        reconnectJob?.cancel()
+        autoReconnectEnabled.set(false)
+        reconnectAttempt.set(0)
+        scope.launch {
+            reconnectMutex.withLock {
+                reconnectJob?.cancel()
+                reconnectJob = null
+            }
+        }
        gatewayClient.disconnect()
+        try {
+            restClient.client.dispatcher.cancelAll()
+            restClient.client.connectionPool.evictAll()
+        } catch (_: Exception) {}
        _status.value = HostStatus.OFFLINE
    }

--- a/app/src/main/java/app/hermes/mobile/core/service/HermesTaskForegroundService.kt
+++ b/app/src/main/java/app/hermes/mobile/core/service/HermesTaskForegroundService.kt
@ -0,0 +1,118 @@
+package app.hermes.mobile.core.service
+
+import android.app.Notification
+import android.app.NotificationChannel
+import android.app.NotificationManager
+import android.app.PendingIntent
+import android.app.Service
+import android.content.Context
+import android.content.Intent
+import android.content.pm.ServiceInfo
+import android.os.Build
+import android.os.IBinder
+import androidx.core.app.NotificationCompat
+import androidx.core.app.ServiceCompat
+import app.hermes.mobile.HermesApplication
+import app.hermes.mobile.MainActivity
+import app.hermes.mobile.R
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.launch
+
+class HermesTaskForegroundService : Service() {
+
+    private val serviceScope = CoroutineScope(SupervisorJob() + Dispatchers.Main)
+    
+    companion object {
+        private const val CHANNEL_ID = "hermes_agent_active_channel"
+        private const val NOTIFICATION_ID = 1001
+
+        fun startIfRequired(context: Context, hasActiveTasks: Boolean) {
+            val intent = Intent(context, HermesTaskForegroundService::class.java)
+            if (hasActiveTasks) {
+                if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
+                    context.startForegroundService(intent)
+                } else {
+                    context.startService(intent)
+                }
+            }
+        }
+    }
+
+    override fun onCreate() {
+        super.onCreate()
+        createNotificationChannel()
+        startForegroundWithNotification()
+
+        val app = application as HermesApplication
+        val repo = app.container.unifiedSessionRepo
+
+        serviceScope.launch {
+            repo.hasActiveTasks.collectLatest { hasActive ->
+                if (!hasActive) {
+                    stopSelf()
+                }
+            }
+        }
+    }
+
+    override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
+        startForegroundWithNotification()
+        return START_NOT_STICKY
+    }
+
+    override fun onBind(intent: Intent?): IBinder? = null
+
+    override fun onDestroy() {
+        super.onDestroy()
+        serviceScope.cancel()
+    }
+
+    private fun startForegroundWithNotification() {
+        val intent = Intent(this, MainActivity::class.java).apply {
+            flags = Intent.FLAG_ACTIVITY_SINGLE_TOP or Intent.FLAG_ACTIVITY_CLEAR_TOP
+        }
+        val pendingIntentFlags = if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
+            PendingIntent.FLAG_IMMUTABLE or PendingIntent.FLAG_UPDATE_CURRENT
+        } else {
+            PendingIntent.FLAG_UPDATE_CURRENT
+        }
+        val pendingIntent = PendingIntent.getActivity(this, 0, intent, pendingIntentFlags)
+
+        val notification: Notification = NotificationCompat.Builder(this, CHANNEL_ID)
+            .setContentTitle("Hermes Agent active")
+            .setContentText("Syncing active remote agent session...")
+            .setSmallIcon(android.R.drawable.stat_notify_sync)
+            .setContentIntent(pendingIntent)
+            .setOngoing(true)
+            .build()
+
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
+            ServiceCompat.startForeground(
+                this,
+                NOTIFICATION_ID,
+                notification,
+                ServiceInfo.FOREGROUND_SERVICE_TYPE_DATA_SYNC
+            )
+        } else {
+            startForeground(NOTIFICATION_ID, notification)
+        }
+    }
+
+    private fun createNotificationChannel() {
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
+            val name = "Agent Status"
+            val descriptionText = "Shows when a remote agent is actively processing a task"
+            val importance = NotificationManager.IMPORTANCE_LOW
+            val channel = NotificationChannel(CHANNEL_ID, name, importance).apply {
+                description = descriptionText
+            }
+            val notificationManager: NotificationManager =
+                getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager
+            notificationManager.createNotificationChannel(channel)
+        }
+    }
+}
--- a/app/src/main/java/app/hermes/mobile/core/storage/MigrationHelper.kt
+++ b/app/src/main/java/app/hermes/mobile/core/storage/MigrationHelper.kt
@ -1,6 +1,10 @@
 package app.hermes.mobile.core.storage

 import android.content.Context
+import androidx.datastore.core.DataStore
+import androidx.datastore.preferences.core.Preferences
+import androidx.datastore.preferences.core.booleanPreferencesKey
+import androidx.datastore.preferences.core.edit
 import androidx.datastore.preferences.core.stringPreferencesKey
 import app.hermes.mobile.core.model.HermesConnection
 import app.hermes.mobile.core.model.HostStatus
@ -11,33 +15,49 @@ import kotlinx.serialization.json.Json
 object MigrationHelper {
    private val json = Json { ignoreUnknownKeys = true }
    private val connectionsKey = stringPreferencesKey("saved_connections")
+    private val migrationCompletedKey = booleanPreferencesKey("migration_completed")

-    suspend fun migrateLegacyConnections(context: Context, hostDao: HostDao) {
+    suspend fun migrateLegacyConnections(context: Context, hostDao: HostDao): Boolean {
+        return migrateLegacyConnections(context.dataStore, hostDao)
+    }
+
+    suspend fun migrateLegacyConnections(dataStore: DataStore<Preferences>, hostDao: HostDao): Boolean {
        try {
-            val preferences = context.dataStore.data.firstOrNull() ?: return
-            val raw = preferences[connectionsKey] ?: return
-            if (raw.isBlank()) return
+            val preferences = dataStore.data.firstOrNull() ?: return false
+            val isCompleted = preferences[migrationCompletedKey] ?: false
+            if (isCompleted) {
+                return false
+            }

-            val legacyList = json.decodeFromString<List<HermesConnection>>(raw)
-            for (legacy in legacyList) {
-                val existing = hostDao.getHost(legacy.id)
-                if (existing == null) {
-                    hostDao.insertOrUpdateHost(
-                        HostEntity(
-                            id = legacy.id,
-                            displayName = legacy.name,
-                            baseUrl = legacy.baseUrl,
-                            allowCleartext = legacy.allowCleartext,
-                            enabled = true,
-                            lastSeenAt = legacy.createdAt,
-                            lastKnownStatus = HostStatus.OFFLINE.name,
-                            certificateFingerprint = null
+            val raw = preferences[connectionsKey]
+            if (!raw.isNullOrBlank()) {
+                val legacyList = json.decodeFromString<List<HermesConnection>>(raw)
+                for (legacy in legacyList) {
+                    val existing = hostDao.getHost(legacy.id)
+                    if (existing == null) {
+                        hostDao.insertOrUpdateHost(
+                            HostEntity(
+                                id = legacy.id,
+                                displayName = legacy.name,
+                                baseUrl = legacy.baseUrl,
+                                allowCleartext = legacy.allowCleartext,
+                                enabled = true,
+                                lastSeenAt = legacy.createdAt,
+                                lastKnownStatus = HostStatus.OFFLINE.name,
+                                certificateFingerprint = null
+                            )
                        )
-                    )
+                    }
                }
            }
+
+            dataStore.edit { prefs ->
+                prefs[migrationCompletedKey] = true
+                prefs.remove(connectionsKey)
+            }
+            return true
        } catch (_: Exception) {
-            // Ignore migration failure gracefully
+            return false
        }
    }
 }
--- a/app/src/test/java/app/hermes/mobile/core/network/GracefulCloseTest.kt
+++ b/app/src/test/java/app/hermes/mobile/core/network/GracefulCloseTest.kt
@ -0,0 +1,90 @@
+package app.hermes.mobile.core.network
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import okhttp3.Response
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import okhttp3.mockwebserver.MockResponse
+import okhttp3.mockwebserver.MockWebServer
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import java.util.concurrent.atomic.AtomicBoolean
+
+class GracefulCloseTest {
+
+    private lateinit var server: MockWebServer
+    private lateinit var client: JsonRpcGatewayClient
+
+    @Before
+    fun setUp() {
+        server = MockWebServer()
+        server.start()
+        client = JsonRpcGatewayClient()
+    }
+
+    @After
+    fun tearDown() {
+        client.disconnect()
+        try {
+            server.shutdown()
+        } catch (_: Exception) {}
+    }
+
+    @Test
+    fun testServerInitiatedGracefulCloseHandshake() = runBlocking {
+        val serverReceivedClosingAck = CompletableDeferred<Int>()
+        var serverWs: WebSocket? = null
+
+        server.enqueue(
+            MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
+                override fun onOpen(webSocket: WebSocket, response: Response) {
+                    serverWs = webSocket
+                    webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""")
+                }
+
+                override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
+                    serverReceivedClosingAck.complete(code)
+                }
+            })
+        )
+
+        val wsUrl = "ws://${server.hostName}:${server.port}/api/ws"
+        client.connect(wsUrl, allowCleartext = true)
+        client.awaitGatewayReady(5000)
+        assertEquals(ConnectionState.Connected, client.connectionState.value)
+
+        // Server initiates graceful close
+        serverWs?.close(1000, "Server stopping")
+
+        // Client onClosing acknowledges close frame
+        val ackCode = withTimeout(5000) {
+            serverReceivedClosingAck.await()
+        }
+        assertEquals("Client must acknowledge graceful close with code 1000", 1000, ackCode)
+    }
+
+    @Test
+    fun testClientDisconnectTransitionsStateAndFailsPendingRequests() = runBlocking {
+        server.enqueue(
+            MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
+                override fun onOpen(webSocket: WebSocket, response: Response) {
+                    webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""")
+                }
+            })
+        )
+
+        val wsUrl = "ws://${server.hostName}:${server.port}/api/ws"
+        client.connect(wsUrl, allowCleartext = true)
+        client.awaitGatewayReady(5000)
+        assertEquals(ConnectionState.Connected, client.connectionState.value)
+
+        // Disconnect immediately cleans up connection
+        client.disconnect()
+        assertEquals(ConnectionState.Disconnected, client.connectionState.value)
+    }
+}
--- a/app/src/test/java/app/hermes/mobile/core/network/ReadyDeferredRaceTest.kt
+++ b/app/src/test/java/app/hermes/mobile/core/network/ReadyDeferredRaceTest.kt
@ -0,0 +1,125 @@
+package app.hermes.mobile.core.network
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import okhttp3.Response
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import okhttp3.mockwebserver.MockResponse
+import okhttp3.mockwebserver.MockWebServer
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import java.io.IOException
+
+class ReadyDeferredRaceTest {
+
+    private lateinit var server1: MockWebServer
+    private lateinit var server2: MockWebServer
+    private lateinit var client: JsonRpcGatewayClient
+
+    @Before
+    fun setUp() {
+        server1 = MockWebServer()
+        server1.start()
+        server2 = MockWebServer()
+        server2.start()
+        client = JsonRpcGatewayClient()
+    }
+
+    @After
+    fun tearDown() {
+        client.disconnect()
+        try {
+            server1.shutdown()
+        } catch (_: Exception) {}
+        try {
+            server2.shutdown()
+        } catch (_: Exception) {}
+    }
+
+    @Test
+    fun testAwaitGatewayReadyDoesNotHangOnStaleDeferredWhenConnectReinvoked() = runBlocking {
+        // Server 1 accepts socket but never sends gateway.ready
+        server1.enqueue(
+            MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
+                override fun onOpen(webSocket: WebSocket, response: Response) {
+                    // Intentionally hang without gateway.ready
+                }
+            })
+        )
+
+        // Server 2 accepts socket and sends gateway.ready
+        val server2WsDeferred = CompletableDeferred<WebSocket>()
+        server2.enqueue(
+            MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
+                override fun onOpen(webSocket: WebSocket, response: Response) {
+                    server2WsDeferred.complete(webSocket)
+                    webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"2.0.0"}}}""")
+                }
+            })
+        )
+
+        val wsUrl1 = "ws://${server1.hostName}:${server1.port}/api/ws"
+        val wsUrl2 = "ws://${server2.hostName}:${server2.port}/api/ws"
+
+        // 1. Connect to Server 1
+        client.connect(wsUrl1, allowCleartext = true)
+
+        // 2. Start waiting for gateway.ready in background
+        val awaiter = async {
+            try {
+                client.awaitGatewayReady(4000)
+                true
+            } catch (e: Exception) {
+                false
+            }
+        }
+
+        // Give a brief moment for awaiter to capture deferred
+        delay(50)
+
+        // 3. Immediately re-invoke connect to Server 2 before Server 1 completes
+        client.connect(wsUrl2, allowCleartext = true)
+
+        // 4. Awaiting on the active connection must succeed promptly when Server 2 sends gateway.ready
+        val activeAwaiter = async {
+            client.awaitGatewayReady(4000)
+            true
+        }
+
+        val ready = withTimeout(4000) {
+            activeAwaiter.await()
+        }
+
+        assertTrue("Active connection awaitGatewayReady must succeed", ready)
+        assertEquals("Active connection state must be Connected", ConnectionState.Connected, client.connectionState.value)
+    }
+
+    @Test
+    fun testAwaitGatewayReadyReturnsImmediatelyIfAlreadyConnected() = runBlocking {
+        server1.enqueue(
+            MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
+                override fun onOpen(webSocket: WebSocket, response: Response) {
+                    webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""")
+                }
+            })
+        )
+
+        client.connect("ws://${server1.hostName}:${server1.port}/api/ws", allowCleartext = true)
+        client.awaitGatewayReady(5000)
+        assertEquals(ConnectionState.Connected, client.connectionState.value)
+
+        // Calling awaitGatewayReady again when already Connected should return immediately (within 100ms)
+        withTimeout(100) {
+            client.awaitGatewayReady(5000)
+        }
+        assertTrue(true)
+    }
+}
--- a/app/src/test/java/app/hermes/mobile/core/runtime/ConnectReadinessTest.kt
+++ b/app/src/test/java/app/hermes/mobile/core/runtime/ConnectReadinessTest.kt
@ -0,0 +1,107 @@
+package app.hermes.mobile.core.runtime
+
+import app.hermes.mobile.core.model.HermesHost
+import app.hermes.mobile.core.model.HermesHostId
+import app.hermes.mobile.core.model.HostStatus
+import app.hermes.mobile.core.security.InMemoryTokenVault
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import okhttp3.Response
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import okhttp3.mockwebserver.Dispatcher
+import okhttp3.mockwebserver.MockResponse
+import okhttp3.mockwebserver.MockWebServer
+import okhttp3.mockwebserver.RecordedRequest
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+
+class ConnectReadinessTest {
+
+    private lateinit var server: MockWebServer
+
+    @Before
+    fun setUp() {
+        server = MockWebServer()
+    }
+
+    @After
+    fun tearDown() {
+        try {
+            server.shutdown()
+        } catch (_: Exception) {}
+    }
+
+    @Test
+    fun testConnectDoesNotReturnSuccessUntilGatewayReadyReceived() = runBlocking {
+        val serverWsDeferred = CompletableDeferred<WebSocket>()
+
+        server.dispatcher = object : Dispatcher() {
+            override fun dispatch(request: RecordedRequest): MockResponse {
+                val path = request.path ?: ""
+                if (path.contains("/api/status")) {
+                    return MockResponse().setResponseCode(200).setBody("""{"version":"1.0.0","auth_required":false}""")
+                }
+                if (path.contains("/api/ws")) {
+                    return MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
+                        override fun onOpen(webSocket: WebSocket, response: Response) {
+                            serverWsDeferred.complete(webSocket)
+                            // Intentionally DO NOT send gateway.ready yet
+                        }
+                    })
+                }
+                return MockResponse().setResponseCode(404)
+            }
+        }
+        server.start()
+
+        val host = HermesHost(
+            id = HermesHostId("h-readiness"),
+            displayName = "Readiness Host",
+            baseUrl = "http://${server.hostName}:${server.port}",
+            allowCleartext = true,
+            enabled = true,
+            lastKnownStatus = HostStatus.OFFLINE
+        )
+
+        val runtime = HermesHostRuntime(
+            initialHost = host,
+            tokenVault = InMemoryTokenVault()
+        )
+
+        // Launch connect() asynchronously
+        val connectJob = async {
+            runtime.connect()
+        }
+
+        // Wait for socket to be opened on server side
+        val serverWs = withTimeout(5000) {
+            serverWsDeferred.await()
+        }
+
+        // Wait 300ms: during this time connectJob must NOT be completed because gateway.ready was not sent
+        delay(300)
+        assertFalse("connect() must not complete before gateway.ready is received", connectJob.isCompleted)
+        assertEquals("Host status must be CONNECTING before gateway.ready", HostStatus.CONNECTING, runtime.status.value)
+
+        // Now send gateway.ready from server
+        serverWs.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0","session_count":0}}}""")
+
+        // Now connect() must complete with success
+        val result = withTimeout(5000) {
+            connectJob.await()
+        }
+
+        assertTrue("connect() must succeed after gateway.ready", result.isSuccess)
+        assertEquals("Host status must be ONLINE after gateway.ready", HostStatus.ONLINE, runtime.status.value)
+
+        runtime.close()
+    }
+}
--- a/app/src/test/java/app/hermes/mobile/core/runtime/ReconnectLimitTest.kt
+++ b/app/src/test/java/app/hermes/mobile/core/runtime/ReconnectLimitTest.kt
@ -0,0 +1,79 @@
+package app.hermes.mobile.core.runtime
+
+import app.hermes.mobile.core.model.HermesHost
+import app.hermes.mobile.core.model.HermesHostId
+import app.hermes.mobile.core.model.HermesServerStatus
+import app.hermes.mobile.core.model.HostStatus
+import app.hermes.mobile.core.network.HermesRestClient
+import app.hermes.mobile.core.security.InMemoryTokenVault
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import java.io.IOException
+import java.util.concurrent.atomic.AtomicInteger
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class ReconnectLimitTest {
+
+    @Test
+    fun testMaxReconnectAttemptsStopsLoopAndResetsOnNetworkRestore() = runTest {
+        val attempts = AtomicInteger(0)
+
+        val failingRestClient = object : HermesRestClient() {
+            override suspend fun getStatus(baseUrl: String, allowCleartext: Boolean): Result<HermesServerStatus> {
+                attempts.incrementAndGet()
+                return Result.failure(IOException("Simulated 500 server error"))
+            }
+        }
+
+        val host = HermesHost(
+            id = HermesHostId("host-limit"),
+            displayName = "Limit Host",
+            baseUrl = "http://mock-host:9119",
+            allowCleartext = true,
+            enabled = true,
+            lastKnownStatus = HostStatus.OFFLINE
+        )
+
+        val runtime = HermesHostRuntime(
+            initialHost = host,
+            restClient = failingRestClient,
+            tokenVault = InMemoryTokenVault(),
+            scope = backgroundScope
+        )
+
+        // Initial connect fails due to simulated 500
+        val initialRes = runtime.connect()
+        assertTrue("Initial connect must fail", initialRes.isFailure)
+
+        // Trigger automatic reconnect loop
+        runtime.scheduleReconnect()
+
+        // Advance virtual time through all backoff delays (1s, 2s, 4s, 8s, 16s, etc.)
+        advanceTimeBy(60_000)
+        advanceUntilIdle()
+
+        // After max attempts (5), autoReconnect must stop and status must be ERROR
+        assertTrue("Status must be ERROR after reaching max reconnect attempts",
+            runtime.status.value == HostStatus.ERROR || runtime.status.value == HostStatus.OFFLINE
+        )
+        assertFalse("Auto reconnect must be stopped after max attempts", runtime.isAutoReconnectActive())
+        assertTrue("Reconnect attempt count must be at least MAX_RECONNECT_ATTEMPTS (5)",
+            runtime.getReconnectAttemptCount() >= HermesHostRuntime.MAX_RECONNECT_ATTEMPTS
+        )
+
+        // Now simulate network restore
+        runtime.onNetworkRestored()
+
+        // Verify attempts counter is reset to 0
+        assertEquals("Reconnect attempts must be reset to 0 upon network restore", 0, runtime.getReconnectAttemptCount())
+        assertTrue("Auto reconnect must be re-enabled on network restore", runtime.isAutoReconnectActive())
+
+        runtime.close()
+    }
+}
--- a/app/src/test/java/app/hermes/mobile/core/runtime/ReconnectSingleFlightTest.kt
+++ b/app/src/test/java/app/hermes/mobile/core/runtime/ReconnectSingleFlightTest.kt
@ -0,0 +1,127 @@
+package app.hermes.mobile.core.runtime
+
+import app.hermes.mobile.core.model.HermesHost
+import app.hermes.mobile.core.model.HermesHostId
+import app.hermes.mobile.core.model.HostStatus
+import app.hermes.mobile.core.network.ConnectionState
+import app.hermes.mobile.core.network.HermesRestClient
+import app.hermes.mobile.core.network.JsonRpcGatewayClient
+import app.hermes.mobile.core.security.InMemoryTokenVault
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.asCoroutineDispatcher
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.joinAll
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runTest
+import okhttp3.OkHttpClient
+import okhttp3.Response
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import okhttp3.mockwebserver.Dispatcher
+import okhttp3.mockwebserver.MockResponse
+import okhttp3.mockwebserver.MockWebServer
+import okhttp3.mockwebserver.RecordedRequest
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class ReconnectSingleFlightTest {
+
+    private lateinit var server: MockWebServer
+
+    @Before
+    fun setUp() {
+        server = MockWebServer()
+    }
+
+    @After
+    fun tearDown() {
+        try {
+            server.shutdown()
+        } catch (_: Exception) {}
+    }
+
+    @Test
+    fun testParallelReconnectTriggersResultInSingleActiveReconnectLoop() = runBlocking {
+        val concurrentRequests = AtomicInteger(0)
+        val maxConcurrentRequests = AtomicInteger(0)
+        val totalStatusRequests = AtomicInteger(0)
+
+        server.dispatcher = object : Dispatcher() {
+            override fun dispatch(request: RecordedRequest): MockResponse {
+                val path = request.path ?: ""
+                if (path.contains("/api/status")) {
+                    totalStatusRequests.incrementAndGet()
+                    val current = concurrentRequests.incrementAndGet()
+                    var max = maxConcurrentRequests.get()
+                    while (current > max && !maxConcurrentRequests.compareAndSet(max, current)) {
+                        max = maxConcurrentRequests.get()
+                    }
+                    Thread.sleep(100) // Hold request briefly to expose concurrency
+                    concurrentRequests.decrementAndGet()
+                    return MockResponse()
+                        .setResponseCode(200)
+                        .setBody("""{"version":"1.0.0","auth_required":false}""")
+                }
+                if (path.contains("/api/ws")) {
+                    return MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
+                        override fun onOpen(webSocket: WebSocket, response: Response) {
+                            // Immediately close to trigger reconnect
+                            webSocket.close(1001, "Simulated disconnect")
+                        }
+                    })
+                }
+                return MockResponse().setResponseCode(404)
+            }
+        }
+        server.start()
+
+        val host = HermesHost(
+            id = HermesHostId("host-sf"),
+            displayName = "Single Flight Host",
+            baseUrl = "http://${server.hostName}:${server.port}",
+            allowCleartext = true,
+            enabled = true,
+            lastKnownStatus = HostStatus.ONLINE
+        )
+
+        val runtime = HermesHostRuntime(
+            initialHost = host,
+            tokenVault = InMemoryTokenVault()
+        )
+
+        // Connect first to enable autoReconnect
+        runtime.connect()
+
+        // Fire 20 parallel failure / reconnect triggers across a multi-thread pool
+        val pool = Executors.newFixedThreadPool(8).asCoroutineDispatcher()
+        val jobs = (1..20).map {
+            runtime.scope.launch(pool) {
+                runtime.scheduleReconnect()
+            }
+        }
+        jobs.joinAll()
+        pool.close()
+
+        // Wait for reconnects to process
+        kotlinx.coroutines.delay(1500)
+
+        runtime.close()
+
+        // Under single flight, reconnect attempts are serialized, max concurrent requests must be <= 1
+        assertEquals("Max concurrent connect requests must be at most 1", 1, maxConcurrentRequests.get())
+    }
+}
--- a/app/src/test/java/app/hermes/mobile/core/runtime/RuntimeSubscriptionGapTest.kt
+++ b/app/src/test/java/app/hermes/mobile/core/runtime/RuntimeSubscriptionGapTest.kt
@ -0,0 +1,137 @@
+package app.hermes.mobile.core.runtime
+
+import app.hermes.mobile.core.model.GatewayEvent
+import app.hermes.mobile.core.model.HermesHost
+import app.hermes.mobile.core.model.HermesHostId
+import app.hermes.mobile.core.model.HostGatewayEvent
+import app.hermes.mobile.core.model.HostStatus
+import app.hermes.mobile.core.network.HermesRestClient
+import app.hermes.mobile.core.network.JsonRpcGatewayClient
+import app.hermes.mobile.core.security.InMemoryTokenVault
+import app.hermes.mobile.core.storage.FakeHostDao
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.asCoroutineDispatcher
+import kotlinx.coroutines.async
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.joinAll
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeout
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicInteger
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class RuntimeSubscriptionGapTest {
+
+    @Test
+    fun testInitialEventDeliveredWithoutSubscriptionRaceGap() = runBlocking {
+        val hostDao = FakeHostDao()
+        val tokenVault = InMemoryTokenVault()
+        val restClient = HermesRestClient()
+
+        val host = HermesHost(
+            id = HermesHostId("h-gap-1"),
+            displayName = "Gap Host",
+            baseUrl = "http://localhost:8080",
+            allowCleartext = true,
+            enabled = true,
+            lastKnownStatus = HostStatus.OFFLINE
+        )
+
+        // Custom factory that immediately emits an event as soon as runtime is created
+        var manager: HermesConnectionManager? = null
+        manager = HermesConnectionManager(
+            hostDao = hostDao,
+            tokenVault = tokenVault,
+            restClient = restClient,
+            runtimeFactory = { parentScope, h ->
+                val childScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+                val rt = HermesHostRuntime(
+                    initialHost = h,
+                    restClient = restClient,
+                    tokenVault = tokenVault,
+                    scope = childScope
+                )
+                // Runtime fires an immediate event right upon instantiation
+                rt.gatewayClient.handleIncomingMessage(
+                    """{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}"""
+                )
+                rt
+            }
+        )
+
+        val receivedEvents = ConcurrentLinkedQueue<HostGatewayEvent>()
+        val collectorJob = launch(Dispatchers.Default) {
+            manager.allEvents.collect { event ->
+                receivedEvents.add(event)
+            }
+        }
+
+        // Create runtime
+        val runtime = manager.getOrCreateRuntime(host)
+        assertNotNull(runtime)
+
+        // Verify the event emitted during/immediately after creation was collected
+        val received = withTimeout(3000) {
+            while (receivedEvents.isEmpty()) {
+                delay(20)
+            }
+            receivedEvents.poll()
+        }
+
+        assertNotNull("Initial event must not be dropped due to subscription gap", received)
+        assertEquals(host.id, received?.hostId)
+        assertTrue(received?.event is GatewayEvent.GatewayReadyEvent)
+
+        collectorJob.cancel()
+    }
+
+    @Test
+    fun testConcurrentGetOrCreateRuntimeDoesNotDeadlock() = runBlocking {
+        val hostDao = FakeHostDao()
+        val tokenVault = InMemoryTokenVault()
+        val manager = HermesConnectionManager(
+            hostDao = hostDao,
+            tokenVault = tokenVault
+        )
+
+        val host = HermesHost(
+            id = HermesHostId("h-concurrent"),
+            displayName = "Concurrent Host",
+            baseUrl = "http://localhost:9090",
+            allowCleartext = true,
+            enabled = true
+        )
+
+        val threadPool = Executors.newFixedThreadPool(8).asCoroutineDispatcher()
+        val runtimes = ConcurrentLinkedQueue<HermesHostRuntime>()
+
+        val jobs = (1..30).map {
+            launch(threadPool) {
+                val rt = manager.getOrCreateRuntime(host)
+                runtimes.add(rt)
+            }
+        }
+
+        withTimeout(5000) {
+            jobs.joinAll()
+        }
+        threadPool.close()
+
+        assertEquals(30, runtimes.size)
+        val firstRt = runtimes.peek()
+        assertTrue("All returned runtimes must be the identical instance", runtimes.all { it === firstRt })
+    }
+}
--- a/app/src/test/java/app/hermes/mobile/core/storage/MigrationOnceTest.kt
+++ b/app/src/test/java/app/hermes/mobile/core/storage/MigrationOnceTest.kt
@ -0,0 +1,101 @@
+package app.hermes.mobile.core.storage
+
+import androidx.datastore.preferences.core.PreferenceDataStoreFactory
+import androidx.datastore.preferences.core.edit
+import androidx.datastore.preferences.core.stringPreferencesKey
+import androidx.datastore.preferences.core.booleanPreferencesKey
+import app.hermes.mobile.core.model.HermesConnection
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.runBlocking
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Rule
+import org.junit.Test
+import org.junit.rules.TemporaryFolder
+
+class MigrationOnceTest {
+
+    @get:Rule
+    val tempFolder = TemporaryFolder()
+
+    private val json = Json { ignoreUnknownKeys = true }
+    private val connectionsKey = stringPreferencesKey("saved_connections")
+    private val migrationCompletedKey = booleanPreferencesKey("migration_completed")
+
+    @Test
+    fun testMigrationRunsOnceAndSetsCompletionFlagAndClearsLegacy() = runBlocking {
+        val testFile = tempFolder.newFile("datastore_test_1.preferences_pb")
+        val dataStore = PreferenceDataStoreFactory.create(produceFile = { testFile })
+        val hostDao = FakeHostDao()
+
+        // 1. Prepopulate legacy DataStore with connections
+        val legacyConnections = listOf(
+            HermesConnection(
+                id = "c1",
+                name = "Legacy Host 1",
+                baseUrl = "https://legacy1.example.com",
+                allowCleartext = false,
+                createdAt = 1000L
+            ),
+            HermesConnection(
+                id = "c2",
+                name = "Legacy Host 2",
+                baseUrl = "http://legacy2.example.com",
+                allowCleartext = true,
+                createdAt = 2000L
+            )
+        )
+        dataStore.edit { preferences ->
+            preferences[connectionsKey] = json.encodeToString(legacyConnections)
+        }
+
+        // 2. Run migration first time
+        val migratedFirst = MigrationHelper.migrateLegacyConnections(dataStore, hostDao)
+        assertTrue("First migration run must return true", migratedFirst)
+
+        // 3. Verify hosts are in DAO
+        val hosts = hostDao.getHosts()
+        assertEquals(2, hosts.size)
+        assertEquals("Legacy Host 1", hosts.find { it.id == "c1" }?.displayName)
+        assertEquals("Legacy Host 2", hosts.find { it.id == "c2" }?.displayName)
+
+        // 4. Verify completion flag is set and legacy key is removed
+        val prefsAfter = dataStore.data.first()
+        assertEquals(true, prefsAfter[migrationCompletedKey])
+        assertNull("Legacy connections key must be cleared after successful migration", prefsAfter[connectionsKey])
+
+        // 5. Subsequent run must be a no-op (return false) and not re-process
+        val migratedSecond = MigrationHelper.migrateLegacyConnections(dataStore, hostDao)
+        assertFalse("Second migration run must return false (skipped)", migratedSecond)
+    }
+
+    @Test
+    fun testMigrationRetriesIfFlagNotSet() = runBlocking {
+        val testFile = tempFolder.newFile("datastore_test_2.preferences_pb")
+        val dataStore = PreferenceDataStoreFactory.create(produceFile = { testFile })
+        val hostDao = FakeHostDao()
+
+        val legacyConnections = listOf(
+            HermesConnection(
+                id = "c3",
+                name = "Legacy Host 3",
+                baseUrl = "https://legacy3.example.com",
+                createdAt = 3000L
+            )
+        )
+        dataStore.edit { preferences ->
+            preferences[connectionsKey] = json.encodeToString(legacyConnections)
+        }
+
+        // Run migration
+        MigrationHelper.migrateLegacyConnections(dataStore, hostDao)
+
+        val hosts = hostDao.getHosts()
+        assertEquals(1, hosts.size)
+        assertEquals("c3", hosts[0].id)
+    }
+}