git.iko.soy/opencode-android
on a40d3a2 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/ui/chat/ChatViewModel.kt
518 lines 24 kB history raw
1package soy.iko.opencode.ui.chat
3import androidx.lifecycle.ViewModel
4import androidx.lifecycle.viewModelScope
5import soy.iko.opencode.data.model.Agent
6import soy.iko.opencode.data.model.Command
7import soy.iko.opencode.data.model.MessageWithParts
8import soy.iko.opencode.data.model.ModelOption
9import soy.iko.opencode.data.model.Permission
10import soy.iko.opencode.data.model.PermissionReplied
11import soy.iko.opencode.data.model.PermissionResponse
12import soy.iko.opencode.data.model.PermissionUpdated
13import soy.iko.opencode.data.model.SessionDeleted
14import soy.iko.opencode.data.model.SessionUpdated
15import soy.iko.opencode.data.model.TextPart
16import soy.iko.opencode.data.model.defaultOption
17import soy.iko.opencode.data.model.toOptions
18import soy.iko.opencode.data.network.EventStreamClient
19import soy.iko.opencode.data.network.NetworkConfig
20import soy.iko.opencode.data.repo.SessionRepository
21import soy.iko.opencode.di.AppContainer
22import soy.iko.opencode.R
23import soy.iko.opencode.util.runCatchingCancellable
24import soy.iko.opencode.util.safeExceptionSummary
25import android.util.Log
26import kotlinx.coroutines.ExperimentalCoroutinesApi
27import kotlinx.coroutines.channels.BufferOverflow
28import kotlinx.coroutines.flow.MutableSharedFlow
29import kotlinx.coroutines.flow.MutableStateFlow
30import kotlinx.coroutines.flow.SharedFlow
31import kotlinx.coroutines.flow.SharingStarted
32import kotlinx.coroutines.flow.StateFlow
33import kotlinx.coroutines.flow.asSharedFlow
34import kotlinx.coroutines.flow.asStateFlow
35import kotlinx.coroutines.CancellationException
36import kotlinx.coroutines.flow.collectLatest
37import kotlinx.coroutines.flow.debounce
38import kotlinx.coroutines.flow.drop
39import kotlinx.coroutines.flow.first
40import kotlinx.coroutines.flow.filterIsInstance
41import kotlinx.coroutines.flow.flatMapLatest
42import kotlinx.coroutines.flow.flowOf
43import kotlinx.coroutines.flow.map
44import kotlinx.coroutines.flow.merge
45import kotlinx.coroutines.flow.onEach
46import kotlinx.coroutines.flow.retryWhen
47import kotlinx.coroutines.flow.stateIn
48import kotlinx.coroutines.delay
49import kotlinx.coroutines.launch
51@OptIn(ExperimentalCoroutinesApi::class, kotlinx.coroutines.FlowPreview::class)
52class ChatViewModel(
53 private val container: AppContainer,
54 private val sessionId: String,
55) : ViewModel() {
57 private val connection get() = container.activeConnection.value
59 val connected: Boolean get() = connection != null
61 private val _loading = MutableStateFlow(true)
62 val loading: StateFlow<Boolean> = _loading.asStateFlow()
64 /** Separate from [loading]: tracks the manual reconnect() flow so its spinner
65 * doesn't conflict with the messages flow's loading state. */
66 private val _reconnecting = MutableStateFlow(false)
67 val reconnecting: StateFlow<Boolean> = _reconnecting.asStateFlow()
69 /** When true, the debounced draft collector skips persisting an empty draft so
70 * [send]'s deliberate non-persistence of the cleared draft isn't undone by the
71 * debounce timer firing before the send completes. */
72 private val suppressDraftPersist = java.util.concurrent.atomic.AtomicBoolean(false)
74 /** Per-catalog reload triggers: incrementing one causes [observeCatalog]'s
75 * collectLatest to cancel any in-flight fetch and start a fresh one, so a
76 * manual reload supersedes a stale observeCatalog fetch. */
77 private val _modelsReload = MutableStateFlow(0)
78 private val _agentsReload = MutableStateFlow(0)
79 private val _commandsReload = MutableStateFlow(0)
81 val messages: StateFlow<List<MessageWithParts>> =
82 container.activeConnection
83 .flatMapLatest { conn ->
84 conn?.repository?.observeMessages(sessionId) ?: flowOf(emptyList())
85 }
86 .onEach {
87 _loading.value = false
88 }
89 .retryWhen { cause, _ ->
90 _loading.value = false
91 _errorEvents.tryEmit(container.friendlyError(cause))
92 // Delay before retrying to avoid a tight loop on persistent errors.
93 delay(NetworkConfig.retryInitialDelayMs)
94 true
95 }
96 .stateIn(
97 scope = viewModelScope,
98 started = SharingStarted.WhileSubscribed(NetworkConfig.stateFlowSubscriptionTimeoutMs),
99 initialValue = emptyList(),
100 )
102 /** Whether the conversation has any messages. Derived separately so the top bar
103 * (share button enabled state) can observe this cheap boolean instead of the
104 * full messages list, avoiding per-token recomposition of the app bar during
105 * streaming. */
106 val hasMessages: StateFlow<Boolean> = messages
107 .map { it.isNotEmpty() }
108 .stateIn(
109 scope = viewModelScope,
110 started = SharingStarted.WhileSubscribed(NetworkConfig.stateFlowSubscriptionTimeoutMs),
111 initialValue = false,
112 )
114 private val _running = MutableStateFlow(false)
115 val running: StateFlow<Boolean> = _running.asStateFlow()
117 /** One-shot error events surfaced as snackbars. A SharedFlow (not StateFlow) so each
118 * emission is delivered independently — two rapid failures both get a snackbar
119 * instead of the second silently overwriting the first. */
120 private val _errorEvents = MutableSharedFlow<String>(
121 extraBufferCapacity = NetworkConfig.snackbarEventBufferCapacity,
122 onBufferOverflow = BufferOverflow.DROP_OLDEST,
123 )
124 val errorEvents: SharedFlow<String> = _errorEvents.asSharedFlow()
126 private val _models = MutableStateFlow<List<ModelOption>>(emptyList())
127 val models: StateFlow<List<ModelOption>> = _models.asStateFlow()
129 private val _modelsLoading = MutableStateFlow(true)
130 val modelsLoading: StateFlow<Boolean> = _modelsLoading.asStateFlow()
132 private val _modelsError = MutableStateFlow(false)
133 val modelsError: StateFlow<Boolean> = _modelsError.asStateFlow()
135 private val _selectedModel = MutableStateFlow<ModelOption?>(null)
136 val selectedModel: StateFlow<ModelOption?> = _selectedModel.asStateFlow()
138 val connectionState: StateFlow<EventStreamClient.ConnectionState> =
139 container.activeConnection
140 .flatMapLatest { it?.events?.state ?: flowOf(EventStreamClient.ConnectionState.Disconnected) }
141 .stateIn(
142 scope = viewModelScope,
143 started = SharingStarted.WhileSubscribed(NetworkConfig.stateFlowSubscriptionTimeoutMs),
144 initialValue = EventStreamClient.ConnectionState.Disconnected,
145 )
147 private val _pendingPermission = MutableStateFlow<Permission?>(null)
148 val pendingPermission: StateFlow<Permission?> = _pendingPermission.asStateFlow()
150 private val _agents = MutableStateFlow<List<Agent>>(emptyList())
151 val agents: StateFlow<List<Agent>> = _agents.asStateFlow()
153 private val _agentsLoading = MutableStateFlow(true)
154 val agentsLoading: StateFlow<Boolean> = _agentsLoading.asStateFlow()
156 private val _agentsError = MutableStateFlow(false)
157 val agentsError: StateFlow<Boolean> = _agentsError.asStateFlow()
159 private val _selectedAgent = MutableStateFlow<String?>(null)
160 val selectedAgent: StateFlow<String?> = _selectedAgent.asStateFlow()
162 private val _commands = MutableStateFlow<List<Command>>(emptyList())
163 val commands: StateFlow<List<Command>> = _commands.asStateFlow()
165 private val _commandsLoading = MutableStateFlow(true)
166 val commandsLoading: StateFlow<Boolean> = _commandsLoading.asStateFlow()
168 private val _commandsError = MutableStateFlow(false)
169 val commandsError: StateFlow<Boolean> = _commandsError.asStateFlow()
171 private val _sessionTitle = MutableStateFlow<String?>(null)
172 val sessionTitle: StateFlow<String?> = _sessionTitle.asStateFlow()
174 /** Set when the current session is deleted via SSE, so the UI can navigate away. */
175 private val _sessionDeleted = MutableStateFlow(false)
176 val sessionDeleted: StateFlow<Boolean> = _sessionDeleted.asStateFlow()
178 /** The text of the last send that failed, surfaced so the UI can offer a retry. */
179 private val _failedDraft = MutableStateFlow<String?>(null)
180 val failedDraft: StateFlow<String?> = _failedDraft.asStateFlow()
182 /** Per-session draft, persisted so it survives navigation/process death. */
183 private val _draft = MutableStateFlow(container.draftStore.get(sessionId))
184 val draft: StateFlow<String> = _draft.asStateFlow()
186 init {
187 // DraftStore loads SharedPreferences asynchronously; on a cold start the
188 // synchronous get() above returns "" until the background load completes.
189 // Observe the ready signal and re-seed the draft so it appears in the UI.
190 viewModelScope.launch {
191 container.draftStore.ready.collect { ready ->
192 if (ready && _draft.value.isEmpty()) {
193 _draft.value = container.draftStore.get(sessionId)
194 }
195 }
196 }
197 }
199 fun selectModel(option: ModelOption) { _selectedModel.value = option }
201 fun selectAgent(name: String?) { _selectedAgent.value = name }
203 fun reloadModels() {
204 val conn = connection ?: return
205 viewModelScope.launch {
206 conn.api.invalidateProvidersCache()
207 _modelsReload.value++
208 }
209 }
211 fun reloadAgents() {
212 val conn = connection ?: return
213 viewModelScope.launch {
214 conn.api.invalidateAgentsCache()
215 _agentsReload.value++
216 }
217 }
219 fun reloadCommands() {
220 val conn = connection ?: return
221 viewModelScope.launch {
222 conn.api.invalidateCommandsCache()
223 _commandsReload.value++
224 }
225 }
227 /**
228 * Shared helper for init-block catalog observers: re-fetches whenever the active
229 * connection or the catalog's reload trigger changes, invoking [onNull] to clear
230 * state on null so stale data from the old server doesn't persist. The
231 * [reloadTrigger] is merged with [activeConnection] so a manual reload (via
232 * reloadModels/reloadAgents/reloadCommands) cancels any in-flight observe fetch
233 * via collectLatest, preventing a stale observe result from overwriting a fresh
234 * reload result. Failure is non-fatal — the error flag is surfaced to the UI.
235 */
236 private fun <T> observeCatalog(
237 tag: String,
238 loading: MutableStateFlow<Boolean>,
239 error: MutableStateFlow<Boolean>,
240 fetch: suspend (soy.iko.opencode.data.network.OpencodeApiClient) -> T,
241 onSuccess: (T) -> Unit,
242 onNull: () -> Unit,
243 reloadTrigger: StateFlow<Int>,
244 ) {
245 viewModelScope.launch {
246 merge(container.activeConnection, reloadTrigger).collectLatest { _ ->
247 val conn = container.activeConnection.value
248 if (conn == null) { onNull(); loading.value = false; error.value = false; return@collectLatest }
249 loading.value = true
250 error.value = false
251 runCatchingCancellable { fetch(conn.api) }
252 // fetch() does HTTP via withRetry, so the failure can be a
253 // ClientRequestException whose message embeds the full request URL
254 // (may contain auth/paths). Log only a scrubbed summary.
255 .onFailure { Log.w("ChatViewModel", "Failed to load $tag: ${safeExceptionSummary(it)}"); error.value = true }
256 .getOrNull()?.let(onSuccess)
257 loading.value = false
258 }
259 }
260 }
262 fun updateDraft(text: String) {
263 _draft.value = text
264 }
266 init {
267 // Debounce draft persistence so we don't write to disk on every keystroke.
268 // Skip persisting empty drafts that were set by send() (suppressed via
269 // suppressDraftPersist) — send() clears the in-memory draft immediately for
270 // UI feedback but deliberately doesn't persist the clear until the send
271 // succeeds, so a failed send can restore the draft.
272 viewModelScope.launch {
273 _draft.drop(1).debounce(NetworkConfig.draftDebounceMs).collect { text ->
274 if (text.isBlank() && suppressDraftPersist.get()) return@collect
275 runCatchingCancellable { container.draftStore.set(sessionId, text) }
276 .onFailure { Log.w("ChatViewModel", "Failed to persist draft", it) }
277 }
278 }
279 // Reset per-connection state when the active server changes so stale spinners,
280 // permission dialogs, errors, and agent selections from the old server don't
281 // persist into the new one.
282 viewModelScope.launch {
283 container.activeConnection.collectLatest { conn ->
284 if (conn == null) return@collectLatest
285 _running.value = false
286 _pendingPermission.value = null
287 _failedDraft.value = null
288 _selectedAgent.value = null
289 _sessionTitle.value = null
290 try {
291 conn.events.events.collect { event ->
292 if (SessionRepository.isIdle(event, sessionId)) _running.value = false
293 if (SessionRepository.isError(event, sessionId)) {
294 _running.value = false
295 _errorEvents.tryEmit(container.string(R.string.error_agent_reported))
296 }
297 when (event) {
298 is PermissionUpdated ->
299 if (event.properties.sessionID == sessionId) _pendingPermission.value = event.properties
300 is PermissionReplied ->
301 if (event.properties.sessionID == sessionId && event.properties.permissionID == _pendingPermission.value?.id) _pendingPermission.value = null
302 is SessionUpdated ->
303 if (event.properties.info.id == sessionId) _sessionTitle.value = event.properties.info.displayTitle
304 is SessionDeleted ->
305 if (event.properties.info?.id == sessionId || event.properties.sessionID == sessionId) {
306 _sessionDeleted.value = true
307 }
308 else -> {}
309 }
310 }
311 } catch (e: CancellationException) {
312 throw e
313 } catch (e: Exception) {
314 Log.w("ChatViewModel", "SSE event collector error", e)
315 }
316 }
317 }
318 // Load the model catalog; preselect the server default. Failure is non-fatal —
319 // sending with no model just uses the server's default agent/model.
320 observeCatalog(
321 tag = "model catalog",
322 loading = _modelsLoading,
323 error = _modelsError,
324 fetch = { it.providers() },
325 onSuccess = { resp ->
326 val options = resp.toOptions()
327 _models.value = options
328 _selectedModel.value = resp.defaultOption(options)
329 },
330 onNull = { _models.value = emptyList(); _selectedModel.value = null },
331 reloadTrigger = _modelsReload,
332 )
333 // Load the agent catalog (non-fatal).
334 observeCatalog(
335 tag = "agent catalog",
336 loading = _agentsLoading,
337 error = _agentsError,
338 fetch = { it.agents() },
339 onSuccess = { _agents.value = it },
340 onNull = { _agents.value = emptyList() },
341 reloadTrigger = _agentsReload,
342 )
343 // Load the command catalog (non-fatal).
344 observeCatalog(
345 tag = "command catalog",
346 loading = _commandsLoading,
347 error = _commandsError,
348 fetch = { it.commands() },
349 onSuccess = { _commands.value = it },
350 onNull = { _commands.value = emptyList() },
351 reloadTrigger = _commandsReload,
352 )
353 // Resolve the human-readable session title for the app bar (non-fatal).
354 // Skips the REST call when the SSE event collector already delivered the title
355 // (SessionUpdated fires shortly after connect), avoiding a redundant full
356 // session-list download.
357 viewModelScope.launch {
358 container.activeConnection.collectLatest { conn ->
359 if (conn == null) return@collectLatest
360 if (_sessionTitle.value != null) return@collectLatest
361 runCatchingCancellable { conn.repository.listSessions() }
362 .getOrNull()
363 ?.firstOrNull { it.id == sessionId }
364 ?.let { _sessionTitle.value = it.displayTitle }
365 }
366 }
367 // If the SSE stream drops mid-run, the run indicator would spin forever;
368 // reset it so the UI doesn't look stuck while the banner shows "Reconnecting…".
369 viewModelScope.launch {
370 container.activeConnection.collectLatest { conn ->
371 if (conn == null) return@collectLatest
372 try {
373 conn.events.state.collect { state ->
374 if ((state == EventStreamClient.ConnectionState.Disconnected ||
375 state == EventStreamClient.ConnectionState.Failed) && _running.value) {
376 _running.value = false
377 }
378 }
379 } catch (e: CancellationException) {
380 throw e
381 } catch (e: Exception) {
382 Log.w("ChatViewModel", "SSE state collector error", e)
383 }
384 }
385 }
386 }
388 /** Sends [text]; returns true on success so the caller can clear the draft only then. */
389 fun send(text: String): Boolean {
390 val conn = connection ?: return false
391 val trimmed = text.trim()
392 if (trimmed.isEmpty() || !_running.compareAndSet(false, true)) return false
393 _failedDraft.value = null
394 // Clear the in-memory draft for the UI immediately, but don't persist the clear
395 // yet — if the send fails and the process dies before we restore, the draft
396 // would be lost forever. The persisted draft is cleared only on success.
397 // suppressDraftPersist prevents the debounced collector from persisting the
398 // empty draft before the send resolves.
399 suppressDraftPersist.set(true)
400 _draft.value = ""
401 viewModelScope.launch {
402 runCatchingCancellable {
403 conn.repository.sendPrompt(
404 sessionId,
405 trimmed,
406 model = _selectedModel.value?.ref,
407 agent = _selectedAgent.value,
408 )
409 }.onFailure {
410 suppressDraftPersist.set(false)
411 _failedDraft.value = trimmed
412 // Only restore the draft if the user hasn't typed anything new since.
413 if (_draft.value.isBlank()) updateDraft(trimmed)
414 _errorEvents.tryEmit(container.friendlyError(it))
415 _running.value = false
416 }.onSuccess {
417 suppressDraftPersist.set(false)
418 // Send succeeded — now it's safe to persist the empty draft.
419 container.draftStore.set(sessionId, "")
420 // Don't reset _running here: the agent continues streaming via SSE.
421 // _running is cleared on SessionIdle/SessionError (see event collector)
422 // or when the SSE stream drops (see connection state watcher below).
423 }
424 }
425 return true
426 }
428 /** Re-send the last draft whose send failed, if any. */
429 fun retryFailed() {
430 val draft = _failedDraft.value ?: return
431 // Don't clear _failedDraft until send() accepts the text — if _running is
432 // already true, send() returns false and the draft would be lost forever.
433 if (send(draft)) {
434 _failedDraft.value = null
435 }
436 }
438 /** Invoke a slash-command by name via the server's /command endpoint. */
439 fun runCommand(command: Command) {
440 val conn = connection ?: return
441 if (!_running.compareAndSet(false, true)) return
442 viewModelScope.launch {
443 runCatchingCancellable {
444 conn.repository.runCommand(sessionId, command.name, agent = command.agent)
445 }.onFailure {
446 _errorEvents.tryEmit(container.friendlyError(it))
447 _running.value = false
448 }
449 }
450 }
452 fun abort() {
453 val conn = connection ?: return
454 viewModelScope.launch {
455 runCatchingCancellable { conn.repository.abort(sessionId) }
456 .onSuccess { _running.value = false }
457 .onFailure { _errorEvents.tryEmit(container.friendlyError(it)) }
458 }
459 }
461 /** Reconnect to the most recently used server profile (used when the connection is gone). */
462 fun reconnect() {
463 if (container.activeConnection.value != null) return
464 // Guard against concurrent reconnect calls: the user can tap the button
465 // multiple times while _reconnecting is true, which would launch parallel
466 // connect() coroutines that race to set the active connection, leaking the
467 // intermediate connections.
468 if (!_reconnecting.compareAndSet(false, true)) return
469 viewModelScope.launch {
470 try {
471 val recent = runCatchingCancellable { container.profileStore.profiles.first() }
472 .getOrDefault(emptyList())
473 .firstOrNull()
474 if (recent == null) {
475 _errorEvents.tryEmit(container.string(R.string.no_servers_to_reconnect))
476 return@launch
477 }
478 runCatchingCancellable {
479 val conn = container.connect(recent)
480 conn.api.ping()
481 }.onFailure {
482 container.disconnect()
483 _errorEvents.tryEmit(container.friendlyError(it))
484 }
485 } finally {
486 _reconnecting.value = false
487 }
488 }
489 }
491 fun respondPermission(permission: Permission, response: PermissionResponse) {
492 val conn = connection ?: return
493 // Clear optimistically; permission.replied will confirm. If the call fails we
494 // restore the pending permission so the user can retry instead of being stuck
495 // with a dismissed dialog and a paused tool run.
496 _pendingPermission.value = null
497 viewModelScope.launch {
498 runCatchingCancellable { conn.api.respondPermission(sessionId, permission.id, response) }
499 .onFailure {
500 _errorEvents.tryEmit(container.friendlyError(it))
501 // Only restore if no new permission has arrived in the meantime.
502 if (_pendingPermission.value == null) _pendingPermission.value = permission
503 }
504 }
505 }
507 override fun onCleared() {
508 super.onCleared()
509 // Flush any pending debounced draft so it survives navigation. Uses an
510 // asynchronous apply() (not a synchronous commit) so the main thread isn't
511 // blocked on disk I/O — Android's SharedPreferences framework guarantees
512 // pending apply() writes are flushed before the process exits.
513 val pending = _draft.value
514 if (pending.isNotEmpty()) {
515 container.draftStore.flushDraft(sessionId, pending)
516 }
517 }