From 45c0d99c0b070b86bb91bf7e6828a4e54304deca Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 24 Jul 2026 19:18:16 +0000 Subject: [PATCH] fix: stop dropping streamed chat events and cross-session poll writes Four independent defects behind "replies arrive late / not at all / the stop button sticks": - The SSE reader pushed parsed frames with trySend, which fails silently once the channel is full. During a fast reply the collector falls behind and whole chunks of the answer were discarded. It now suspends on a buffered channel, and flushes a frame left over when the stream ends mid-message. - connect() dropped an already-connected runtime back to Connecting on every health recheck, and the activity repository restarted the SSE stream on any runtime-state emission. Each catalog refresh therefore tore down the live stream mid-reply. Health rechecks now keep a connected runtime connected, and the stream survives transient churn, stopping only when the runtime actually goes away. - The send-side response poll wrote into whatever chat happened to be on screen. Switching sessions while a turn was in flight let the old session's poll overwrite the new session's transcript and clear its running state. Every write is now guarded on the target session still being the open one. - openSession kept the previous chat's running state, so opening another chat mid-turn showed the composer already stuck on the stop button. Also: a text delta arriving before the part event that introduces it was dropped; it now starts the part instead. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01RLvEkDwiWT8R4S23gdPsSB --- .../android/core/api/OpenCodeApiClient.kt | 24 ++++-- .../repository/RuntimeActivityRepository.kt | 76 ++++++++++++------- .../android/feature/chat/ChatViewModel.kt | 46 ++++++++--- .../runtime/local/LocalRuntimeTarget.kt | 6 +- .../runtime/remote/RemoteRuntimeTarget.kt | 6 +- .../RuntimeActivityRepositoryTest.kt | 48 ++++++++++++ .../android/feature/chat/ChatViewModelTest.kt | 68 ++++++++++++++++- 7 files changed, 226 insertions(+), 48 deletions(-) diff --git a/app/src/main/java/com/opencode/android/core/api/OpenCodeApiClient.kt b/app/src/main/java/com/opencode/android/core/api/OpenCodeApiClient.kt index 8589bc51..19ea583c 100644 --- a/app/src/main/java/com/opencode/android/core/api/OpenCodeApiClient.kt +++ b/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 = callbackFlow { + private fun singleEventStream(): Flow = 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 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() diff --git a/app/src/main/java/com/opencode/android/data/repository/RuntimeActivityRepository.kt b/app/src/main/java/com/opencode/android/data/repository/RuntimeActivityRepository.kt index 76ab43e9..bc94c0a8 100644 --- a/app/src/main/java/com/opencode/android/data/repository/RuntimeActivityRepository.kt +++ b/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 }) diff --git a/app/src/main/java/com/opencode/android/feature/chat/ChatViewModel.kt b/app/src/main/java/com/opencode/android/feature/chat/ChatViewModel.kt index 2a771d8e..5694deae 100644 --- a/app/src/main/java/com/opencode/android/feature/chat/ChatViewModel.kt +++ b/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 diff --git a/app/src/main/java/com/opencode/android/runtime/local/LocalRuntimeTarget.kt b/app/src/main/java/com/opencode/android/runtime/local/LocalRuntimeTarget.kt index af31c170..ca1c5b9f 100644 --- a/app/src/main/java/com/opencode/android/runtime/local/LocalRuntimeTarget.kt +++ b/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) { diff --git a/app/src/main/java/com/opencode/android/runtime/remote/RemoteRuntimeTarget.kt b/app/src/main/java/com/opencode/android/runtime/remote/RemoteRuntimeTarget.kt index 5992ebfd..e097903e 100644 --- a/app/src/main/java/com/opencode/android/runtime/remote/RemoteRuntimeTarget.kt +++ b/app/src/main/java/com/opencode/android/runtime/remote/RemoteRuntimeTarget.kt @@ -44,7 +44,11 @@ class RemoteRuntimeTarget( override val state: StateFlow = mutableState.asStateFlow() override suspend fun connect(): Result { - 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) { diff --git a/app/src/test/java/com/opencode/android/data/repository/RuntimeActivityRepositoryTest.kt b/app/src/test/java/com/opencode/android/data/repository/RuntimeActivityRepositoryTest.kt index 83af2b80..7ab5b8bc 100644 --- a/app/src/test/java/com/opencode/android/data/repository/RuntimeActivityRepositoryTest.kt +++ b/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 { diff --git a/app/src/test/java/com/opencode/android/feature/chat/ChatViewModelTest.kt b/app/src/test/java/com/opencode/android/feature/chat/ChatViewModelTest.kt index 5a93f7b2..c539c55c 100644 --- a/app/src/test/java/com/opencode/android/feature/chat/ChatViewModelTest.kt +++ b/app/src/test/java/com/opencode/android/feature/chat/ChatViewModelTest.kt @@ -22,6 +22,7 @@ import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.test.StandardTestDispatcher import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.resetMain +import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.setMain import org.junit.After @@ -496,6 +497,69 @@ class ChatViewModelTest { assertFalse(viewModel.uiState.value.isRunning) } + @Test + fun `switching sessions mid-poll does not corrupt the newly opened session`() = + runTest(dispatcher) { + val backend = FakeBackend() + val viewModel = ChatViewModel(backend) + + viewModel.sendMessage("Hello") + // Advance only up to the poll loop's first delay() — session s1's background poll is + // now parked mid-flight, about to fetch s1's messages once time moves forward. + runCurrent() + + backend.historyMessagesBySession = mapOf( + "s1" to listOf(sessionAssistantMessage("s1", "m-s1", "Stale reply for the old chat")), + "s2" to listOf(sessionAssistantMessage("s2", "m-s2", "Reply for the other chat")) + ) + viewModel.openSession("s2", "Other chat") + runCurrent() + + // Let s1's stale poll loop run its course; its updates must all be no-ops now that the + // screen has moved on to s2. + advanceUntilIdle() + + assertEquals("s2", viewModel.uiState.value.sessionId) + assertEquals( + listOf("Reply for the other chat"), + viewModel.uiState.value.messages.map { it.text } + ) + } + + @Test + fun `opening a different session while the previous one is running clears the stop button`() = + runTest(dispatcher) { + val backend = FakeBackend() + val viewModel = ChatViewModel(backend) + + viewModel.sendMessage("Long task") + advanceUntilIdle() + assertTrue(viewModel.uiState.value.isRunning) + + viewModel.openSession("s2", "Another chat") + + assertFalse(viewModel.uiState.value.isRunning) + } + + private fun sessionAssistantMessage(sessionId: String, messageId: String, text: String) = + OpenCodeMessage( + info = OpenCodeMessageInfo( + id = messageId, + sessionId = sessionId, + role = "assistant", + time = OpenCodeTime(created = 1) + ), + parts = listOf( + OpenCodePart( + id = "$messageId-p", + sessionId = sessionId, + messageId = messageId, + type = "text", + text = text + ) + ) + ) + private class FakeBackend : OpenCodeBackend { override val id: String = "fake" override val displayName: String = "Fake" @@ -504,6 +568,7 @@ class ChatViewModelTest { var createSessionCalls = 0 var lastCreateDirectory: String? = null var historyMessages: List = emptyList() + var historyMessagesBySession: Map> = emptyMap() val sentPrompts = mutableListOf>() val permissionResponses = mutableListOf() val abortedSessions = mutableListOf() @@ -520,7 +585,8 @@ class ChatViewModelTest { time = OpenCodeTime(created = 1) ) } - override suspend fun listMessages(sessionId: String): List = historyMessages + override suspend fun listMessages(sessionId: String): List = + historyMessagesBySession[sessionId] ?: historyMessages override suspend fun listProviders(): ProviderCatalog = ProviderCatalog() override suspend fun listAgents(): List = emptyList() override suspend fun sendMessage(sessionId: String, request: PromptRequest) {