fix: stop dropping streamed chat events and cross-session poll writes

app/src/main/java/com/opencode/android/core/api/OpenCodeApiClient.kt

@@ -7,13 +7,15 @@ import com.google.gson.JsonObject
 import com.google.gson.reflect.TypeToken
 import com.opencode.android.core.security.OpenCodeUrl
 import com.opencode.android.data.connection.ConnectionProfile
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.retryWhen
 import kotlinx.coroutines.withContext
 import okhttp3.Credentials
@@ -396,12 +398,13 @@ class OpenCodeApiClient(
         true
     }
 
-    private fun singleEventStream(): Flow<OpenCodeEvent> = callbackFlow {
+    private fun singleEventStream(): Flow<OpenCodeEvent> = channelFlow {
         val eventClient = httpClient.newBuilder()
             .readTimeout(0, TimeUnit.MILLISECONDS)
             .build()
         val request = requestBuilder("event")
             .header("Accept", "text/event-stream")
+            .header("Cache-Control", "no-cache")
             .get()
             .build()
         val call = eventClient.newCall(request)
@@ -422,7 +425,9 @@ class OpenCodeApiClient(
                             when {
                                 line.isEmpty() -> {
                                     if (data.isNotEmpty()) {
-                                        trySend(eventParser.parse(data.toString()))
+                                        // Suspending send (never trySend): a dropped frame loses a
+                                        // chunk of the assistant's reply for good.
+                                        send(eventParser.parse(data.toString()))
                                         data.setLength(0)
                                     }
                                 }
@@ -432,18 +437,22 @@ class OpenCodeApiClient(
                                 }
                             }
                         }
+                        // A stream that ends mid-frame still carries a usable event.
+                        if (data.isNotEmpty()) send(eventParser.parse(data.toString()))
                     }
-                    if (isActive) throw IOException("OpenCode event stream closed")
+                    throw IOException("OpenCode event stream closed")
                 }
+            } catch (cancellation: CancellationException) {
+                throw cancellation
             } catch (error: Throwable) {
-                if (isActive) close(error)
+                close(error)
             }
         }
         awaitClose {
             call.cancel()
             readerJob.cancel()
         }
-    }
+    }.buffer(EVENT_BUFFER_CAPACITY)
 
     private suspend fun <T> get(
         path: String,
@@ -594,6 +603,9 @@ class OpenCodeApiClient(
     companion object {
         private val JSON_MEDIA_TYPE = "application/json; charset=utf-8".toMediaType()
         private const val MAX_ERROR_BODY_CHARS = 240
+
+        /** Deep enough that a burst of stream frames is queued rather than dropped. */
+        private const val EVENT_BUFFER_CAPACITY = 512
         private const val PROVIDER_AUTH_TIMEOUT_MINUTES = 6L
 
         fun defaultHttpClient(): OkHttpClient = OkHttpClient.Builder()

app/src/main/java/com/opencode/android/data/repository/RuntimeActivityRepository.kt

@@ -4,7 +4,10 @@ import com.opencode.android.core.api.OpenCodeEvent
 import com.opencode.android.core.api.PermissionRequest
 import com.opencode.android.runtime.RuntimeRegistry
 import com.opencode.android.runtime.RuntimeState
+import com.opencode.android.runtime.RuntimeTarget
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -60,42 +63,59 @@ class RuntimeActivityRepository(
                 mutableState.value = RuntimeActivityState()
                 if (target == null) return@selected
 
-                target.state.collectLatest state@ { runtimeState ->
-                    mutableState.update {
-                        it.copy(
-                            activeSessionIds = emptySet(),
-                            completedSessionIds = emptySet(),
-                            permissions = emptyList(),
-                            streamError = null
-                        )
-                    }
-                    if (runtimeState !is RuntimeState.Connected) return@state
-
-                    flow { emitAll(target.events()) }
-                        .retryWhen { error, attempt ->
-                            mutableState.update {
-                                it.copy(
-                                    streamError = error.message
-                                        ?: "OpenCodeイベント接続に失敗しました"
-                                )
+                // The event stream is kept alive across transient runtime-state churn (e.g. a
+                // health recheck that briefly reports Connecting). Restarting it on every such
+                // blip used to drop whole replies on the floor.
+                coroutineScope {
+                    var streamJob: Job? = null
+                    target.state.collect { runtimeState ->
+                        when (runtimeState) {
+                            is RuntimeState.Connected -> {
+                                if (streamJob?.isActive != true) {
+                                    streamJob = launch { streamEvents(target) }
+                                }
                             }
-                            if (retryDelayMillis > 0L) {
-                                val backoff = (retryDelayMillis * (1L shl attempt.toInt().coerceAtMost(4)))
-                                    .coerceAtMost(maxRetryDelayMillis)
-                                delay(backoff)
+                            RuntimeState.Connecting -> Unit
+                            else -> {
+                                streamJob?.cancel()
+                                streamJob = null
+                                mutableState.update {
+                                    it.copy(
+                                        activeSessionIds = emptySet(),
+                                        completedSessionIds = emptySet(),
+                                        permissions = emptyList()
+                                    )
+                                }
                             }
-                            true
-                        }
-                        .collect { event ->
-                            mutableState.update { it.copy(streamError = null) }
-                            mutableEvents.emit(event)
-                            handle(event)
                         }
+                    }
                 }
             }
         }
     }
 
+    private suspend fun streamEvents(target: RuntimeTarget) {
+        flow { emitAll(target.events()) }
+            .retryWhen { error, attempt ->
+                mutableState.update {
+                    it.copy(
+                        streamError = error.message ?: "OpenCodeイベント接続に失敗しました"
+                    )
+                }
+                if (retryDelayMillis > 0L) {
+                    val backoff = (retryDelayMillis * (1L shl attempt.toInt().coerceAtMost(4)))
+                        .coerceAtMost(maxRetryDelayMillis)
+                    delay(backoff)
+                }
+                true
+            }
+            .collect { event ->
+                mutableState.update { it.copy(streamError = null) }
+                mutableEvents.emit(event)
+                handle(event)
+            }
+    }
+
     fun resolvePermission(permissionId: String) {
         mutableState.update { current ->
             current.copy(permissions = current.permissions.filterNot { it.id == permissionId })

app/src/main/java/com/opencode/android/feature/chat/ChatViewModel.kt

@@ -323,6 +323,10 @@ class ChatViewModel(
 
     fun openSession(sessionId: String, title: String = "") {
         val currentBackend = backend ?: return
+        // Switching away from whatever chat was previously loaded here must drop its running
+        // state too — otherwise the composer opens the new chat already stuck on the stop button
+        // because the old chat happened to be mid-turn when the user navigated away.
+        val switchingSession = _uiState.value.sessionId != sessionId
         streamedParts.clear()
         _uiState.update {
             it.copy(
@@ -332,6 +336,8 @@ class ChatViewModel(
                 messages = emptyList(),
                 permissions = emptyList(),
                 pendingQuestions = emptyList(),
+                isRunning = if (switchingSession) false else it.isRunning,
+                isThinking = if (switchingSession) false else it.isThinking,
                 error = null
             )
         }
@@ -416,6 +422,9 @@ class ChatViewModel(
         }
 
         viewModelScope.launch {
+            // Captured once the target session is known so onFailure below can tell whether the
+            // failure still concerns the chat currently on screen.
+            var capturedSessionId: String? = null
             runCatching {
                 val existingSessionId = _uiState.value.sessionId
                 val session = if (existingSessionId == null) {
@@ -427,6 +436,7 @@ class ChatViewModel(
                     null
                 }
                 val targetSessionId = existingSessionId ?: requireNotNull(session).id
+                capturedSessionId = targetSessionId
                 if (session != null) {
                     _uiState.update {
                         it.copy(sessionId = session.id, sessionTitle = session.title)
@@ -457,11 +467,17 @@ class ChatViewModel(
                 _uiState.update { it.copy(attachments = emptyList(), imagePreviews = emptyList()) }
                 clearDraft(targetSessionId)
                 var sessionCompleted = false
+                // Every update below is guarded on the chat still being on screen: if the user
+                // switches to another session, this poll must not keep overwriting its transcript
+                // or clearing its running state with data that belongs to the old one.
+                fun isStillActive() = _uiState.value.sessionId == targetSessionId
                 val pollFinished = withTimeoutOrNull(RESPONSE_POLL_TIMEOUT_MS) {
-                    while (_uiState.value.isRunning) {
+                    while (isStillActive() && _uiState.value.isRunning) {
                         kotlinx.coroutines.delay(RESPONSE_POLL_INTERVAL_MS)
+                        if (!isStillActive()) return@withTimeoutOrNull
                         runCatching { currentBackend.listMessages(targetSessionId) }
                             .onSuccess { serverMessages ->
+                                if (!isStillActive()) return@onSuccess
                                 val previewsById = _uiState.value.messages
                                     .associate { it.id to it.imagePreviews }
                                 val uiMessages = serverMessages.mapNotNull(::toUiMessage).map { message ->
@@ -475,20 +491,24 @@ class ChatViewModel(
                                     _uiState.update { it.copy(messages = uiMessages) }
                                 }
                             }
+                        if (!isStillActive()) return@withTimeoutOrNull
                         runCatching { currentBackend.session(targetSessionId) }
                             .onSuccess { sessionInfo ->
                                 if (sessionInfo.time.completed != null) {
                                     sessionCompleted = true
-                                    _uiState.update { it.copy(isRunning = false, isThinking = false) }
+                                    if (isStillActive()) {
+                                        _uiState.update { it.copy(isRunning = false, isThinking = false) }
+                                    }
                                 }
                             }
                     }
                 }
                 // Some runtimes deliver the final message but drop session.idle. Do not
                 // leave the UI in the running state when the bounded fallback poll ends.
-                if (sessionCompleted || pollFinished == null) {
+                if (isStillActive() && (sessionCompleted || pollFinished == null)) {
                     runCatching { currentBackend.listMessages(targetSessionId) }
                         .onSuccess { serverMessages ->
+                            if (!isStillActive()) return@onSuccess
                             val hasResponse = serverMessages.any { message ->
                                 message.info.role == "assistant" && message.info.id !in messageIdsBeforeSend
                             }
@@ -507,11 +527,13 @@ class ChatViewModel(
                         }
                 }
             }.onFailure { error ->
-                _uiState.update {
-                    it.copy(
-                        isRunning = false,
-                        isThinking = false
-                    )
+                if (capturedSessionId == null || _uiState.value.sessionId == capturedSessionId) {
+                    _uiState.update {
+                        it.copy(
+                            isRunning = false,
+                            isThinking = false
+                        )
+                    }
                 }
                 reportError(error.safeMessage())
             }
@@ -697,11 +719,13 @@ class ChatViewModel(
             }
             is OpenCodeEvent.MessagePartDelta -> {
                 if (event.sessionId != activeSession || event.field != "text") return
-                val messageParts = streamedParts[event.messageId] ?: return
-                val existing = messageParts[event.partId] ?: return
-                val updatedPart = when (existing) {
+                val messageParts = streamedParts.getOrPut(event.messageId) { linkedMapOf() }
+                val updatedPart = when (val existing = messageParts[event.partId]) {
                     is ChatPart.Text -> existing.copy(text = existing.text + event.delta)
                     is ChatPart.Reasoning -> existing.copy(text = existing.text + event.delta)
+                    // A delta can outrun the part event that introduces it; start the part here
+                    // rather than dropping the text on the floor.
+                    null -> ChatPart.Text(event.partId, event.delta)
                     else -> return
                 }
                 messageParts[event.partId] = updatedPart

app/src/main/java/com/opencode/android/runtime/local/LocalRuntimeTarget.kt

@@ -71,7 +71,11 @@ class LocalRuntimeTarget(
             return Result.failure(IllegalStateException(state.describe()))
         }
 
-        mutableState.value = RuntimeState.Connecting
+        // Keep an already connected runtime connected while re-checking health, otherwise every
+        // catalog refresh restarts the event stream and drops in-flight replies.
+        if (mutableState.value !is RuntimeState.Connected) {
+            mutableState.value = RuntimeState.Connecting
+        }
         return runCatching { backend.health() }
             .onSuccess { health ->
                 mutableState.value = if (health.healthy) {

app/src/main/java/com/opencode/android/runtime/remote/RemoteRuntimeTarget.kt

@@ -44,7 +44,11 @@ class RemoteRuntimeTarget(
     override val state: StateFlow<RuntimeState> = mutableState.asStateFlow()
 
     override suspend fun connect(): Result<OpenCodeHealth> {
-        mutableState.value = RuntimeState.Connecting
+        // Re-checking an already connected runtime must not flap the state, or every refresh would
+        // tear down the live event stream that hangs off it.
+        if (mutableState.value !is RuntimeState.Connected) {
+            mutableState.value = RuntimeState.Connecting
+        }
         return runCatching { backend.health() }
             .onSuccess { health ->
                 mutableState.value = if (health.healthy) {

app/src/test/java/com/opencode/android/data/repository/RuntimeActivityRepositoryTest.kt

@@ -88,6 +88,54 @@ class RuntimeActivityRepositoryTest {
         assertEquals("イベント接続", repository.state.value.logs.single().title)
     }
 
+    @Test
+    fun `a brief Connecting blip does not restart the event stream`() = runTest {
+        val dispatcher = StandardTestDispatcher(testScheduler)
+        val target = FakeTarget()
+        val registry = RuntimeRegistry(
+            store = FakeStore(selectedRuntimeId = target.id),
+            localTarget = target,
+            remoteFactory = { error("unused") }
+        )
+        RuntimeActivityRepository(registry, TestScope(dispatcher))
+
+        target.state.value = RuntimeState.Connected("1.18.3")
+        advanceUntilIdle()
+        assertEquals(1, target.eventCalls)
+
+        // A health recheck on an already-connected runtime dips back to Connecting and returns;
+        // the existing SSE connection must survive it rather than being torn down and reopened.
+        target.state.value = RuntimeState.Connecting
+        advanceUntilIdle()
+        target.state.value = RuntimeState.Connected("1.18.3")
+        advanceUntilIdle()
+
+        assertEquals(1, target.eventCalls)
+    }
+
+    @Test
+    fun `going disconnected then reconnecting does reopen the stream`() = runTest {
+        val dispatcher = StandardTestDispatcher(testScheduler)
+        val target = FakeTarget()
+        val registry = RuntimeRegistry(
+            store = FakeStore(selectedRuntimeId = target.id),
+            localTarget = target,
+            remoteFactory = { error("unused") }
+        )
+        RuntimeActivityRepository(registry, TestScope(dispatcher))
+
+        target.state.value = RuntimeState.Connected("1.18.3")
+        advanceUntilIdle()
+        assertEquals(1, target.eventCalls)
+
+        target.state.value = RuntimeState.Disconnected
+        advanceUntilIdle()
+        target.state.value = RuntimeState.Connected("1.18.3")
+        advanceUntilIdle()
+
+        assertEquals(2, target.eventCalls)
+    }
+
     private class FakeStore(
         override var selectedRuntimeId: String?
     ) : RuntimeConnectionStore {