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) {