git.iko.soy/opencode-android
on d29e57b help sign in
opencode-android/app/src/main/java/soy/iko/opencode/ui/chat/ChatViewModel.kt
557 lines 26 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 /** A follow-up the user typed while a run was active. send() queues it here and
118 * auto-sends once the run completes (SessionIdle), so the user isn't blocked with
119 * no way to send and no indication why the Send button is gone. */
120 private val _queuedFollowUp = MutableStateFlow<String?>(null)
121 val queuedFollowUp: StateFlow<String?> = _queuedFollowUp.asStateFlow()
123 /** One-shot error events surfaced as snackbars. A SharedFlow (not StateFlow) so each
124 * emission is delivered independently — two rapid failures both get a snackbar
125 * instead of the second silently overwriting the first. */
126 private val _errorEvents = MutableSharedFlow<String>(
127 extraBufferCapacity = NetworkConfig.snackbarEventBufferCapacity,
128 onBufferOverflow = BufferOverflow.DROP_OLDEST,
129 )
130 val errorEvents: SharedFlow<String> = _errorEvents.asSharedFlow()
132 private val _models = MutableStateFlow<List<ModelOption>>(emptyList())
133 val models: StateFlow<List<ModelOption>> = _models.asStateFlow()
135 private val _modelsLoading = MutableStateFlow(true)
136 val modelsLoading: StateFlow<Boolean> = _modelsLoading.asStateFlow()
138 private val _modelsError = MutableStateFlow(false)
139 val modelsError: StateFlow<Boolean> = _modelsError.asStateFlow()
141 private val _selectedModel = MutableStateFlow<ModelOption?>(null)
142 val selectedModel: StateFlow<ModelOption?> = _selectedModel.asStateFlow()
144 val connectionState: StateFlow<EventStreamClient.ConnectionState> =
145 container.activeConnection
146 .flatMapLatest { it?.events?.state ?: flowOf(EventStreamClient.ConnectionState.Disconnected) }
147 .stateIn(
148 scope = viewModelScope,
149 started = SharingStarted.WhileSubscribed(NetworkConfig.stateFlowSubscriptionTimeoutMs),
150 initialValue = EventStreamClient.ConnectionState.Disconnected,
151 )
153 private val _pendingPermission = MutableStateFlow<Permission?>(null)
154 val pendingPermission: StateFlow<Permission?> = _pendingPermission.asStateFlow()
156 private val _agents = MutableStateFlow<List<Agent>>(emptyList())
157 val agents: StateFlow<List<Agent>> = _agents.asStateFlow()
159 private val _agentsLoading = MutableStateFlow(true)
160 val agentsLoading: StateFlow<Boolean> = _agentsLoading.asStateFlow()
162 private val _agentsError = MutableStateFlow(false)
163 val agentsError: StateFlow<Boolean> = _agentsError.asStateFlow()
165 private val _selectedAgent = MutableStateFlow<String?>(null)
166 val selectedAgent: StateFlow<String?> = _selectedAgent.asStateFlow()
168 private val _commands = MutableStateFlow<List<Command>>(emptyList())
169 val commands: StateFlow<List<Command>> = _commands.asStateFlow()
171 private val _commandsLoading = MutableStateFlow(true)
172 val commandsLoading: StateFlow<Boolean> = _commandsLoading.asStateFlow()
174 private val _commandsError = MutableStateFlow(false)
175 val commandsError: StateFlow<Boolean> = _commandsError.asStateFlow()
177 private val _sessionTitle = MutableStateFlow<String?>(null)
178 val sessionTitle: StateFlow<String?> = _sessionTitle.asStateFlow()
180 /** Set when the current session is deleted via SSE, so the UI can navigate away. */
181 private val _sessionDeleted = MutableStateFlow(false)
182 val sessionDeleted: StateFlow<Boolean> = _sessionDeleted.asStateFlow()
184 /** The text of the last send that failed, surfaced so the UI can offer a retry. */
185 private val _failedDraft = MutableStateFlow<String?>(null)
186 val failedDraft: StateFlow<String?> = _failedDraft.asStateFlow()
188 /** Per-session draft, persisted so it survives navigation/process death. */
189 private val _draft = MutableStateFlow(container.draftStore.get(sessionId))
190 val draft: StateFlow<String> = _draft.asStateFlow()
192 init {
193 // DraftStore loads SharedPreferences asynchronously; on a cold start the
194 // synchronous get() above returns "" until the background load completes.
195 // Observe the ready signal and re-seed the draft so it appears in the UI.
196 viewModelScope.launch {
197 container.draftStore.ready.collect { ready ->
198 if (ready && _draft.value.isEmpty()) {
199 _draft.value = container.draftStore.get(sessionId)
200 }
201 }
202 }
203 }
205 fun selectModel(option: ModelOption) { _selectedModel.value = option }
207 fun selectAgent(name: String?) { _selectedAgent.value = name }
209 fun reloadModels() {
210 val conn = connection ?: return
211 viewModelScope.launch {
212 conn.api.invalidateProvidersCache()
213 _modelsReload.value++
214 }
215 }
217 fun reloadAgents() {
218 val conn = connection ?: return
219 viewModelScope.launch {
220 conn.api.invalidateAgentsCache()
221 _agentsReload.value++
222 }
223 }
225 fun reloadCommands() {
226 val conn = connection ?: return
227 viewModelScope.launch {
228 conn.api.invalidateCommandsCache()
229 _commandsReload.value++
230 }
231 }
233 /**
234 * Shared helper for init-block catalog observers: re-fetches whenever the active
235 * connection or the catalog's reload trigger changes, invoking [onNull] to clear
236 * state on null so stale data from the old server doesn't persist. The
237 * [reloadTrigger] is merged with [activeConnection] so a manual reload (via
238 * reloadModels/reloadAgents/reloadCommands) cancels any in-flight observe fetch
239 * via collectLatest, preventing a stale observe result from overwriting a fresh
240 * reload result. Failure is non-fatal — the error flag is surfaced to the UI.
241 */
242 private fun <T> observeCatalog(
243 tag: String,
244 loading: MutableStateFlow<Boolean>,
245 error: MutableStateFlow<Boolean>,
246 fetch: suspend (soy.iko.opencode.data.network.OpencodeApiClient) -> T,
247 onSuccess: (T) -> Unit,
248 onNull: () -> Unit,
249 reloadTrigger: StateFlow<Int>,
250 ) {
251 viewModelScope.launch {
252 merge(container.activeConnection, reloadTrigger).collectLatest { _ ->
253 val conn = container.activeConnection.value
254 if (conn == null) { onNull(); loading.value = false; error.value = false; return@collectLatest }
255 loading.value = true
256 error.value = false
257 runCatchingCancellable { fetch(conn.api) }
258 // fetch() does HTTP via withRetry, so the failure can be a
259 // ClientRequestException whose message embeds the full request URL
260 // (may contain auth/paths). Log only a scrubbed summary.
261 .onFailure { Log.w("ChatViewModel", "Failed to load $tag: ${safeExceptionSummary(it)}"); error.value = true }
262 .getOrNull()?.let(onSuccess)
263 loading.value = false
264 }
265 }
266 }
268 fun updateDraft(text: String) {
269 _draft.value = text
270 }
272 init {
273 // Debounce draft persistence so we don't write to disk on every keystroke.
274 // Skip persisting empty drafts that were set by send() (suppressed via
275 // suppressDraftPersist) — send() clears the in-memory draft immediately for
276 // UI feedback but deliberately doesn't persist the clear until the send
277 // succeeds, so a failed send can restore the draft.
278 viewModelScope.launch {
279 _draft.drop(1).debounce(NetworkConfig.draftDebounceMs).collect { text ->
280 if (text.isBlank() && suppressDraftPersist.get()) return@collect
281 runCatchingCancellable { container.draftStore.set(sessionId, text) }
282 .onFailure { Log.w("ChatViewModel", "Failed to persist draft", it) }
283 }
284 }
285 // Reset per-connection state when the active server changes so stale spinners,
286 // permission dialogs, errors, and agent selections from the old server don't
287 // persist into the new one.
288 viewModelScope.launch {
289 container.activeConnection.collectLatest { conn ->
290 if (conn == null) return@collectLatest
291 _running.value = false
292 _pendingPermission.value = null
293 _failedDraft.value = null
294 _queuedFollowUp.value = null
295 _selectedAgent.value = null
296 _sessionTitle.value = null
297 try {
298 conn.events.events.collect { event ->
299 if (SessionRepository.isIdle(event, sessionId)) {
300 _running.value = false
301 // Auto-send a queued follow-up once the previous run finishes,
302 // so the user's drafted-while-running message isn't lost.
303 val queued = _queuedFollowUp.value
304 if (queued != null) {
305 _queuedFollowUp.value = null
306 send(queued)
307 }
308 }
309 if (SessionRepository.isError(event, sessionId)) {
310 _running.value = false
311 _queuedFollowUp.value = null
312 _errorEvents.tryEmit(container.string(R.string.error_agent_reported))
313 }
314 when (event) {
315 is PermissionUpdated ->
316 if (event.properties.sessionID == sessionId) _pendingPermission.value = event.properties
317 is PermissionReplied ->
318 if (event.properties.sessionID == sessionId && event.properties.permissionID == _pendingPermission.value?.id) _pendingPermission.value = null
319 is SessionUpdated ->
320 if (event.properties.info.id == sessionId) _sessionTitle.value = event.properties.info.displayTitle
321 is SessionDeleted ->
322 if (event.properties.info?.id == sessionId || event.properties.sessionID == sessionId) {
323 _sessionDeleted.value = true
324 }
325 else -> {}
326 }
327 }
328 } catch (e: CancellationException) {
329 throw e
330 } catch (e: Exception) {
331 Log.w("ChatViewModel", "SSE event collector error", e)
332 }
333 }
334 }
335 // Load the model catalog; preselect the server default. Failure is non-fatal —
336 // sending with no model just uses the server's default agent/model.
337 observeCatalog(
338 tag = "model catalog",
339 loading = _modelsLoading,
340 error = _modelsError,
341 fetch = { it.providers() },
342 onSuccess = { resp ->
343 val options = resp.toOptions()
344 _models.value = options
345 _selectedModel.value = resp.defaultOption(options)
346 },
347 onNull = { _models.value = emptyList(); _selectedModel.value = null },
348 reloadTrigger = _modelsReload,
349 )
350 // Load the agent catalog (non-fatal).
351 observeCatalog(
352 tag = "agent catalog",
353 loading = _agentsLoading,
354 error = _agentsError,
355 fetch = { it.agents() },
356 onSuccess = { _agents.value = it },
357 onNull = { _agents.value = emptyList() },
358 reloadTrigger = _agentsReload,
359 )
360 // Load the command catalog (non-fatal).
361 observeCatalog(
362 tag = "command catalog",
363 loading = _commandsLoading,
364 error = _commandsError,
365 fetch = { it.commands() },
366 onSuccess = { _commands.value = it },
367 onNull = { _commands.value = emptyList() },
368 reloadTrigger = _commandsReload,
369 )
370 // Resolve the human-readable session title for the app bar (non-fatal).
371 // Skips the REST call when the SSE event collector already delivered the title
372 // (SessionUpdated fires shortly after connect), avoiding a redundant full
373 // session-list download.
374 viewModelScope.launch {
375 container.activeConnection.collectLatest { conn ->
376 if (conn == null) return@collectLatest
377 if (_sessionTitle.value != null) return@collectLatest
378 runCatchingCancellable { conn.repository.listSessions() }
379 .getOrNull()
380 ?.firstOrNull { it.id == sessionId }
381 ?.let { _sessionTitle.value = it.displayTitle }
382 }
383 }
384 // If the SSE stream drops mid-run, the run indicator would spin forever;
385 // reset it so the UI doesn't look stuck while the banner shows "Reconnecting…".
386 viewModelScope.launch {
387 container.activeConnection.collectLatest { conn ->
388 if (conn == null) return@collectLatest
389 try {
390 conn.events.state.collect { state ->
391 if ((state == EventStreamClient.ConnectionState.Disconnected ||
392 state == EventStreamClient.ConnectionState.Failed) && _running.value) {
393 _running.value = false
394 }
395 }
396 } catch (e: CancellationException) {
397 throw e
398 } catch (e: Exception) {
399 Log.w("ChatViewModel", "SSE state collector error", e)
400 }
401 }
402 }
403 }
405 /** Sends [text]; returns true on success so the caller can clear the draft only then. */
406 fun send(text: String): Boolean {
407 val conn = connection ?: return false
408 val trimmed = text.trim()
409 if (trimmed.isEmpty() || !_running.compareAndSet(false, true)) return false
410 _failedDraft.value = null
411 // Clear the in-memory draft for the UI immediately, but don't persist the clear
412 // yet — if the send fails and the process dies before we restore, the draft
413 // would be lost forever. The persisted draft is cleared only on success.
414 // suppressDraftPersist prevents the debounced collector from persisting the
415 // empty draft before the send resolves.
416 suppressDraftPersist.set(true)
417 _draft.value = ""
418 viewModelScope.launch {
419 runCatchingCancellable {
420 conn.repository.sendPrompt(
421 sessionId,
422 trimmed,
423 model = _selectedModel.value?.ref,
424 agent = _selectedAgent.value,
425 )
426 }.onFailure {
427 suppressDraftPersist.set(false)
428 _failedDraft.value = trimmed
429 // Only restore the draft if the user hasn't typed anything new since.
430 if (_draft.value.isBlank()) updateDraft(trimmed)
431 _errorEvents.tryEmit(container.friendlyError(it))
432 _running.value = false
433 }.onSuccess {
434 suppressDraftPersist.set(false)
435 // Send succeeded — now it's safe to persist the empty draft.
436 container.draftStore.set(sessionId, "")
437 // Don't reset _running here: the agent continues streaming via SSE.
438 // _running is cleared on SessionIdle/SessionError (see event collector)
439 // or when the SSE stream drops (see connection state watcher below).
440 }
441 }
442 return true
443 }
445 /** Re-send the last draft whose send failed, if any. */
446 fun retryFailed() {
447 val draft = _failedDraft.value ?: return
448 // Don't clear _failedDraft until send() accepts the text — if _running is
449 // already true, send() returns false and the draft would be lost forever.
450 if (send(draft)) {
451 _failedDraft.value = null
452 }
453 }
455 /**
456 * Queue [text] to be sent automatically when the current run finishes, or clear any
457 * queued follow-up when [text] is blank. Used when the user taps Send while a run
458 * is already active — the Send button is replaced by Stop during a run, but the
459 * input field stays enabled, so a follow-up typed mid-run would otherwise have
460 * nowhere to go.
461 */
462 fun queueFollowUp(text: String) {
463 val trimmed = text.trim()
464 _queuedFollowUp.value = trimmed.takeIf { it.isNotEmpty() }
465 }
467 /** Manually re-fetch messages by forcing the SSE stream to reconnect, which
468 * triggers the repository's re-seed-from-REST logic. Gives the user a recovery
469 * path when the SSE stream silently drops and the auto-reconnect re-seed is slow
470 * or fails — a pull-to-refresh or top-bar refresh forces a fresh fetch under the
471 * existing generation-guarded merge. */
472 fun refreshMessages() {
473 val conn = connection ?: return
474 conn.events.triggerReconnect()
475 }
477 /** Invoke a slash-command by name via the server's /command endpoint. */
478 fun runCommand(command: Command) {
479 val conn = connection ?: return
480 if (!_running.compareAndSet(false, true)) return
481 viewModelScope.launch {
482 runCatchingCancellable {
483 conn.repository.runCommand(sessionId, command.name, agent = command.agent)
484 }.onFailure {
485 _errorEvents.tryEmit(container.friendlyError(it))
486 _running.value = false
487 }
488 }
489 }
491 fun abort() {
492 val conn = connection ?: return
493 viewModelScope.launch {
494 runCatchingCancellable { conn.repository.abort(sessionId) }
495 .onSuccess { _running.value = false }
496 .onFailure { _errorEvents.tryEmit(container.friendlyError(it)) }
497 }
498 }
500 /** Reconnect to the most recently used server profile (used when the connection is gone). */
501 fun reconnect() {
502 if (container.activeConnection.value != null) return
503 // Guard against concurrent reconnect calls: the user can tap the button
504 // multiple times while _reconnecting is true, which would launch parallel
505 // connect() coroutines that race to set the active connection, leaking the
506 // intermediate connections.
507 if (!_reconnecting.compareAndSet(false, true)) return
508 viewModelScope.launch {
509 try {
510 val recent = runCatchingCancellable { container.profileStore.profiles.first() }
511 .getOrDefault(emptyList())
512 .firstOrNull()
513 if (recent == null) {
514 _errorEvents.tryEmit(container.string(R.string.no_servers_to_reconnect))
515 return@launch
516 }
517 runCatchingCancellable {
518 val conn = container.connect(recent)
519 conn.api.ping()
520 }.onFailure {
521 container.disconnect()
522 _errorEvents.tryEmit(container.friendlyError(it))
523 }
524 } finally {
525 _reconnecting.value = false
526 }
527 }
528 }
530 fun respondPermission(permission: Permission, response: PermissionResponse) {
531 val conn = connection ?: return
532 // Clear optimistically; permission.replied will confirm. If the call fails we
533 // restore the pending permission so the user can retry instead of being stuck
534 // with a dismissed dialog and a paused tool run.
535 _pendingPermission.value = null
536 viewModelScope.launch {
537 runCatchingCancellable { conn.api.respondPermission(sessionId, permission.id, response) }
538 .onFailure {
539 _errorEvents.tryEmit(container.friendlyError(it))
540 // Only restore if no new permission has arrived in the meantime.
541 if (_pendingPermission.value == null) _pendingPermission.value = permission
542 }
543 }
544 }
546 override fun onCleared() {
547 super.onCleared()
548 // Flush any pending debounced draft so it survives navigation. Uses an
549 // asynchronous apply() (not a synchronous commit) so the main thread isn't
550 // blocked on disk I/O — Android's SharedPreferences framework guarantees
551 // pending apply() writes are flushed before the process exits.
552 val pending = _draft.value
553 if (pending.isNotEmpty()) {
554 container.draftStore.flushDraft(sessionId, pending)
555 }
556 }