1package soy.iko.opencode.di
3import android.content.Context
4import android.net.ConnectivityManager
5import android.net.Network
6import android.net.NetworkCapabilities
7import android.net.NetworkRequest
9import soy.iko.opencode.data.model.AssistantMessage
10import soy.iko.opencode.data.model.BusEvent
11import soy.iko.opencode.data.model.FilePromptPart
12import soy.iko.opencode.data.model.MessagePartUpdated
13import soy.iko.opencode.data.model.MessageUpdated
14import soy.iko.opencode.data.model.Permission
15import soy.iko.opencode.data.model.PermissionReplied
16import soy.iko.opencode.data.model.PermissionResponse
17import soy.iko.opencode.data.model.PermissionUpdated
18import soy.iko.opencode.data.model.StepFinishPart
19import soy.iko.opencode.data.model.ServerProfile
20import soy.iko.opencode.data.model.SessionError
21import soy.iko.opencode.data.model.SessionIdle
22import soy.iko.opencode.data.model.ToolPart
23import soy.iko.opencode.data.model.TODO_WRITE_TOOL
24import soy.iko.opencode.data.model.parseTodos
25import soy.iko.opencode.data.model.statusEnum
26import soy.iko.opencode.data.model.TodoStatus
27import soy.iko.opencode.data.model.inputElement
28import soy.iko.opencode.data.repo.AttachmentDraftStore
29import soy.iko.opencode.data.repo.BackupManager
30import soy.iko.opencode.data.repo.DraftStore
31import soy.iko.opencode.data.repo.ErrorKind
32import soy.iko.opencode.data.repo.MessageCacheStore
33import soy.iko.opencode.data.repo.OutboxMessage
34import soy.iko.opencode.data.repo.OutboxStore
35import soy.iko.opencode.data.repo.RecentModelsStore
36import soy.iko.opencode.data.repo.SearchHistoryStore
37import soy.iko.opencode.data.repo.SessionPrefsStore
38import soy.iko.opencode.data.repo.SessionRepository
39import soy.iko.opencode.data.repo.ProfileStore
40import soy.iko.opencode.data.repo.SettingsStore
41import soy.iko.opencode.data.repo.classifyError
42import soy.iko.opencode.data.repo.responseStatusCode
43import soy.iko.opencode.data.network.EventStreamClient
44import soy.iko.opencode.data.network.HttpClientFactory
45import soy.iko.opencode.data.network.NetworkConfig
46import soy.iko.opencode.data.network.OpencodeApiClient
47import soy.iko.opencode.notification.NotificationChannels
48import soy.iko.opencode.notification.RunForegroundService
49import soy.iko.opencode.util.safeExceptionSummary
50import soy.iko.opencode.notification.SessionNotifications
51import soy.iko.opencode.R
52import soy.iko.opencode.util.runCatchingCancellable
53import kotlinx.coroutines.CoroutineScope
54import kotlinx.coroutines.Dispatchers
55import kotlinx.coroutines.ExperimentalCoroutinesApi
56import kotlinx.coroutines.SupervisorJob
57import kotlinx.coroutines.cancel
58import kotlinx.coroutines.delay
59import kotlinx.coroutines.flow.MutableSharedFlow
60import kotlinx.coroutines.flow.MutableStateFlow
61import kotlinx.coroutines.flow.SharedFlow
62import kotlinx.coroutines.flow.StateFlow
63import kotlinx.coroutines.flow.asSharedFlow
64import kotlinx.coroutines.flow.asStateFlow
65import kotlinx.coroutines.flow.collectLatest
66import kotlinx.coroutines.flow.combine
67import kotlinx.coroutines.flow.debounce
68import kotlinx.coroutines.flow.distinctUntilChanged
69import kotlinx.coroutines.flow.drop
70import kotlinx.coroutines.flow.emptyFlow
71import kotlinx.coroutines.flow.first
72import kotlinx.coroutines.Job
73import kotlinx.coroutines.flow.flatMapLatest
74import kotlinx.coroutines.flow.map
75import kotlinx.coroutines.flow.update
76import kotlinx.coroutines.isActive
77import kotlinx.coroutines.launch
78import kotlinx.coroutines.sync.Mutex
79import kotlinx.coroutines.sync.withLock
80import kotlinx.coroutines.withTimeoutOrNull
81import java.util.concurrent.atomic.AtomicBoolean
84 * Result of probing a server URL to check connectivity and whether authentication
85 * is required, without persisting a profile or opening a long-lived connection.
87sealed class ProbeResult {
88 /** Server is reachable and accepts unauthenticated requests. */
89 object Reachable : ProbeResult()
91 /** Server is reachable but rejected the probe with an authentication error. */
92 object NeedsAuth : ProbeResult()
94 /** Server could not be reached (host unresolved, connection refused, timeout, etc.). */
95 data class Unreachable(val error: String) : ProbeResult()
99 * Hand-written service locator held by [soy.iko.opencode.OpencodeApp]. Owns the
100 * process-wide singletons and the currently active [OpencodeConnection].
102open class AppContainer private constructor(
103 private val appContext: Context?,
104 private val skipInit: Boolean,
105 @Suppress("unused") private val testMode: Boolean,
107 constructor(context: Context) : this(context.applicationContext, false, false)
108 protected constructor() : this(null, true, true)
110 private val appScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
111 open val profileStore: ProfileStore by lazy { ProfileStore(appContext!!) }
112 open val settingsStore: SettingsStore by lazy { SettingsStore(appContext!!) }
113 open val draftStore: DraftStore by lazy { DraftStore(appContext!!, appScope) }
114 open val attachmentDraftStore: AttachmentDraftStore by lazy { AttachmentDraftStore(appContext!!) }
115 open val sessionPrefsStore: SessionPrefsStore by lazy { SessionPrefsStore(appContext!!) }
116 open val messageCacheStore: MessageCacheStore by lazy { MessageCacheStore(appContext!!) }
117 open val outboxStore: OutboxStore by lazy { OutboxStore(appContext!!) }
118 open val searchHistoryStore: SearchHistoryStore by lazy { SearchHistoryStore(appContext!!) }
119 open val recentModelsStore: RecentModelsStore by lazy { RecentModelsStore(appContext!!) }
120 open val backupManager: BackupManager by lazy { BackupManager(profileStore, settingsStore, sessionPrefsStore) }
122 /** True while the outbox is actively flushing queued messages to the server, so the UI
123 * can show a "sending queued messages…" indicator. */
124 private val _outboxSending = MutableStateFlow(false)
125 open val outboxSending: StateFlow<Boolean> = _outboxSending.asStateFlow()
127 /** Bumped to nudge an immediate outbox flush (after enqueue, or a manual "Send now"),
128 * independent of connection/online changes. */
129 private val _outboxFlushTrigger = MutableStateFlow(0)
130 open fun flushOutbox() { _outboxFlushTrigger.update { it + 1 } }
132 private val _activeConnection = MutableStateFlow<OpencodeConnection?>(null)
133 open val activeConnection: StateFlow<OpencodeConnection?> = _activeConnection.asStateFlow()
135 /** True while a background auto-reconnect to the most-recent server is in flight. */
136 private val _reconnecting = MutableStateFlow(false)
137 open val reconnecting: StateFlow<Boolean> = _reconnecting.asStateFlow()
140 * Set once a startup auto-reconnect succeeds. The server list collects this and
141 * (once) navigates straight to the session list so a returning user skips it.
142 * [consumeAutoConnect] guards against re-firing on screen re-entry.
144 private val _autoConnectDone = MutableStateFlow(false)
145 open val autoConnectDone: StateFlow<Boolean> = _autoConnectDone.asStateFlow()
146 private val autoConnectConsumed = AtomicBoolean(false)
147 open fun consumeAutoConnect(): Boolean = autoConnectConsumed.compareAndSet(false, true)
149 /** Text shared into the app (ACTION_SEND), prefilled into a session draft. */
150 private val _pendingShare = MutableStateFlow<String?>(null)
151 open val pendingShare: StateFlow<String?> = _pendingShare.asStateFlow()
152 open fun setPendingShare(text: String) { _pendingShare.value = text }
153 open fun consumePendingShare(): String? {
154 val current = _pendingShare.value ?: return null
155 return if (_pendingShare.compareAndSet(current, null)) current else null
158 /** Image/file Uris (as strings) shared into the app, staged as attachments in the next
159 * opened session. Consumed once by the first ChatScreen to compose after the share. */
160 private val _pendingSharedMedia = MutableStateFlow<List<String>>(emptyList())
161 open val pendingSharedMedia: StateFlow<List<String>> = _pendingSharedMedia.asStateFlow()
162 open fun setPendingSharedMedia(uris: List<String>) { _pendingSharedMedia.value = uris }
163 open fun consumePendingSharedMedia(): List<String> {
164 val current = _pendingSharedMedia.value
165 if (current.isEmpty()) return emptyList()
166 return if (_pendingSharedMedia.compareAndSet(current, emptyList())) current else emptyList()
170 * Set from an external "New session" trigger (launcher shortcut / QS tile). The nav host
171 * consumes it once connected, creating a fresh session and opening it. Uses a counter
172 * (not a boolean) so re-tapping the shortcut while a prior request is still pending
173 * increments the value and re-fires the LaunchedEffect — a boolean StateFlow wouldn't
174 * emit when set to the same value, making the request unrecoverable by re-tapping.
176 private val _pendingNewSession = MutableStateFlow(0)
177 open val pendingNewSession: StateFlow<Int> = _pendingNewSession.asStateFlow()
178 open fun requestNewSession() { _pendingNewSession.update { it + 1 } }
179 open fun consumePendingNewSession(): Boolean {
181 val current = _pendingNewSession.value
182 if (current == 0) return false
183 if (_pendingNewSession.compareAndSet(current, current - 1)) return true
187 /** A one-shot signal to navigate to the Diagnostics screen (e.g. from the crash-relaunch
188 * prompt). Consumed by OpencodeApp's NavHost. */
189 private val _pendingDiagnostics = MutableStateFlow(false)
190 open val pendingDiagnostics: StateFlow<Boolean> = _pendingDiagnostics.asStateFlow()
191 open fun requestDiagnostics() { _pendingDiagnostics.value = true }
192 open fun consumePendingDiagnostics(): Boolean = _pendingDiagnostics.compareAndSet(true, false)
195 * A session id to open from an external trigger (a notification tap or a deep link),
196 * paired with the originating server's profile id (if known) so a tap after the user
197 * has switched servers routes back to the server that ran the session. The nav host
198 * consumes it once a connection is active (switching to [profileId] first if needed).
200 data class PendingOpenSession(val sessionId: String, val profileId: String?)
202 private val _pendingOpenSession = MutableStateFlow<PendingOpenSession?>(null)
203 open val pendingOpenSession: StateFlow<PendingOpenSession?> = _pendingOpenSession.asStateFlow()
204 open fun requestOpenSession(id: String, profileId: String? = null) {
205 _pendingOpenSession.value = PendingOpenSession(id, profileId)
207 open fun consumePendingOpenSession(): PendingOpenSession? {
208 val current = _pendingOpenSession.value ?: return null
209 return if (_pendingOpenSession.compareAndSet(current, null)) current else null
213 * A file path requested to be opened (via deep link `opencode://file/{path}` or future
214 * entry points), consumed by the NavHost to navigate to the file viewer. Mirrors the
215 * session open-signal pattern: a StateFlow the NavHost observes, drained atomically.
217 private val _pendingOpenFile = MutableStateFlow<String?>(null)
218 open val pendingOpenFile: StateFlow<String?> = _pendingOpenFile.asStateFlow()
219 open fun requestOpenFile(path: String) { _pendingOpenFile.value = path }
220 open fun consumePendingOpenFile(): String? {
221 val current = _pendingOpenFile.value ?: return null
222 return if (_pendingOpenFile.compareAndSet(current, null)) current else null
226 * The session the user is currently viewing (or null when not in a chat). Drives the
227 * unread tracker: messages arriving for any *other* session mark it unread, and
228 * opening a session clears its unread state.
230 private val _currentSession = MutableStateFlow<String?>(null)
231 open val currentSession: StateFlow<String?> = _currentSession.asStateFlow()
233 /** Session ids that received activity while not being viewed, mapped to the count
234 * of unread message events. A count (not just a presence set) lets the session
235 * list badge show "3 unread" instead of a bare dot, so the user can tell whether
236 * one reply or a whole burst arrived. */
237 private val _unread = MutableStateFlow<Map<String, Int>>(emptyMap())
238 open val unread: StateFlow<Map<String, Int>> = _unread.asStateFlow()
240 /** Session ids the user has muted. A muted session doesn't badge unread (the count is
241 * still tracked internally so unmuting can restore it, but the badge and completion
242 * notification are suppressed). Persisted via SessionPrefsStore so the choice survives
244 private val _mutedSessions = MutableStateFlow<Set<String>>(emptySet())
245 open val mutedSessions: StateFlow<Set<String>> = _mutedSessions.asStateFlow()
247 /** True when any assistant run is actively streaming across all sessions. Drives the
248 * disconnect confirmation on the session list (disconnecting mid-run kills it). */
249 private val _anyRunActive = MutableStateFlow(false)
250 open val anyRunActive: StateFlow<Boolean> = _anyRunActive.asStateFlow()
252 /** Session ids currently streaming an assistant run, surfaced so the session list can
253 * badge the row whose run is active. Mirrors [activeRuns] as an immutable snapshot so
254 * Compose can skip recomposition when the set hasn't changed. */
255 private val _runningSessionIds = MutableStateFlow<Set<String>>(emptySet())
256 open val runningSessionIds: StateFlow<Set<String>> = _runningSessionIds.asStateFlow()
258 /** Latest task-plan progress text for the active run (e.g. "Step 2 of 5"), surfaced in
259 * the foreground-service notification so the user can see what the agent is working on
260 * without opening the app. Null when no plan is available or no run is active. */
261 private val _runProgressText = MutableStateFlow<String?>(null)
262 open val runProgressText: StateFlow<String?> = _runProgressText.asStateFlow()
264 /** Whether the device currently has network connectivity. Distinct from the SSE
265 * connection state so the UI can tell "you're offline" (device) from "server
266 * unreachable" (credentials/host). Updated by the [networkCallback]. */
267 private val _isOnline = MutableStateFlow(true)
268 open val isOnline: StateFlow<Boolean> = _isOnline.asStateFlow()
270 /** Whether a foreground Activity is currently showing. Distinct from [currentSession],
271 * which stays set while the app is merely backgrounded (the chat screen isn't disposed
272 * when the user locks their phone). Permission/completion notifications need the real
273 * foreground signal so they still fire for a session the user "has open" but has walked
274 * away from — the core of the run-in-the-background use case. Set from MainActivity's
276 private val _isForeground = MutableStateFlow(false)
277 open val isForeground: StateFlow<Boolean> = _isForeground.asStateFlow()
278 open fun setForeground(foreground: Boolean) { _isForeground.value = foreground }
280 /** True when the user is actively looking at [sessionId] (app foregrounded AND that
281 * session is the one on screen), so an in-app affordance (the permission dialog) is
282 * handling it and a notification would be redundant. */
283 private fun isActivelyViewing(sessionId: String): Boolean =
284 _isForeground.value && sessionId == _currentSession.value
286 open fun setCurrentSession(id: String?) {
287 _currentSession.value = id
289 _unread.update { it - id }
290 unreadMessageIds.remove(id)
292 SessionNotifications.cancel(it, id)
293 SessionNotifications.cancelPermission(it, id)
294 SessionNotifications.cancelError(it, id)
299 /** Clear unread badge and dedup set for a deleted session so its stale count doesn't
300 * linger in the session list badge or leak memory until the next connect/disconnect. */
301 open fun clearUnread(id: String) {
302 _unread.update { it - id }
303 unreadMessageIds.remove(id)
306 /** Mark a session as read (clear its unread badge) — the "Mark read" notification action. */
307 open fun markSessionRead(id: String) = clearUnread(id)
309 /** Mark a session as unread with a count of 1 (or restore a prior count). Used by the
310 * session list's "Mark as unread" overflow action so a user can re-flag a conversation
311 * they want to revisit. No-op when the session is the one currently being viewed — a
312 * viewed session clears its own unread badge via [setCurrentSession], so marking it
313 * unread would race and immediately clear. Muted sessions are still marked (the count is
314 * tracked so unmuting restores it) but the list UI hides the badge while muted. */
315 open fun markSessionUnread(id: String, count: Int = 1) {
316 if (id == _currentSession.value) return
317 _unread.update { it + (id to count) }
320 /** Toggle a session's mute state. A muted session suppresses the unread badge and the
321 * completion notification so a noisy conversation doesn't keep interrupting. Persisted
322 * via SessionPrefsStore so it survives process death. */
323 fun toggleSessionMute(id: String) {
324 val muted = id in _mutedSessions.value
325 val next = if (muted) _mutedSessions.value - id else _mutedSessions.value + id
326 _mutedSessions.value = next
328 kotlinx.coroutines.CoroutineScope(kotlinx.coroutines.Dispatchers.IO).launch {
329 runCatchingCancellable { sessionPrefsStore.setMuted(id, !muted) }
334 /** True when [id] is muted (unread badge and completion notification suppressed). */
335 fun isSessionMuted(id: String): Boolean = id in _mutedSessions.value
337 /** Clear every session's unread badge at once — the session list's "Mark all read" action. */
338 open fun clearAllUnread() {
339 _unread.update { emptyMap() }
342 /** Restore a session's unread badge after a failed server switch reconnects.
343 * Preserves the prior [count] so a session badged with "5 unread" before the
344 * switch attempt still shows 5 (not 1) after the restore. No-op for the session
345 * the user is currently viewing, since a viewed session is never badged. */
346 open fun restoreUnread(id: String, count: Int) {
347 if (id != _currentSession.value) {
348 _unread.update { it + (id to count) }
352 /** Resolve a localized string — view models reach resources through the container. */
353 open fun string(id: Int, vararg formatArgs: Any): String {
354 if (appContext == null) return ""
355 return if (formatArgs.isEmpty()) appContext.getString(id) else appContext.getString(id, *formatArgs)
359 * Convert a throwable into a user-facing message. Classifies the error by concrete
360 * type (network, timeout, HTTP status) rather than string-matching class names, so
361 * the message reflects what actually went wrong without leaking internal URLs/state.
363 open fun friendlyError(t: Throwable): String =
364 friendlyErrorFor(t, activeConnection.value?.profile?.baseUrl.orEmpty())
367 * Same classification as [friendlyError] but accepts an explicit base URL, so callers
368 * that aren't operating on the active connection (e.g. [probeServer]) can still
369 * produce a user-facing message with the right server address.
371 open fun friendlyErrorFor(t: Throwable, baseUrl: String): String =
372 when (classifyError(t)) {
373 ErrorKind.NOT_REACHABLE, ErrorKind.NETWORK ->
374 string(R.string.error_not_reachable, baseUrl)
375 ErrorKind.TIMEOUT -> string(R.string.error_timeout)
376 ErrorKind.SERVER -> string(R.string.error_server)
377 // Show only the HTTP status (e.g. 401, 404) — never the request URL,
378 // which a ClientRequestException carries in its message and which we
379 // promised not to leak (it can include auth or internal paths).
380 ErrorKind.CLIENT -> {
381 val code = responseStatusCode(t)
383 // Auth failures reuse the connection banner text: the request itself is
384 // fine, the credentials aren't, so the fix is re-editing the profile —
385 // not "request failed (401)" which implies an opaque program error.
386 401, 403 -> string(R.string.connection_failed)
387 404 -> string(R.string.error_not_found)
388 else -> code?.let { string(R.string.error_client_status, it) }
389 ?: string(R.string.error_generic)
392 ErrorKind.UNKNOWN -> string(R.string.error_generic)
396 * Drive the [RunForegroundService] from the process-lived [appScope] rather than from the
397 * Activity's composition. Previously the start/stop was wired to a
398 * `collectAsStateWithLifecycle` collector in the root composable, which pauses while the
399 * Activity is STOPPED — so a run *started while backgrounded* (e.g. via the notification
400 * inline-reply) never acquired foreground priority and its long-lived SSE stream could be
401 * killed by Doze, and a run that *ended* while backgrounded left the "working…"
402 * notification lingering until the app was reopened. Collecting here reacts to
403 * [anyRunActive] regardless of Activity state.
405 private fun observeRunForegroundService() {
406 val ctx = appContext ?: return
408 // Debounce so a run that starts then fails/idles almost immediately doesn't
409 // dispatch startForegroundService() then stopService() back-to-back — if the stop
410 // wins that race before the service calls startForeground(), Android raises the
411 // "did not then call startForeground()" crash ~5s later. Collapsing the window
412 // emits only the settled state; distinctUntilChanged drops a no-op re-emit when a
413 // brief true→false→true flurry settles back to the value already applied.
415 .debounce(NetworkConfig.runForegroundDebounceMs)
416 .distinctUntilChanged()
419 // When exactly one run is active, resolve its title (and id, for the
420 // Stop action) so the notification identifies which session is running
421 // and can be aborted from the notification. Multiple concurrent runs
422 // fall back to the generic title (naming one would be misleading).
423 val activeId = synchronized(activeRuns) {
424 activeRuns.toList().singleOrNull()
426 val title = activeId?.let { sid ->
427 runCatchingCancellable {
428 activeConnection.value?.repository?.listSessions()
429 ?.firstOrNull { it.id == sid }?.displayTitle
432 RunForegroundService.start(ctx, title, activeId, _runProgressText.value)
434 RunForegroundService.stop(ctx)
438 // Separate collector: update the FGS notification's progress text ("Step 2 of 5")
439 // as the agent works through its task plan. Re-sends the service intent with the
440 // updated progress, which the service uses to rebuild the notification.
443 .drop(1) // skip initial value — the anyRunActive collector handles the first start
444 .distinctUntilChanged()
445 .collect { progress ->
446 if (_anyRunActive.value) {
447 val activeId = synchronized(activeRuns) { activeRuns.toList().singleOrNull() }
448 val title = activeId?.let { sid ->
449 runCatchingCancellable {
450 activeConnection.value?.repository?.listSessions()
451 ?.firstOrNull { it.id == sid }?.displayTitle
454 RunForegroundService.start(ctx, title, activeId, progress)
461 * Reconcile active-run tracking on an SSE *reconnect*. A run that completes during a stream
462 * outage never delivers its `SessionIdle`, so its id would stay in [activeRuns] and pin
463 * [anyRunActive] (and the foreground service + "working…" notification) on indefinitely. On a
464 * reconnect we clear the tracking and let live streaming events re-populate it: a genuinely
465 * in-progress run re-adds itself within moments via [isLiveRunActivity], while a run that
466 * finished during the outage produces no further live events (its completion arrives only via
467 * the REST re-seed), so it correctly stays cleared. The first connect of a connection is not a
468 * reconnect; `connect()` already resets [activeRuns], so only subsequent Connected transitions
471 private fun observeRunReconcileOnReconnect() {
473 activeConnection.collectLatest { conn ->
474 if (conn == null) return@collectLatest
475 var hasConnectedBefore = false
476 conn.events.state.collect { state ->
477 if (state == EventStreamClient.ConnectionState.Connected) {
478 if (hasConnectedBefore) {
479 synchronized(activeRuns) {
484 hasConnectedBefore = true
492 * Flush the offline outbox whenever there's an active connection and the device is
493 * online (and on any manual [flushOutbox] nudge). Queued messages are only sent to the
494 * server they were composed for (matched by profile id), so a reconnect to a different
495 * server never misfires them against a session id that doesn't exist there. Uses a plain
496 * `collect` (not collectLatest) so an in-flight send is never cancelled mid-flight and
497 * send order is preserved; each message carries a stable idempotency key (its persistent
498 * outbox id) so even a re-send across flushes is deduplicated server-side.
500 private fun observeOutbox() {
502 combine(activeConnection, isOnline, _outboxFlushTrigger) { conn, online, _ -> conn to online }
503 .collect { (conn, online) ->
504 if (conn == null || !online) return@collect
510 private suspend fun flushOutboxNow(conn: OpencodeConnection) {
511 val pending = outboxStore.messages.value
512 .filter { it.profileId == conn.profile.id }
513 .sortedBy { it.createdAt }
514 if (pending.isEmpty()) return
515 _outboxSending.value = true
517 for (msg in pending) {
518 val result = runCatchingCancellable {
519 conn.repository.sendPrompt(
522 attachments = msg.attachments.map {
523 FilePromptPart(mime = it.mime, url = it.url, filename = it.filename)
527 // Reuse the persistent outbox id as a stable idempotency key so a
528 // later flush of a message whose earlier POST reached the server (but
529 // whose response was lost) is deduplicated server-side instead of
530 // starting a duplicate agent run.
531 idempotencyKey = msg.id,
534 Log.w("AppContainer", "Outbox flush failed for ${msg.sessionId}: ${safeExceptionSummary(it)}")
536 if (result.isSuccess) {
537 outboxStore.remove(msg.id)
540 // A permanently-undeliverable message (a non-408/429 4xx — e.g. the target
541 // session was deleted server-side, yielding a 404) must be dropped, not left at
542 // the head of the queue. The loop stops at the first failure to preserve send
543 // order, so a poison message would otherwise block every later queued message
544 // for this profile forever. Transient failures (offline again, 5xx, 408/429)
545 // keep the queue intact and retry on the next trigger.
546 // 401/403 (bad/expired credentials) are NOT permanent: the message itself is
547 // deliverable once the user fixes the profile. Dropping it would silently lose
548 // the user's composed offline prompt on a recoverable auth failure. They stay
549 // in the queue and retry on the next flush after auth is restored.
550 val status = result.exceptionOrNull()?.let { responseStatusCode(it) }
551 // Permanent = a non-408/429/401/403 4xx, matching withRetry's non-retryable rule
552 // except for auth failures, which are recoverable by re-editing the profile.
553 val permanent = when (status) {
554 null, 408, 429, 401, 403 -> false
561 "Dropping undeliverable outbox message ${msg.id} for ${msg.sessionId} (HTTP $status): ${string(R.string.outbox_dropped_text)}",
563 outboxStore.remove(msg.id)
564 // Previously this silent drop only hit logcat, so the user got no signal
565 // that their composed reply never landed. Surface it on the ERROR channel.
566 // postError is the only public entry point that posts there; it renders a
567 // fixed title/text and interpolates its title arg, so the localized
568 // outbox_dropped_title is passed as that arg (a user-facing message rather
569 // than an opaque session id). Tap routes back to the originating profile.
572 runCatchingCancellable {
573 SessionNotifications.postError(
575 string(R.string.outbox_dropped_title),
579 Log.w("AppContainer", "outbox drop notification failed: ${safeExceptionSummary(it)}")
587 _outboxSending.value = false
591 /** Release resources held for the process lifetime (network callback, app scope). */
592 open fun shutdown() {
593 // Cancel the app scope first so any coroutine holding the connection mutex
594 // (e.g. a connect() suspended on profileStore) is cancelled and releases the
595 // mutex. Otherwise the runBlocking below would deadlock waiting for a coroutine
596 // that can't be cancelled until after the runBlocking completes.
598 // Bounded runBlocking so a stuck mutex or a slow close() can't ANR the app.
599 // 2 seconds is generous: appScope.cancel() already triggered cancellation of
600 // any coroutine holding the mutex; we're just waiting for it to unwind.
601 kotlinx.coroutines.runBlocking {
602 withTimeoutOrNull(2_000) {
603 connectionMutex.withLock {
604 runCatching { _activeConnection.value?.close() }
605 _activeConnection.value = null
608 // If the timeout fired, force-clear the connection so it isn't left dangling.
609 _activeConnection.value = null
611 val cm = appContext?.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
612 networkCallback?.let { runCatching { cm?.unregisterNetworkCallback(it) } }
613 if (!skipInit) soy.iko.opencode.data.repo.CrashLogger.get(appContext!!).shutdown()
614 if (!skipInit) draftStore.shutdown()
618 * Watch the SSE bus for new message activity and badge any session that isn't
619 * currently open. This powers the unread dot on the session list so the user can
620 * tell which conversations got a reply while they were elsewhere.
622 * Uses [flatMapLatest] so a server switch (which replaces the active connection)
623 * re-subscribes to the new event stream and drops the old one.
625 @OptIn(ExperimentalCoroutinesApi::class)
626 private fun observeMessageActivity() {
628 // Retry the observer if it fails for any reason (transient flow exception,
629 // cancellation due to scope issues). Without this, a single failure would
630 // permanently disable unread tracking and completion notifications.
632 runCatchingCancellable {
634 .flatMapLatest { conn ->
635 // Capture the originating connection so notification coroutines
636 // resolve titles and route permission responses against THIS
637 // server, not whichever happens to be active when they resume.
638 // A server switch during a listSessions() round-trip would
639 // otherwise resolve the title (or POST the permission reply)
640 // against the new server, where the session/permission doesn't
641 // exist — showing an opaque id or silently losing the reply.
642 conn?.events?.events?.map { conn to it } ?: emptyFlow()
644 .collect { (conn, event) ->
645 // A permission request needs the user's approval. Post a heads-up
646 // notification (with Allow/Reject actions) unless the user is actively
647 // viewing this session, where the in-app permission dialog handles it.
648 if (event is PermissionUpdated) {
649 val psid = event.properties.sessionID.takeIf { it.isNotBlank() } ?: return@collect
650 if (!isActivelyViewing(psid)) appScope.launch { notifyPermission(event.properties, conn) }
653 // The permission was answered (in-app, from the notification, or
654 // auto-resolved): clear its heads-up notification.
655 if (event is PermissionReplied) {
656 val psid = event.properties.sessionID?.takeIf { it.isNotBlank() } ?: return@collect
657 appContext?.let { SessionNotifications.cancelPermission(it, psid) }
660 // A run failed. Clear its active-run tracking (an error isn't always
661 // followed by a SessionIdle) and notify unless it's being viewed.
662 if (event is SessionError) {
663 val esid = event.properties.sessionID ?: return@collect
664 val wasRunning = synchronized(activeRuns) {
665 activeRuns.remove(esid).also { removed ->
666 if (removed) publishRunState()
669 if (wasRunning && !isActivelyViewing(esid)) {
670 appScope.launch { notifySessionError(esid, conn.repository, conn.profile.id) }
674 // SessionIdle is not "message activity" (don't badge as unread)
675 // but signals a run finished — fire a completion notification if the
676 // session isn't currently being viewed and was actively streaming.
677 if (event is SessionIdle) {
678 val idleSid = event.properties.sessionID ?: return@collect
679 // Clear the run state unconditionally — even for the session on
680 // screen. Gating the removal on `idleSid != currentSession` (as
681 // this once did) leaks the id in activeRuns and pins anyRunActive
682 // true forever whenever a run finishes while being viewed. Only
683 // the completion *notification* is suppressed for the viewed one.
684 val wasRunning = synchronized(activeRuns) {
685 activeRuns.remove(idleSid).also { removed ->
686 if (removed) publishRunState()
689 // Clear the progress text when the run finishes.
690 _runProgressText.value = null
691 if (wasRunning && !isActivelyViewing(idleSid)) {
692 // Fire this off the collector's coroutine: notifySessionCompleted
693 // does a network listSessions() to resolve the title, and blocking
694 // the shared-event collector here would stall the SharedFlow,
695 // overflowing its DROP_OLDEST buffer and silently dropping live
696 // parts/updates for every other subscriber (e.g. the message reducer).
697 appScope.launch { notifySessionCompleted(idleSid, conn.repository, conn.profile.id) }
701 // Track the agent's task-plan progress (from todowrite tool calls) so the
702 // foreground-service notification can show "Step 2 of 5" instead of just a
703 // chronometer. The todowrite tool sends the full plan on every call, so the
704 // latest call's input is a complete snapshot.
705 if (event is MessagePartUpdated) {
706 val part = event.properties.part
707 if (part is ToolPart && part.tool.equals(TODO_WRITE_TOOL, ignoreCase = true)) {
708 val todos = parseTodos(part.state.inputElement())
709 if (todos.isNotEmpty()) {
710 val completed = todos.count { it.statusEnum() == TodoStatus.COMPLETED }
711 _runProgressText.value = appContext?.getString(
712 soy.iko.opencode.R.string.notif_running_progress,
719 val sid = sessionOf(event) ?: return@collect
720 // Re-read currentSession here, right before mutating the badge state:
721 // the user may have opened this session between when this event was
722 // enqueued and now (setCurrentSession clears its badge and dedup set).
723 // A stale earlier check would race a badge — and a dedup entry — back
724 // in for the session that's actually on screen.
725 if (sid != _currentSession.value) {
726 // Skip the unread badge for muted sessions — the count is still tracked
727 // in unreadMessageIds (so unmuting can't re-count the same messages),
728 // but the visible badge is suppressed so a muted conversation doesn't
729 // interrupt. The completion notification is also gated on isSessionMuted
730 // (see notifySessionCompleted).
731 if (sid !in _mutedSessions.value) {
732 // Increment the unread count once per distinct *message*, not per
733 // event: a single reply emits many message.updated / message.part.
734 // updated events (one per streamed token, plus cost/token refreshes),
735 // which would otherwise inflate the badge into the dozens/hundreds.
736 val messageId = messageIdOfEvent(event)
737 val counted = unreadMessageIds.getOrPut(sid) {
738 java.util.Collections.synchronizedSet(mutableSetOf())
740 if (messageId == null || counted.add(messageId)) {
741 _unread.update { current ->
742 // Guard again inside the atomic update lambda so a retry
743 // (or an open that landed just now) can't reintroduce the
744 // badge for the session on screen.
745 if (sid == _currentSession.value) current
746 else current + (sid to (current[sid] ?: 0) + 1)
751 // Track sessions actively streaming so we know which idle events
752 // represent a finished run worth notifying about.
753 if (isLiveRunActivity(event)) {
754 // The add-under-lock is required both for thread-safety and for the
755 // access-order refresh (keeps a continuously-streaming session from
756 // being evicted), so it runs per event.
757 synchronized(activeRuns) {
759 // Derive the flag from the set state *inside* the same lock that
760 // guards the set, so a concurrent reconnect sweep (which clears
761 // both under this lock) can't interleave and leave the flag pinned
762 // true with an empty set — a stuck foreground service + "working…".
766 } }.onFailure { Log.w("AppContainer", "Message activity observer failed, will retry: ${safeExceptionSummary(it)}") }
768 delay(NetworkConfig.observerRetryDelayMs)
773 /** True if [event] is live streaming activity worth tracking as an active run. A
774 * trailing message.updated that lands *after* the run finished (final cost/token totals,
775 * carrying a completion time) is NOT live streaming: with no further SessionIdle to
776 * follow it, re-adding the session would pin anyRunActive true indefinitely. */
777 private fun isLiveRunActivity(event: BusEvent): Boolean {
778 if (event !is MessagePartUpdated && event !is MessageUpdated) return false
779 // For MessageUpdated, only an in-flight *assistant* message counts as live activity.
780 // A UserMessage or UnknownMessage update (e.g. a metadata refresh of an existing user
781 // message) would otherwise add the session to activeRuns with no following SessionIdle
782 // to clear it — pinning anyRunActive true and the foreground notification indefinitely.
783 // Mirrors SessionRepository.isRunActivity's guard exactly.
784 if (event is MessageUpdated) {
785 val info = event.properties.info
786 return info is AssistantMessage && !info.isComplete && info.error == null
788 // A step-finish part is the trailing completion marker for a run (it carries the
789 // final cost/token totals) and arrives after the last stream delta. Like a completed
790 // message.updated, treating it as live would re-add the session to activeRuns with no
791 // following SessionIdle to clear it — pinning anyRunActive true indefinitely when such
792 // a part is replayed/reordered after the run already went idle.
793 val part = (event as MessagePartUpdated).properties.part
794 return part !is StepFinishPart
797 /** Snapshot [activeRuns] into [_runningSessionIds] and derive [_anyRunActive] from it.
798 * MUST be called under the [activeRuns] monitor so the published flag and snapshot can't
799 * diverge from the set a concurrent SSE add()/remove() is mutating — the same invariant
800 * [connectLocked]/[teardownActiveLocked] rely on. */
801 private fun publishRunState() {
802 _anyRunActive.value = activeRuns.isNotEmpty()
803 _runningSessionIds.value = activeRuns.toSet()
806 /** Session ids currently streaming an assistant run (best-effort, in-process). Backed by
807 * an access-order [java.util.LinkedHashMap] so re-adding an already-present id on each
808 * streaming event refreshes its recency (a plain LinkedHashSet.add is a no-op that does
809 * NOT reorder). Entries are drained by [SessionIdle] (and cleared wholesale on reconnect
810 * reconcile); we intentionally do NOT cap the set, because evicting the eldest would drop
811 * its completion notification when its SessionIdle eventually arrives — a busy server
812 * with many concurrent runs is still bounded by the server's own concurrency, so growth
813 * is not unbounded in practice. */
814 private val activeRuns: MutableSet<String> = java.util.Collections.synchronizedSet(
815 java.util.Collections.newSetFromMap(java.util.LinkedHashMap<String, Boolean>(16, 0.75f, true)),
818 /** Per non-viewed session, the set of message ids already reflected in its unread
819 * count. De-duplicates the many streaming events that share one message id so the
820 * badge counts messages, not deltas. Cleared when a session is viewed or on connect. */
821 private val unreadMessageIds = java.util.concurrent.ConcurrentHashMap<String, MutableSet<String>>()
823 /** Guards connect/disconnect so concurrent callers can't leak an old connection. */
824 private val connectionMutex = Mutex()
826 /** Only persist the `lastUsed` timestamp if it's older than this, to avoid
827 * a DataStore + encrypted-prefs write on every rapid reconnect. */
828 private companion object {
829 const val LAST_USED_SAVE_THRESHOLD_MS = 60_000L
832 /** Resolve the human-readable title for [sessionId] (best-effort), falling back to the id.
833 * Uses [originRepo] — the repository of the connection that emitted the event — so a
834 * server switch during the listSessions() round-trip doesn't resolve the title against
835 * an unrelated server (showing an opaque id, or worse, a colliding session's title). */
836 private suspend fun resolveSessionTitle(sessionId: String, originRepo: SessionRepository): String =
837 runCatchingCancellable {
838 originRepo.listSessions()?.firstOrNull { it.id == sessionId }?.displayTitle
839 }.getOrNull() ?: sessionId
841 /** Resolve the title for [sessionId] (best-effort) and post a completion notification.
842 * [originProfileId] is the id of the server that ran the session; it's embedded in the
843 * inline-reply action so a follow-up routes back to THIS server even after the user switches
844 * connections (mirrors [notifyPermission]). */
845 private suspend fun notifySessionCompleted(sessionId: String, originRepo: SessionRepository, originProfileId: String) {
846 val title = resolveSessionTitle(sessionId, originRepo)
847 // Re-check viewing right before posting: the call site checks !isActivelyViewing(sid)
848 // at event time, but resolveSessionTitle() suspends on a listSessions() round-trip.
849 // If the user opened the session in that gap, setCurrentSession already cancelled the
850 // prior notification — posting a fresh one here would leave a "session ready" heads-up
851 // lingering while they're already looking at the finished run.
852 if (isActivelyViewing(sessionId)) return
853 // Suppress the completion notification for muted sessions — the whole point of muting
854 // is that a noisy conversation doesn't interrupt.
855 if (sessionId in _mutedSessions.value) return
856 appContext?.let { SessionNotifications.postCompleted(it, sessionId, title, originProfileId) }
859 /** Post a heads-up permission notification for [permission], resolving its session title.
860 * [originConn] is the connection that emitted the permission event — its profile id is
861 * embedded in the notification so the receiver routes the response back to THIS server. */
862 private suspend fun notifyPermission(permission: Permission, originConn: OpencodeConnection) {
863 val sessionId = permission.sessionID.takeIf { it.isNotBlank() } ?: return
864 val title = resolveSessionTitle(sessionId, originConn.repository)
865 // Re-check viewing just before posting — see notifySessionCompleted for the race.
866 if (isActivelyViewing(sessionId)) return
867 appContext?.let { SessionNotifications.postPermission(it, permission, title, originConn.profile.id) }
870 /** Post an error notification for a failed background run. [originProfileId] is embedded
871 * in the tap intent so the session opens under the server that ran it, not whichever
872 * is active when the user taps — mirroring [notifySessionCompleted]. */
873 private suspend fun notifySessionError(sessionId: String, originRepo: SessionRepository, originProfileId: String) {
874 val title = resolveSessionTitle(sessionId, originRepo)
875 // Re-check viewing just before posting — see notifySessionCompleted for the race.
876 if (isActivelyViewing(sessionId)) return
877 appContext?.let { SessionNotifications.postError(it, sessionId, title, originProfileId) }
880 /** Respond to a permission from a notification action (Allow once / Always / Reject),
881 * off any UI. Runs on the process-lived app scope; [onDone] fires when the call resolves
882 * so the receiver's goAsync() result can finish. Its Boolean is true only when a live
883 * connection existed AND respondPermission succeeded — the receiver uses it to decide
884 * whether to dismiss the notification. A missing connection makes the respond a no-op,
885 * so it must report failure (not success) rather than let the notification vanish with
886 * the tool left unanswered.
888 * [profileId] is the id of the server that posted the permission request (embedded in
889 * the notification's PendingIntent). If it doesn't match the currently active
890 * connection's profile, the response is NOT sent: posting it to a different server
891 * would 404 (treated as "already resolved" success by respondPermission), dismissing
892 * the notification while the original server's tool stays paused forever. Reporting
893 * failure instead leaves the notification up so the user can switch back and retry. */
894 open fun respondToPermissionFromNotification(
896 permissionId: String,
897 response: PermissionResponse,
899 onDone: (Boolean) -> Unit,
902 // Guarantee onDone (so the BroadcastReceiver's goAsync() finishes) even if the
903 // scope is cancelled mid-call (process shutdown cancels appScope first). Without
904 // this, a cancelled scope leaves pending.finish() never called → a receiver ANR
905 // and the permission response silently lost.
908 success = runCatchingCancellable {
909 val conn = activeConnection.value ?: return@runCatchingCancellable false
910 // Route to the originating server only. A mismatch means the user
911 // switched servers while the permission was pending; sending the reply
912 // to the wrong server would be silently swallowed as a 404-success.
913 if (profileId != null && conn.profile.id != profileId) return@runCatchingCancellable false
914 conn.api.respondPermission(sessionId, permissionId, response)
917 Log.w("AppContainer", "notif permission respond failed: ${safeExceptionSummary(it)}")
918 }.getOrDefault(false)
925 /** Send a follow-up prompt from a notification's inline reply, off any UI. Uses the
926 * server's default model/agent. Routes through the durable [outboxStore] — the same path
927 * the in-app composer uses ([soy.iko.opencode.ui.chat.ChatViewModel.enqueueOffline]) —
928 * rather than a direct repository send, so a reply composed while disconnected survives to
929 * be flushed on reconnect instead of being silently dropped. [flushOutbox] nudges an
930 * immediate send when a connection is live, so an online reply still goes out now. [onDone]
931 * reports whether the reply was durably enqueued so the receiver only dismisses the
932 * notification on success. */
933 open fun sendPromptFromNotification(sessionId: String, text: String, originProfileId: String?, onDone: (Boolean) -> Unit) {
935 // Guarantee onDone (so the BroadcastReceiver's goAsync() finishes) even if the
936 // scope is cancelled mid-call — see respondToPermissionFromNotification above.
939 // Attribute to the profile that POSTED the notification (embedded in the reply
940 // intent) so the follow-up is always enqueued against the server that owns
941 // [sessionId] — never whichever server happens to be active now, which would POST
942 // A's session id to B, 404, and silently drop the reply. Fall back to the active/
943 // most-recent profile only for a legacy intent that predates the embedded id.
944 val profileId = originProfileId
945 ?: activeConnection.value?.profile?.id
946 ?: runCatchingCancellable { profileStore.profiles.first().firstOrNull()?.id }.getOrNull()
947 enqueued = if (profileId == null) {
950 runCatchingCancellable {
953 id = java.util.UUID.randomUUID().toString(),
954 profileId = profileId,
955 sessionId = sessionId,
958 createdAt = System.currentTimeMillis(),
962 Log.w("AppContainer", "notif reply enqueue failed: ${safeExceptionSummary(it)}")
965 if (enqueued) flushOutbox()
972 /** Abort the active run for [sessionId] from a notification Stop action. Routes to the
973 * originating profile (or the active connection) so the abort reaches the server that
974 * owns the session. Runs on the app scope so it survives the BroadcastReceiver's lifetime. */
975 open fun abortRunFromNotification(sessionId: String, originProfileId: String?, onDone: (Boolean) -> Unit) {
979 val conn = activeConnection.value?.takeIf {
980 originProfileId == null || it.profile.id == originProfileId
981 } ?: activeConnection.value
983 success = runCatchingCancellable { conn.repository.abort(sessionId) }.isSuccess
992 * Retry the last user prompt for [sessionId] from an error notification's "Retry last"
993 * action. Fetches the session's messages, re-sends the most recent user prompt, and
994 * cancels the error notification on success. Routes to the originating profile's
995 * connection (falling back to the active one) so the send reaches the server that owns
996 * the session. A no-op (reported as false) when there's no connection or no prior prompt.
998 open fun retryLastFromNotification(sessionId: String, originProfileId: String?, onDone: (Boolean) -> Unit) {
1002 val conn = activeConnection.value?.takeIf {
1003 originProfileId == null || it.profile.id == originProfileId
1004 } ?: activeConnection.value
1006 success = runCatchingCancellable {
1007 val messages = conn.api.listMessages(sessionId)
1008 val lastPrompt = messages
1009 .lastOrNull { it.info is soy.iko.opencode.data.model.UserMessage }
1011 ?.filterIsInstance<soy.iko.opencode.data.model.TextPart>()
1012 ?.joinToString("\n\n") { it.text }
1013 ?.takeIf { it.isNotBlank() }
1014 if (lastPrompt != null) {
1015 // model=null lets the server reuse the session's current model; we
1016 // don't know the originally-selected model from a notification context.
1017 conn.repository.sendPrompt(sessionId, lastPrompt, model = null)
1030 /** Extract the session id an event pertains to, for message-activity events. */
1031 private fun sessionOf(event: BusEvent): String? = sessionOfEvent(event)
1034 * On cold start, transparently reconnect to the most recently used server so a
1035 * returning user lands in their session list instead of the empty server screen.
1037 private suspend fun autoConnect() {
1038 val recent = runCatchingCancellable { profileStore.profiles.first() }
1039 .getOrDefault(emptyList())
1040 .firstOrNull() ?: return
1041 // The profile read above suspends on DataStore. A manual connect() to a different
1042 // server during that window establishes a connection the user explicitly chose; we must
1043 // not clobber it by connecting to `recent` on top. connectIfIdle() checks "is anything
1044 // active?" and connects atomically under connectionMutex, returning null if the cold-start
1045 // slot was already taken — closing the TOCTOU window an unlocked null-check plus connect()'s
1046 // unconditional close would leave. (disconnectIf below similarly guards the failed-ping
1047 // teardown by identity.)
1048 _reconnecting.value = true
1049 var conn: OpencodeConnection? = null
1051 val ok = runCatchingCancellable {
1052 conn = connectIfIdle(recent) ?: return@runCatchingCancellable false
1055 }.getOrDefault(false)
1057 conn == null -> Unit // slot already taken by a manual connect — leave it be
1058 ok -> _autoConnectDone.value = true
1059 else -> disconnectIf(conn!!)
1062 // Reset in a finally so a CancellationException (e.g. scope cancelled during
1063 // shutdown) can't leave the flag pinned true — runCatchingCancellable rethrows it,
1064 // skipping the line below the try block.
1065 _reconnecting.value = false
1070 * When the device regains connectivity, nudge the active SSE stream to reconnect
1071 * immediately instead of waiting out the exponential backoff.
1073 private val networkCallback: ConnectivityManager.NetworkCallback? = registerNetworkMonitor()
1075 private fun registerNetworkMonitor(): ConnectivityManager.NetworkCallback? {
1076 if (appContext == null) return null
1077 val cm = appContext!!.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
1079 // Seed the online state from the current active network so the indicator is
1080 // correct on cold start rather than defaulting to "online" until a callback fires.
1081 _isOnline.value = runCatching { cm.activeNetwork != null }.getOrDefault(true)
1082 val callback = object : ConnectivityManager.NetworkCallback() {
1083 override fun onAvailable(network: Network) { _isOnline.value = true; activeConnection.value?.events?.triggerReconnect() }
1084 // A matched (internet-capable) network was lost. Don't eagerly flip offline: during a
1085 // Wi-Fi→cellular handoff onLost(WiFi) fires before onAvailable(cellular) and
1086 // cm.activeNetwork is transiently null in that window — flashing offline there would
1087 // skip outbox flushing and blink the indicator. Instead re-check activeNetwork after a
1088 // short grace period and only report offline if there's still no active network.
1090 // This MUST live in onLost, not onUnavailable: onUnavailable is delivered only for
1091 // requestNetwork() with a timeout, never for registerNetworkCallback() — relying on it
1092 // (as this once did) meant offline was never detected after the first onAvailable.
1093 override fun onLost(network: Network) {
1095 delay(NetworkConfig.networkOfflineGraceMs)
1096 _isOnline.value = runCatching { cm.activeNetwork != null }.getOrDefault(true)
1100 // Only match networks that actually provide internet — a capability-less request
1101 // fires onAvailable for transports that can't reach the server (and would trigger
1102 // spurious reconnects), so require NET_CAPABILITY_INTERNET.
1103 val request = NetworkRequest.Builder()
1104 .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
1106 runCatching { cm.registerNetworkCallback(request, callback) }
1107 .onFailure { Log.w("AppContainer", "Failed to register network callback", it) }
1111 /** Open (or re-open) a connection to [profile], replacing any current one.
1112 * Only persists the updated `lastUsed` timestamp if it's stale (older than a
1113 * minute), so rapid reconnect storms don't each trigger a DataStore write. */
1114 open suspend fun connect(profile: ServerProfile): OpencodeConnection =
1115 connectionMutex.withLock { connectLocked(profile) }
1117 /** Connect to the profile with [profileId], looking it up from [profileStore]. Returns
1118 * true on success, false if the profile wasn't found or connect threw. Used by the
1119 * notification body-tap path to route a session open back to the originating server. */
1120 open suspend fun connectByProfileId(profileId: String): Boolean {
1121 val profile = profileStore.profiles.first().find { it.id == profileId } ?: return false
1122 return runCatchingCancellable { connect(profile) }.isSuccess
1125 /** Connect to [profile] only if nothing is currently active, checking-and-connecting
1126 * atomically under [connectionMutex]. Returns null (touching nothing) when a connection is
1127 * already active, so autoConnect() can't clobber a manual connect() the user made during its
1128 * suspending profile read — closing the TOCTOU window an unlocked null-check would leave. */
1129 private suspend fun connectIfIdle(profile: ServerProfile): OpencodeConnection? =
1130 connectionMutex.withLock {
1131 if (_activeConnection.value != null) null else connectLocked(profile)
1134 /** Build and install a connection to [profile], replacing any current one. Caller MUST hold
1135 * [connectionMutex]. */
1136 private suspend fun connectLocked(profile: ServerProfile): OpencodeConnection {
1137 // Capture the previous profile id before closing, so we can preserve unread badges
1138 // when reconnecting to the same server — a brief network blip or manual reconnect
1139 // shouldn't erase "5 unread". Only a genuine server switch clears them.
1140 val previousProfileId = _activeConnection.value?.profile?.id
1141 _activeConnection.value?.close()
1142 _activeConnection.value = null
1143 // Mutate the run set and its derived flag together under activeRuns' monitor: the SSE
1144 // event handler adds ids under the same lock, so an unlocked clear() here could
1145 // interleave with an add()+flag=true and leave _anyRunActive pinned true with a torn-
1146 // down connection and no SessionIdle to clear it — a stuck foreground service.
1147 synchronized(activeRuns) {
1151 if (previousProfileId != profile.id) {
1152 _unread.value = emptyMap()
1153 unreadMessageIds.clear()
1155 val now = System.currentTimeMillis()
1156 val resolved = profileStore.resolve(profile)
1157 val needsSave = (now - resolved.lastUsed) > LAST_USED_SAVE_THRESHOLD_MS
1158 val finalProfile = if (needsSave) resolved.copy(lastUsed = now) else resolved
1159 if (needsSave) profileStore.save(finalProfile)
1160 return OpencodeConnection(finalProfile, messageCacheStore).also { _activeConnection.value = it }
1163 open suspend fun disconnect() {
1164 // Acquire the connection mutex so disconnect is serialized with connect().
1165 // Previously this was fire-and-forget (launch on appScope), which raced with
1166 // a subsequent connect(): connect() could acquire the mutex, create a new
1167 // connection, and then the pending disconnect() would close it.
1168 connectionMutex.withLock { teardownActiveLocked() }
1171 /** Tear down [expected] only if it is still the active connection, with the identity
1172 * check performed *under* the mutex. A caller that checks `activeConnection.value === x`
1173 * itself and then calls [disconnect] has a TOCTOU race: a concurrent connect() to a
1174 * different server can replace the active connection in the gap, and [disconnect] would
1175 * then close that new connection. autoConnect()'s failed-ping teardown uses this. */
1176 private suspend fun disconnectIf(expected: OpencodeConnection) {
1177 connectionMutex.withLock {
1178 if (_activeConnection.value === expected) teardownActiveLocked()
1182 /** Close and reset the active connection. Caller MUST hold [connectionMutex]. */
1183 private suspend fun teardownActiveLocked() {
1184 _activeConnection.value?.close()
1185 _activeConnection.value = null
1186 // Reset run/unread state on disconnect too. With no SSE stream there's no
1187 // SessionIdle to drain activeRuns, so without this the working indicator
1188 // (anyRunActive) and unread badges would stay pinned until the next connect().
1189 // Mutate under activeRuns' monitor (see connect()) so a concurrent SSE add() can't
1190 // re-pin _anyRunActive after the clear.
1191 synchronized(activeRuns) {
1195 _unread.value = emptyMap()
1196 unreadMessageIds.clear()
1199 /** Deferred profile deletions keyed by id, run on the process-lived [appScope] (not a
1200 * ViewModel scope) so navigating away during the undo window still commits the delete
1201 * instead of silently cancelling it — mirroring [CrashLogger.scheduleDelete]. */
1202 private val pendingProfileDeletes = java.util.concurrent.ConcurrentHashMap<String, Job>()
1205 * Schedule the profile [id] for deletion after [delayMs], cancellable via
1206 * [cancelProfileDelete] (the Undo action). [onDeleted]/[onError] run on [appScope]
1207 * after the delete resolves so the caller can clear its optimistic-hide state. A
1208 * repeated schedule for the same id replaces the prior timer.
1210 open fun scheduleProfileDelete(
1213 onDeleted: () -> Unit,
1214 onError: (Throwable) -> Unit,
1216 val job = appScope.launch {
1218 // Claim the delete by removing our own entry BEFORE committing it — the atomic
1219 // remove is the sync point with cancelProfileDelete. Losing the race (an undo
1220 // removed us first) means we skip, honouring cancel's `true`; deleting first
1221 // would let a cancel land in the gap and falsely report undo of a gone profile.
1222 // remove(id, thisJob) also no-ops for a stale job after a reschedule.
1223 val claimed = coroutineContext[Job]?.let { pendingProfileDeletes.remove(id, it) } == true
1224 if (!claimed) return@launch
1225 runCatchingCancellable { profileStore.delete(id) }
1226 .onSuccess { onDeleted() }
1227 .onFailure { onError(it) }
1229 pendingProfileDeletes[id]?.cancel()
1230 pendingProfileDeletes[id] = job
1233 /** Cancel a pending deferred profile delete (the Undo action). Returns true if it was
1234 * still pending (undo succeeded), false if the delete had already fired. */
1235 open fun cancelProfileDelete(id: String): Boolean {
1236 val job = pendingProfileDeletes.remove(id) ?: return false
1241 /** Deferred session deletions keyed by id, run on the process-lived [appScope] (not a
1242 * ViewModel scope) so navigating away during the undo window still commits the delete
1243 * instead of silently cancelling it (which would leave the session deleted-in-UI but
1244 * alive on the server, reappearing on the next refresh). */
1245 private val pendingSessionDeletes = java.util.concurrent.ConcurrentHashMap<String, Job>()
1247 /** One-shot events carrying the id of a session scheduled for deferred deletion from
1248 * outside the session list (e.g. from the chat screen's Delete). The SessionListScreen
1249 * collects these to show an Undo snackbar, mirroring its own VM-level undo events. */
1250 private val _externalSessionUndoEvents = MutableSharedFlow<String>(
1251 extraBufferCapacity = NetworkConfig.snackbarEventBufferCapacity,
1252 onBufferOverflow = kotlinx.coroutines.channels.BufferOverflow.DROP_OLDEST,
1254 val externalSessionUndoEvents: SharedFlow<String> = _externalSessionUndoEvents.asSharedFlow()
1256 /** Emit an undo event for a session delete initiated outside the session list (e.g.
1257 * from the chat screen). The SessionListScreen collects these to show an Undo snackbar. */
1258 fun emitExternalSessionUndo(sessionId: String) {
1259 _externalSessionUndoEvents.tryEmit(sessionId)
1263 * Schedule session [id] for deletion after [delayMs], cancellable via [cancelSessionDelete]
1264 * (the Undo action). Also clears the session's stored draft on a successful delete.
1265 * [onDeleted]/[onError] run on [appScope] after the delete resolves so the caller can update
1266 * its optimistic-hide state. A repeated schedule for the same id replaces the prior timer.
1268 * The owning connection is captured now, at schedule time — NOT re-resolved when the timer
1269 * fires. Otherwise a server switch during the undo window would send the delete to whichever
1270 * server happens to be active at fire time: the original session would never be deleted (its
1271 * row reappears on the next refresh) and, in the rare id-collision case, an unrelated session
1272 * on the new server could be deleted instead. A disconnect closes the captured connection's
1273 * HTTP client, so a delete deferred across a disconnect fails and is surfaced via [onError]
1274 * (the session cannot be deleted on a server we're no longer connected to). No-ops (as an
1275 * error) if there is no active connection at schedule time.
1277 open fun scheduleSessionDelete(
1280 onDeleted: () -> Unit,
1281 onError: (Throwable) -> Unit,
1283 val conn = _activeConnection.value
1284 val job = appScope.launch {
1286 // Claim the delete by removing our own entry BEFORE committing it — the atomic
1287 // remove is the sync point with cancelSessionDelete. Losing the race (an undo
1288 // removed us first) means we skip, honouring cancel's `true`; deleting first
1289 // would let a cancel land in the gap and falsely report undo of a gone session.
1290 // remove(id, thisJob) also no-ops for a stale job after a reschedule.
1291 val claimed = coroutineContext[Job]?.let { pendingSessionDeletes.remove(id, it) } == true
1292 if (!claimed) return@launch
1294 onError(IllegalStateException("No active connection"))
1296 runCatchingCancellable { conn.repository.deleteSession(id) }
1298 draftStore.remove(id)
1299 attachmentDraftStore.remove(id)
1300 sessionPrefsStore.forget(id)
1301 messageCacheStore.remove(conn.profile.id, id)
1302 outboxStore.removeForSession(id)
1306 .onFailure { onError(it) }
1309 pendingSessionDeletes[id]?.cancel()
1310 pendingSessionDeletes[id] = job
1313 /** Cancel a pending deferred session delete (the Undo action). Returns true if it was
1314 * still pending (undo succeeded), false if the delete had already fired. */
1315 open fun cancelSessionDelete(id: String): Boolean {
1316 val job = pendingSessionDeletes.remove(id) ?: return false
1322 * Probe a server URL (without credentials) to check reachability and detect whether
1323 * the server requires authentication. Builds a short-lived HTTP client with the same
1324 * base URL normalization as [connect] but no auth, calls the health endpoint, and
1325 * classifies the outcome:
1327 * - 2xx → [ProbeResult.Reachable] (no auth needed)
1328 * - 401/403 → [ProbeResult.NeedsAuth]
1329 * - anything else → [ProbeResult.Unreachable] with a user-facing message
1331 * The probe client is always closed afterwards so no resources linger. This does not
1332 * touch the active connection or the profile store.
1334 open suspend fun probeServer(baseUrl: String, requireHttps: Boolean, certPin: String?): ProbeResult {
1335 return probeWithProfile(ServerProfile(
1338 baseUrl = baseUrl.trim(),
1341 // Probe the same effective URL save()/connect() will use: HttpClientFactory upgrades
1342 // http->https and installs the cert pinner when these are set, so a probe without them
1343 // would validate a plain-http endpoint the real connection never talks to.
1344 requireHttps = requireHttps,
1350 * Probe a server URL *with* credentials to validate them. Returns true if the
1351 * server accepts the credentials (ping succeeds), false on 401/403, and throws
1352 * on other failures (unreachable, timeout) so the caller can surface a friendly
1353 * message. Mirrors [probeServer]'s short-lived-client pattern.
1355 open suspend fun probeWithCredentials(
1359 requireHttps: Boolean,
1362 val probeProfile = ServerProfile(
1365 baseUrl = baseUrl.trim(),
1366 username = username.trim().takeIf { it.isNotBlank() },
1367 password = password.trim().takeIf { it.isNotEmpty() },
1368 // Match the effective URL/pinning save()/connect() will use (see probeServer).
1369 requireHttps = requireHttps,
1372 val client = HttpClientFactory.create(probeProfile)
1374 val api = OpencodeApiClient(client)
1377 } catch (c: kotlinx.coroutines.CancellationException) {
1379 } catch (e: Exception) {
1380 val status = responseStatusCode(e)
1381 if (status == 401 || status == 403) false else throw e
1383 runCatching { client.close() }
1387 private suspend fun probeWithProfile(profile: ServerProfile): ProbeResult {
1388 val client = HttpClientFactory.create(profile)
1390 val api = OpencodeApiClient(client)
1392 ProbeResult.Reachable
1393 } catch (c: kotlinx.coroutines.CancellationException) {
1395 } catch (e: Exception) {
1396 val status = responseStatusCode(e)
1397 if (status == 401 || status == 403) {
1398 ProbeResult.NeedsAuth
1400 ProbeResult.Unreachable(friendlyErrorFor(e, profile.baseUrl))
1403 runCatching { client.close() }
1407 // Declared last so every property above is fully initialized before these observers start.
1408 // They only launch coroutines into appScope (Dispatchers.IO) — nothing here runs
1409 // synchronously during construction — so placing init at the end removes any dependence on
1410 // the textual declaration order of the fields those observers reference.
1413 NotificationChannels.create(appContext!!)
1414 observeMessageActivity()
1415 observeRunReconcileOnReconnect()
1416 observeRunForegroundService()
1417 // Load the persisted muted-sessions set so isSessionMuted() reflects the choice
1418 // immediately on cold start (before any session list composition).
1420 runCatchingCancellable {
1421 sessionPrefsStore.muted.collect { ids -> _mutedSessions.value = ids }
1424 // Flush right after load() populates the queue. observeOutbox()'s combine is keyed
1425 // on activeConnection/isOnline/_outboxFlushTrigger — NOT on outboxStore.messages — so
1426 // if connect() sets activeConnection before load() finishes reading a (possibly
1427 // multi-MB) outbox.json, the connect-triggered flush sees an empty queue and nothing
1428 // re-triggers it. Nudging the flush here guarantees a loaded queue is sent.
1430 runCatchingCancellable { outboxStore.load() }
1434 appScope.launch { autoConnect() }
1440 * Extract the session id a message-activity event pertains to, or null for events that
1441 * don't describe message activity. This drives the unread badge: any session that
1442 * receives activity while it isn't the currently-viewed one gets badged. Extracted as
1443 * a top-level `internal` function so the rule is unit-testable without an Android
1444 * [Context] (the surrounding [AppContainer] needs one).
1446internal fun sessionOfEvent(event: BusEvent): String? = when (event) {
1447 is MessageUpdated -> event.properties.info.sessionID
1448 is MessagePartUpdated -> event.properties.part.sessionID ?: event.properties.sessionID
1453 * The message id a message-activity event pertains to, used to de-duplicate the unread
1454 * badge so a reply's many streaming events count once. Null for events that carry no
1455 * message id (those fall back to being counted individually).
1457internal fun messageIdOfEvent(event: BusEvent): String? = when (event) {
1458 is MessageUpdated -> event.properties.info.id
1459 is MessagePartUpdated -> event.properties.messageID ?: event.properties.part.messageID
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list