fix: harden coroutine and stream ownership

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

@@ -22,6 +22,7 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.launch
 
 /**
@@ -55,6 +56,11 @@ class ActivityCenterStore(private val scope: CoroutineScope) {
     private val activityBuckets = LinkedHashMap<String, List<Activity>>()
     private val environmentNames = HashMap<String, String>()
     private val streamJobs = HashMap<String, Job>()
+    private var loadJob: Job? = null
+    private var clientGeneration = 0L
+    private var loadGeneration = 0L
+    private var streamGeneration = 0L
+    private var streamingRequested = false
 
     val filteredActivities: List<Activity>
         get() {
@@ -74,41 +80,42 @@ class ActivityCenterStore(private val scope: CoroutineScope) {
         get() = sortedUnique(activities.mapNotNull { it.resourceType })
 
     fun configure(client: ArcaneClient?) {
-        val changed = this.client == null
+        if (this.client === client) return
+        clientGeneration++
+        loadGeneration++
+        loadJob?.cancel()
+        loadJob = null
+        stopStream()
         this.client = client
-        if (changed) {
-            stopStream()
-            activities = emptyList()
-            activityBuckets.clear()
-            environmentNames.clear()
-            environmentIds = emptyList()
-            limit = PAGE_SIZE
-            hasMore = false
-            errorMessage = null
-            streamErrorMessage = null
-        }
+        activities = emptyList()
+        activityBuckets.clear()
+        environmentNames.clear()
+        environmentIds = emptyList()
+        limit = PAGE_SIZE
+        hasMore = false
+        isLoading = false
+        isLoadingMore = false
+        errorMessage = null
+        streamErrorMessage = null
     }
 
     /** Fan out `listPaginated` across all environments, bucket per env, merge + sort. */
     suspend fun load(reset: Boolean = true, refresh: Boolean = false) {
         val client = client ?: return
-        if (reset) {
-            limit = PAGE_SIZE
-            hasMore = false
-        }
+        val owningJob = currentCoroutineContext()[Job]
+        loadJob?.takeIf { it !== owningJob }?.cancel()
+        loadJob = owningJob
+        val expectedClientGeneration = clientGeneration
+        val operationGeneration = ++loadGeneration
+        val pageLimit = if (reset) PAGE_SIZE else limit
         if (activities.isEmpty() || refresh) isLoading = true
         errorMessage = null
         try {
             val environments = resolveEnvironments(client)
-            environmentIds = environments.map { it.id.rawValue }
-            environmentNames.clear()
-            environments.forEach { environmentNames[it.id.rawValue] = it.name }
-
-            val pageLimit = limit
             val results: List<Pair<ActivityEnvironment, List<Activity>?>> = coroutineScope {
                 environments.map { environment ->
                     async {
-                        val data = runCatching {
+                        val data = runSuspendCatching {
                             client.activities.listPaginated(
                                 envId = environment.id,
                                 order = SortOrder.DESCENDING,
@@ -131,56 +138,83 @@ class ActivityCenterStore(private val scope: CoroutineScope) {
                 }
                 val normalized = data.map { normalize(it, environment) }
                 buckets[environment.id.rawValue] = sortActivities(normalized)
-                if (data.size >= limit) anyHasMore = true
+                if (data.size >= pageLimit) anyHasMore = true
             }
 
+            if (!isCurrentLoad(client, expectedClientGeneration, operationGeneration)) return
+            limit = pageLimit
+            environmentIds = environments.map { it.id.rawValue }
+            environmentNames.clear()
+            environments.forEach { environmentNames[it.id.rawValue] = it.name }
             activityBuckets.clear()
             activityBuckets.putAll(buckets)
             hasMore = anyHasMore
             rebuildActivities()
             if (failures > 0) {
                 streamErrorMessage = "Some environments could not load. Pull to refresh."
             }
+            if (streamingRequested) restartStreams()
         } catch (e: CancellationException) {
             throw e
         } catch (e: Throwable) {
-            if (activities.isEmpty()) errorMessage = friendlyErrorMessage(e)
+            if (isCurrentLoad(client, expectedClientGeneration, operationGeneration) && activities.isEmpty()) {
+                errorMessage = friendlyErrorMessage(e)
+            }
         } finally {
-            isLoading = false
+            if (isCurrentLoad(client, expectedClientGeneration, operationGeneration)) isLoading = false
+            if (loadJob === owningJob) loadJob = null
         }
     }
 
     suspend fun loadMore() {
         if (isLoadingMore || !hasMore) return
         isLoadingMore = true
+        val previousLimit = limit
+        val requestedLimit = previousLimit + PAGE_SIZE
         try {
-            limit += PAGE_SIZE
+            limit = requestedLimit
             load(reset = false)
+        } catch (e: CancellationException) {
+            if (limit == requestedLimit) limit = previousLimit
+            throw e
         } finally {
             isLoadingMore = false
         }
     }
 
     /** One coroutine per environment collecting `client.activities.stream(...)`. */
     fun startStream() {
+        streamingRequested = true
+        restartStreams()
+    }
+
+    private fun restartStreams() {
         val client = client ?: return
-        stopStream()
+        cancelStreamJobs()
         streamErrorMessage = null
 
         val environments = environmentIds.map { id ->
             ActivityEnvironment(EnvironmentId(id), environmentNames[id] ?: id)
         }
         if (environments.isEmpty()) return
 
+        val expectedClientGeneration = clientGeneration
+        val expectedStreamGeneration = streamGeneration
         isStreaming = true
         for (environment in environments) {
             streamJobs[environment.id.rawValue] = scope.launch {
-                consumeStream(client, environment)
+                consumeStream(client, environment, expectedClientGeneration, expectedStreamGeneration)
             }
         }
     }
 
     fun stopStream() {
+        streamingRequested = false
+        cancelStreamJobs()
+    }
+
+    private fun cancelStreamJobs() {
+        streamGeneration++
         streamJobs.values.forEach { it.cancel() }
         streamJobs.clear()
         isStreaming = false
@@ -215,7 +249,7 @@ class ActivityCenterStore(private val scope: CoroutineScope) {
         val results: List<Pair<Long?, Boolean>> = coroutineScope {
             targets.map { id ->
                 async {
-                    runCatching {
+                    runSuspendCatching {
                         client.activities.clearHistory(envId = EnvironmentId(id)).deleted
                     }.fold(
                         onSuccess = { it to true },
@@ -235,21 +269,46 @@ class ActivityCenterStore(private val scope: CoroutineScope) {
         return ClearHistoryResult(deleted = deleted, failed = failed)
     }
 
-    private suspend fun consumeStream(client: ArcaneClient, environment: ActivityEnvironment) {
+    private suspend fun consumeStream(
+        client: ArcaneClient,
+        environment: ActivityEnvironment,
+        expectedClientGeneration: Long,
+        expectedStreamGeneration: Long,
+    ) {
+        val owningJob = currentCoroutineContext()[Job]
         try {
             client.activities.stream(envId = environment.id, limit = PAGE_SIZE).collect { event ->
-                apply(event, environment)
+                if (isCurrentStream(client, expectedClientGeneration, expectedStreamGeneration)) {
+                    apply(event, environment)
+                }
             }
         } catch (e: CancellationException) {
             throw e
         } catch (e: Throwable) {
-            streamErrorMessage = "Live updates paused. Pull to refresh."
+            if (isCurrentStream(client, expectedClientGeneration, expectedStreamGeneration)) {
+                streamErrorMessage = "Live updates paused. Pull to refresh."
+            }
         } finally {
-            streamJobs.remove(environment.id.rawValue)
-            isStreaming = streamJobs.isNotEmpty()
+            val id = environment.id.rawValue
+            if (isCurrentStream(client, expectedClientGeneration, expectedStreamGeneration) &&
+                streamJobs[id] === owningJob
+            ) {
+                streamJobs.remove(id)
+                isStreaming = streamJobs.isNotEmpty()
+            }
         }
     }
 
+    private fun isCurrentLoad(client: ArcaneClient, clientGeneration: Long, loadGeneration: Long): Boolean =
+        this.client === client &&
+            this.clientGeneration == clientGeneration &&
+            this.loadGeneration == loadGeneration
+
+    private fun isCurrentStream(client: ArcaneClient, clientGeneration: Long, streamGeneration: Long): Boolean =
+        this.client === client &&
+            this.clientGeneration == clientGeneration &&
+            this.streamGeneration == streamGeneration
+
     private fun apply(event: ActivityStreamEvent, environment: ActivityEnvironment) {
         when (event.type) {
             ActivityStreamEventType.SNAPSHOT -> replaceSnapshot(event.activities, environment)

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

@@ -0,0 +1,19 @@
+package app.getarcane.android.core
+
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.ensureActive
+
+/** Result wrapper for suspending best-effort work that must never consume coroutine cancellation. */
+internal suspend inline fun <T> runSuspendCatching(
+    crossinline block: suspend () -> T,
+): Result<T> =
+    try {
+        val value = block()
+        currentCoroutineContext().ensureActive()
+        Result.success(value)
+    } catch (error: CancellationException) {
+        throw error
+    } catch (error: Throwable) {
+        Result.failure(error)
+    }

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

@@ -9,11 +9,13 @@ import app.getarcane.sdk.errors.ArcaneError
 import app.getarcane.sdk.models.environment.Environment
 import app.getarcane.sdk.streaming.ndjsonFlow
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.launch
@@ -77,6 +79,9 @@ class DashboardStreamStore(
     private var streamJob: Job? = null
     private var shouldRun = false
     private var generation = 0
+    private var nextSnapshotRequest = 0L
+    private val snapshotRequestsByEnvironmentId = HashMap<String, Long>()
+    private val snapshotJobsByEnvironmentId = HashMap<String, Job>()
 
     val aggregate: DashboardStreamAggregateCounts?
         get() {
@@ -120,6 +125,7 @@ class DashboardStreamStore(
                 )
             }
         }
+        snapshotRequestsByEnvironmentId.clear()
         streamUnsupported = false
         streamFailed = false
     }
@@ -139,6 +145,9 @@ class DashboardStreamStore(
     fun stop() {
         shouldRun = false
         generation += 1
+        snapshotRequestsByEnvironmentId.clear()
+        snapshotJobsByEnvironmentId.values.forEach { it.cancel() }
+        snapshotJobsByEnvironmentId.clear()
         streamJob?.cancel()
         streamJob = null
         connected = false
@@ -172,13 +181,14 @@ class DashboardStreamStore(
             }
         }
         val newIds = next.keys - statesByEnvironmentId.keys
+        val removedIds = statesByEnvironmentId.keys - targetIds
         statesByEnvironmentId = next
+        snapshotRequestsByEnvironmentId.keys.retainAll(targetIds)
+        removedIds.forEach { id -> snapshotJobsByEnvironmentId.remove(id)?.cancel() }
 
         if (streamJob != null) {
             val currentGeneration = generation
-            newIds.forEach { id ->
-                scope.launch { refreshEnvironment(id, currentGeneration) }
-            }
+            newIds.forEach { id -> launchSnapshotRefresh(id, currentGeneration) }
         }
     }
 
@@ -261,6 +271,12 @@ class DashboardStreamStore(
     }
 
     private fun applySnapshot(snapshot: DashboardSnapshot, environmentId: String) {
+        snapshotRequestsByEnvironmentId.remove(environmentId)
+        snapshotJobsByEnvironmentId.remove(environmentId)?.cancel()
+        applySnapshotState(snapshot, environmentId)
+    }
+
+    private fun applySnapshotState(snapshot: DashboardSnapshot, environmentId: String) {
         val state = statesByEnvironmentId[environmentId] ?: return
         statesByEnvironmentId = statesByEnvironmentId + (environmentId to state.copy(
             snapshot = snapshot,
@@ -273,6 +289,12 @@ class DashboardStreamStore(
     }
 
     private fun applyError(message: String?, code: DashboardStreamErrorCode?, environmentId: String) {
+        snapshotRequestsByEnvironmentId.remove(environmentId)
+        snapshotJobsByEnvironmentId.remove(environmentId)?.cancel()
+        applyErrorState(message, code, environmentId)
+    }
+
+    private fun applyErrorState(message: String?, code: DashboardStreamErrorCode?, environmentId: String) {
         val state = statesByEnvironmentId[environmentId] ?: return
         statesByEnvironmentId = statesByEnvironmentId + (environmentId to state.copy(
             loading = false,
@@ -290,20 +312,51 @@ class DashboardStreamStore(
 
     private suspend fun refreshEnvironment(environmentId: String, currentGeneration: Int) {
         val activeClient = client ?: return
+        val request = ++nextSnapshotRequest
+        snapshotRequestsByEnvironmentId[environmentId] = request
         try {
             val snapshot = activeClient.snapshot(environmentId)
-            if (currentGeneration == generation && statesByEnvironmentId.containsKey(environmentId)) {
-                applySnapshot(snapshot, environmentId)
+            if (isCurrentSnapshotRequest(activeClient, environmentId, currentGeneration, request)) {
+                snapshotRequestsByEnvironmentId.remove(environmentId)
+                applySnapshotState(snapshot, environmentId)
             }
         } catch (e: CancellationException) {
             throw e
         } catch (e: Throwable) {
-            if (currentGeneration == generation && statesByEnvironmentId.containsKey(environmentId)) {
-                applyError(friendlyErrorMessage(e), null, environmentId)
+            if (isCurrentSnapshotRequest(activeClient, environmentId, currentGeneration, request)) {
+                snapshotRequestsByEnvironmentId.remove(environmentId)
+                applyErrorState(friendlyErrorMessage(e), null, environmentId)
             }
         }
     }
 
+    private fun launchSnapshotRefresh(environmentId: String, currentGeneration: Int) {
+        snapshotJobsByEnvironmentId.remove(environmentId)?.cancel()
+        val job = scope.launch(start = CoroutineStart.LAZY) {
+            val owningJob = currentCoroutineContext()[Job]
+            try {
+                refreshEnvironment(environmentId, currentGeneration)
+            } finally {
+                if (snapshotJobsByEnvironmentId[environmentId] === owningJob) {
+                    snapshotJobsByEnvironmentId.remove(environmentId)
+                }
+            }
+        }
+        snapshotJobsByEnvironmentId[environmentId] = job
+        job.start()
+    }
+
+    private fun isCurrentSnapshotRequest(
+        expectedClient: DashboardStreamClient,
+        environmentId: String,
+        expectedGeneration: Int,
+        request: Long,
+    ): Boolean =
+        client === expectedClient &&
+            generation == expectedGeneration &&
+            snapshotRequestsByEnvironmentId[environmentId] == request &&
+            statesByEnvironmentId.containsKey(environmentId)
+
     private companion object {
         const val MAX_RECONNECT_ATTEMPTS = 20
         const val MAX_RECONNECT_DELAY_MILLIS = 15_000L