test: verify dashboard stream recovery

app/src/main/kotlin/app/getarcane/android/core/DashboardStreamStore.kt

@@ -224,6 +224,8 @@ class DashboardStreamStore(
                         apply(event)
                     }
                 } catch (e: ArcaneError.NotFound) {
+                    // Arcane versions without dashboard/stream continue on the REST dashboard path.
+                    // Do not retry or present a transport-failure banner until the client changes.
                     streamUnsupported = true
                     connected = false
                     return

app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardScreen.kt

@@ -76,6 +76,7 @@ import app.getarcane.android.core.ArcaneDashboardStreamClient
 import app.getarcane.android.core.DashboardActionItemKind
 import app.getarcane.android.core.DashboardActionItemSeverity
 import app.getarcane.android.core.DashboardEnvironmentStreamState
+import app.getarcane.android.core.DashboardStreamAggregateCounts
 import app.getarcane.android.core.DashboardStreamStore
 import app.getarcane.android.core.LocalArcaneManager
 import app.getarcane.android.core.friendlyErrorMessage
@@ -130,6 +131,21 @@ internal data class DashTotals(
     val stopped: Int,
 )
 
+internal fun displayedDashboardTotals(
+    streamAggregate: DashboardStreamAggregateCounts?,
+    restFallback: DashTotals?,
+    streamUpdateCount: Int?,
+): DashTotals? = streamAggregate?.let { aggregate ->
+    DashTotals(
+        running = aggregate.runningContainers,
+        total = aggregate.totalContainers,
+        images = aggregate.totalImages,
+        volumes = restFallback?.volumes,
+        updates = streamUpdateCount,
+        stopped = aggregate.stoppedContainers,
+    )
+} ?: restFallback
+
 internal enum class NeedsAttentionSeverity { Critical, Warning }
 
 internal data class NeedsAttentionItem(
@@ -196,8 +212,7 @@ fun DashboardScreen(
             .forEach { statsHistory.remove(it) }
 
         coroutineScope {
-            enabledEnvironmentIds
-                .take(DashboardStatsMaxStreams)
+            dashboardStatsStreamEnvironmentIds(enabledEnvironmentIds)
                 .forEachIndexed { index, id ->
                     launch {
                         delay(150L * (index + 1))
@@ -379,16 +394,11 @@ fun DashboardScreen(
                     )
                 }
                 item {
-                    val t = streamStore.aggregate?.let { aggregate ->
-                        DashTotals(
-                            running = aggregate.runningContainers,
-                            total = aggregate.totalContainers,
-                            images = aggregate.totalImages,
-                            volumes = totals?.volumes,
-                            updates = displayedImageUpdateCount,
-                            stopped = aggregate.stoppedContainers,
-                        )
-                    } ?: totals
+                    val t = displayedDashboardTotals(
+                        streamAggregate = streamStore.aggregate,
+                        restFallback = totals,
+                        streamUpdateCount = displayedImageUpdateCount,
+                    )
                     Column(verticalArrangement = Arrangement.spacedBy(12.dp)) {
                         Row(horizontalArrangement = Arrangement.spacedBy(12.dp)) {
                             DashboardTile("Updates", t?.updates?.let { "$it" } ?: "—", Icons.Filled.Autorenew, ArcaneGreen, Modifier.weight(1f)) {

app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardStatsHistory.kt

@@ -5,6 +5,9 @@ import app.getarcane.sdk.models.system.SystemStats
 const val DashboardStatsWindowSize = 60
 const val DashboardStatsMaxStreams = 6
 
+internal fun dashboardStatsStreamEnvironmentIds(environmentIds: List<String>): List<String> =
+    environmentIds.distinct().take(DashboardStatsMaxStreams)
+
 data class DashboardStatsSeries(
     val cpu: List<Double> = emptyList(),
     val memory: List<Double> = emptyList(),

app/src/test/java/app/getarcane/android/core/DashboardStreamStoreTest.kt

@@ -1,5 +1,6 @@
 package app.getarcane.android.core
 
+import app.getarcane.sdk.errors.ArcaneError
 import app.getarcane.sdk.models.container.ContainerStatusCounts
 import app.getarcane.sdk.models.environment.Environment
 import app.getarcane.sdk.models.image.ImageUsageCounts
@@ -14,6 +15,7 @@ import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.yield
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
@@ -303,6 +305,60 @@ class DashboardStreamStoreTest {
         assertEquals(0, client.active)
     }
 
+    @Test
+    fun unsupportedStreamStopsReconnectAndResetsForAReplacementClient() = runBlocking {
+        val scope = CoroutineScope(Dispatchers.Unconfined)
+        val unsupportedClient = UnsupportedStreamClient()
+        val store = DashboardStreamStore(scope)
+        try {
+            store.configure(unsupportedClient)
+            store.start()
+            yield()
+
+            assertTrue(store.streamUnsupported)
+            assertFalse(store.streamFailed)
+            assertFalse(store.isStreaming)
+            assertEquals(1, unsupportedClient.started)
+
+            store.retry()
+            assertEquals(1, unsupportedClient.started)
+
+            store.configure(HangingDashboardStreamClient)
+            store.start()
+            assertFalse(store.streamUnsupported)
+            assertTrue(store.isStreaming)
+        } finally {
+            store.stop()
+            scope.cancel()
+        }
+    }
+
+    @Test
+    fun repeatedTransportFailuresEnterBoundedIdleRetry() = runBlocking {
+        val scope = CoroutineScope(Dispatchers.Unconfined)
+        val client = FailingStreamClient()
+        val store = DashboardStreamStore(
+            scope = scope,
+            maxReconnectAttempts = 2,
+            maxReconnectDelayMillis = 1,
+            idleRetryMillis = 60_000,
+        )
+        try {
+            store.configure(client)
+            store.start()
+            withTimeout(1_000) {
+                while (!store.streamFailed) yield()
+            }
+
+            assertEquals(3, client.started)
+            assertTrue(store.isStreaming)
+            assertFalse(store.connected)
+        } finally {
+            store.stop()
+            scope.cancel()
+        }
+    }
+
     private fun snapshotEvent(environmentId: String, running: Int, stopped: Int, images: Int): DashboardStreamEvent =
         DashboardStreamEvent(
             type = "snapshot",
@@ -392,3 +448,28 @@ private class CountingSnapshotClient : DashboardStreamClient {
         }
     }
 }
+
+private class UnsupportedStreamClient : DashboardStreamClient {
+    var started = 0
+
+    override fun stream(): Flow<DashboardStreamEvent> = flow {
+        started++
+        yield()
+        throw ArcaneError.NotFound
+    }
+
+    override suspend fun snapshot(environmentId: String): DashboardSnapshot =
+        error("snapshot should not be called")
+}
+
+private class FailingStreamClient : DashboardStreamClient {
+    var started = 0
+
+    override fun stream(): Flow<DashboardStreamEvent> = flow {
+        started++
+        throw ArcaneError.Transport("offline")
+    }
+
+    override suspend fun snapshot(environmentId: String): DashboardSnapshot =
+        error("snapshot should not be called")
+}

app/src/test/java/app/getarcane/android/ui/screens/DashboardStatsHistoryTest.kt

@@ -6,6 +6,15 @@ import org.junit.Assert.assertNull
 import org.junit.Test
 
 class DashboardStatsHistoryTest {
+    @Test
+    fun liveStatsEnvironmentSelectionIsUniqueAndConnectionBounded() {
+        val selected = dashboardStatsStreamEnvironmentIds(
+            listOf("0", "edge-1", "edge-1", "edge-2", "edge-3", "edge-4", "edge-5", "edge-6"),
+        )
+
+        assertEquals(listOf("0", "edge-1", "edge-2", "edge-3", "edge-4", "edge-5"), selected)
+    }
+
     @Test
     fun appendKeepsRollingCpuAndMemoryWindow() {
         val series = (0..DashboardStatsWindowSize).fold(DashboardStatsSeries()) { acc, index ->

app/src/test/java/app/getarcane/android/ui/screens/DashboardTotalsTest.kt

@@ -0,0 +1,43 @@
+package app.getarcane.android.ui.screens
+
+import app.getarcane.android.core.DashboardStreamAggregateCounts
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+class DashboardTotalsTest {
+    @Test
+    fun missingStreamAggregateUsesTheRestFallback() {
+        val fallback = fallbackTotals()
+
+        assertEquals(
+            fallback,
+            displayedDashboardTotals(
+                streamAggregate = null,
+                restFallback = fallback,
+                streamUpdateCount = null,
+            ),
+        )
+    }
+
+    @Test
+    fun streamAggregateReplacesLiveCountsAndRetainsRestOnlyCounts() {
+        val result = displayedDashboardTotals(
+            streamAggregate = DashboardStreamAggregateCounts(
+                runningContainers = 8,
+                stoppedContainers = 2,
+                totalContainers = 10,
+                totalImages = 14,
+            ),
+            restFallback = fallbackTotals(),
+            streamUpdateCount = 6,
+        )
+
+        assertEquals(
+            DashTotals(running = 8, total = 10, images = 14, volumes = 4, updates = 6, stopped = 2),
+            result,
+        )
+    }
+
+    private fun fallbackTotals(): DashTotals =
+        DashTotals(running = 5, total = 7, images = 11, volumes = 4, updates = 3, stopped = 2)
+}

docs/ios-parity-task-list.md

@@ -861,17 +861,33 @@ implementation work unless current-source or runtime verification finds a regres
   - [ ] Fleet pagination and partial environment failure do not create false totals.
   - [ ] Close as verified or reopen with a focused reproduction.
 
-- [ ] **PAR-V03 — Dashboard stream foundation and live-stats recovery**
+- [x] **PAR-V03 — Dashboard stream foundation and live-stats recovery**
 
-- **Status:** Done/verify
+- **Status:** Complete
 - **Priority:** P1 if reopened
 - **Dependencies:** PAR-008
 - **Scope:** Verify reconnect, version fallback, cancellation, connection bounds, and recovery after
   server/environment changes.
 - **Acceptance criteria:**
-  - [ ] A current target server demonstrates recovery without duplicate streams or stale overwrites.
-  - [ ] Unsupported/legacy behavior is explicit.
-  - [ ] Close as verified or reopen with a focused reproduction.
+  - [x] A current target server demonstrates recovery without duplicate streams or stale overwrites.
+  - [x] Unsupported/legacy behavior is explicit.
+  - [x] Close as verified or reopen with a focused reproduction.
+- **Validation evidence (2026-08-21):**
+  - Source pins: Android base `84f822b393e2b02e8dcaf200081105104d3eb151`,
+    libarcane-kotlin `991dfdc1ee747c171ebf1b5953fe5fb61ceadfb8`, and Arcane
+    `0fd8820822f49e2da25739306bc9bc401253fa9e`.
+  - Michael's physical-device/live-server smoke test on the current target passed dashboard network
+    loss and reconnect, environment switching without stale values, stream-screen departure and
+    reopen, and recovery after force-stop. The PAR-008 regression matrix separately proves one live
+    owner and rejects stale snapshots across refresh, environment removal, and client replacement.
+  - A typed `ArcaneError.NotFound` from `dashboard/stream` is explicitly treated as a legacy server:
+    reconnect stops without a failure banner, REST totals remain authoritative, and a replacement
+    client resets stream support. Repeated transport failures enter bounded idle retry, and live
+    system stats select at most six unique environments.
+  - Focused runs passed 21 tests across `DashboardStreamStoreTest`, `DashboardStatsHistoryTest`, and
+    `DashboardTotalsTest` (0 failures, 0 errors, 0 skipped).
+  - `./gradlew :app:testDebugUnitTest :app:assembleDebug` passed all 148 unit tests and assembled the
+    debug APK; `git diff --check` passed.
 
 - [ ] **PAR-V04 — Update All environments**