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.SessionPrefsStore
36import soy.iko.opencode.data.repo.SessionRepository
37import soy.iko.opencode.data.repo.ProfileStore
38import soy.iko.opencode.data.repo.SettingsStore
39import soy.iko.opencode.data.repo.classifyError
40import soy.iko.opencode.data.repo.responseStatusCode
41import soy.iko.opencode.data.network.EventStreamClient
42import soy.iko.opencode.data.network.HttpClientFactory
43import soy.iko.opencode.data.network.NetworkConfig
44import soy.iko.opencode.data.network.OpencodeApiClient
45import soy.iko.opencode.notification.NotificationChannels
46import soy.iko.opencode.notification.RunForegroundService
47import soy.iko.opencode.util.safeExceptionSummary
48import soy.iko.opencode.notification.SessionNotifications
49import soy.iko.opencode.R
50import soy.iko.opencode.util.runCatchingCancellable
51import kotlinx.coroutines.CoroutineScope
52import kotlinx.coroutines.Dispatchers
53import kotlinx.coroutines.ExperimentalCoroutinesApi
54import kotlinx.coroutines.SupervisorJob
55import kotlinx.coroutines.cancel
56import kotlinx.coroutines.delay
57import kotlinx.coroutines.flow.MutableSharedFlow
58import kotlinx.coroutines.flow.MutableStateFlow
59import kotlinx.coroutines.flow.SharedFlow
60import kotlinx.coroutines.flow.StateFlow
61import kotlinx.coroutines.flow.asSharedFlow
62import kotlinx.coroutines.flow.asStateFlow
63import kotlinx.coroutines.flow.collectLatest
64import kotlinx.coroutines.flow.combine
65import kotlinx.coroutines.flow.debounce
66import kotlinx.coroutines.flow.distinctUntilChanged
67import kotlinx.coroutines.flow.drop
68import kotlinx.coroutines.flow.emptyFlow
69import kotlinx.coroutines.flow.first
70import kotlinx.coroutines.Job
71import kotlinx.coroutines.flow.flatMapLatest
72import kotlinx.coroutines.flow.map
73import kotlinx.coroutines.flow.update
74import kotlinx.coroutines.isActive
75import kotlinx.coroutines.launch
76import kotlinx.coroutines.sync.Mutex
77import kotlinx.coroutines.sync.withLock
78import kotlinx.coroutines.withTimeoutOrNull
79import java.util.concurrent.atomic.AtomicBoolean
82 * Result of probing a server URL to check connectivity and whether authentication
83 * is required, without persisting a profile or opening a long-lived connection.
85sealed class ProbeResult {
86 /** Server is reachable and accepts unauthenticated requests. */
87 object Reachable : ProbeResult()
89 /** Server is reachable but rejected the probe with an authentication error. */
90 object NeedsAuth : ProbeResult()
92 /** Server could not be reached (host unresolved, connection refused, timeout, etc.). */
93 data class Unreachable(val error: String) : ProbeResult()
97 * Hand-written service locator held by [soy.iko.opencode.OpencodeApp]. Owns the
98 * process-wide singletons and the currently active [OpencodeConnection].
100open class AppContainer private constructor(
101 private val appContext: Context?,
102 private val skipInit: Boolean,
103 @Suppress("unused") private val testMode: Boolean,
105 constructor(context: Context) : this(context.applicationContext, false, false)
106 protected constructor() : this(null, true, true)
108 private val appScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
109 open val profileStore: ProfileStore by lazy { ProfileStore(appContext!!) }
110 open val settingsStore: SettingsStore by lazy { SettingsStore(appContext!!) }
111 open val draftStore: DraftStore by lazy { DraftStore(appContext!!, appScope) }
112 open val attachmentDraftStore: AttachmentDraftStore by lazy { AttachmentDraftStore(appContext!!) }
113 open val sessionPrefsStore: SessionPrefsStore by lazy { SessionPrefsStore(appContext!!) }
114 open val messageCacheStore: MessageCacheStore by lazy { MessageCacheStore(appContext!!) }
115 open val outboxStore: OutboxStore by lazy { OutboxStore(appContext!!) }
116 open val backupManager: BackupManager by lazy { BackupManager(profileStore, settingsStore, sessionPrefsStore) }
118 /** True while the outbox is actively flushing queued messages to the server, so the UI
119 * can show a "sending queued messages…" indicator. */
120 private val _outboxSending = MutableStateFlow(false)
121 open val outboxSending: StateFlow<Boolean> = _outboxSending.asStateFlow()
123 /** Bumped to nudge an immediate outbox flush (after enqueue, or a manual "Send now"),
124 * independent of connection/online changes. */
125 private val _outboxFlushTrigger = MutableStateFlow(0)
126 open fun flushOutbox() { _outboxFlushTrigger.update { it + 1 } }
128 private val _activeConnection = MutableStateFlow<OpencodeConnection?>(null)
129 open val activeConnection: StateFlow<OpencodeConnection?> = _activeConnection.asStateFlow()
131 /** True while a background auto-reconnect to the most-recent server is in flight. */
132 private val _reconnecting = MutableStateFlow(false)
133 open val reconnecting: StateFlow<Boolean> = _reconnecting.asStateFlow()
136 * Set once a startup auto-reconnect succeeds. The server list collects this and
137 * (once) navigates straight to the session list so a returning user skips it.
138 * [consumeAutoConnect] guards against re-firing on screen re-entry.
140 private val _autoConnectDone = MutableStateFlow(false)
141 open val autoConnectDone: StateFlow<Boolean> = _autoConnectDone.asStateFlow()
142 private val autoConnectConsumed = AtomicBoolean(false)
143 open fun consumeAutoConnect(): Boolean = autoConnectConsumed.compareAndSet(false, true)
145 /** Text shared into the app (ACTION_SEND), prefilled into a session draft. */
146 private val _pendingShare = MutableStateFlow<String?>(null)
147 open val pendingShare: StateFlow<String?> = _pendingShare.asStateFlow()
148 open fun setPendingShare(text: String) { _pendingShare.value = text }
149 open fun consumePendingShare(): String? {
150 val current = _pendingShare.value ?: return null
151 return if (_pendingShare.compareAndSet(current, null)) current else null
154 /** Image/file Uris (as strings) shared into the app, staged as attachments in the next
155 * opened session. Consumed once by the first ChatScreen to compose after the share. */
156 private val _pendingSharedMedia = MutableStateFlow<List<String>>(emptyList())
157 open val pendingSharedMedia: StateFlow<List<String>> = _pendingSharedMedia.asStateFlow()
158 open fun setPendingSharedMedia(uris: List<String>) { _pendingSharedMedia.value = uris }
159 open fun consumePendingSharedMedia(): List<String> {
160 val current = _pendingSharedMedia.value
161 if (current.isEmpty()) return emptyList()
162 return if (_pendingSharedMedia.compareAndSet(current, emptyList())) current else emptyList()
166 * Set from an external "New session" trigger (launcher shortcut / QS tile). The nav host
167 * consumes it once connected, creating a fresh session and opening it. Uses a counter
168 * (not a boolean) so re-tapping the shortcut while a prior request is still pending
169 * increments the value and re-fires the LaunchedEffect — a boolean StateFlow wouldn't
170 * emit when set to the same value, making the request unrecoverable by re-tapping.
172 private val _pendingNewSession = MutableStateFlow(0)
173 open val pendingNewSession: StateFlow<Int> = _pendingNewSession.asStateFlow()
174 open fun requestNewSession() { _pendingNewSession.update { it + 1 } }
175 open fun consumePendingNewSession(): Boolean {
177 val current = _pendingNewSession.value
178 if (current == 0) return false
179 if (_pendingNewSession.compareAndSet(current, current - 1)) return true
183 /** A one-shot signal to navigate to the Diagnostics screen (e.g. from the crash-relaunch
184 * prompt). Consumed by OpencodeApp's NavHost. */
185 private val _pendingDiagnostics = MutableStateFlow(false)
186 open val pendingDiagnostics: StateFlow<Boolean> = _pendingDiagnostics.asStateFlow()
187 open fun requestDiagnostics() { _pendingDiagnostics.value = true }
188 open fun consumePendingDiagnostics(): Boolean = _pendingDiagnostics.compareAndSet(true, false)
191 * A session id to open from an external trigger (a notification tap or a deep link),
192 * paired with the originating server's profile id (if known) so a tap after the user
193 * has switched servers routes back to the server that ran the session. The nav host
194 * consumes it once a connection is active (switching to [profileId] first if needed).
196 data class PendingOpenSession(val sessionId: String, val profileId: String?)
198 private val _pendingOpenSession = MutableStateFlow<PendingOpenSession?>(null)
199 open val pendingOpenSession: StateFlow<PendingOpenSession?> = _pendingOpenSession.asStateFlow()
200 open fun requestOpenSession(id: String, profileId: String? = null) {
201 _pendingOpenSession.value = PendingOpenSession(id, profileId)
203 open fun consumePendingOpenSession(): PendingOpenSession? {
204 val current = _pendingOpenSession.value ?: return null
205 return if (_pendingOpenSession.compareAndSet(current, null)) current else null
209 * A file path requested to be opened (via deep link `opencode://file/{path}` or future
210 * entry points), consumed by the NavHost to navigate to the file viewer. Mirrors the
211 * session open-signal pattern: a StateFlow the NavHost observes, drained atomically.
213 private val _pendingOpenFile = MutableStateFlow<String?>(null)
214 open val pendingOpenFile: StateFlow<String?> = _pendingOpenFile.asStateFlow()
215 open fun requestOpenFile(path: String) { _pendingOpenFile.value = path }
216 open fun consumePendingOpenFile(): String? {
217 val current = _pendingOpenFile.value ?: return null
218 return if (_pendingOpenFile.compareAndSet(current, null)) current else null
222 * The session the user is currently viewing (or null when not in a chat). Drives the
223 * unread tracker: messages arriving for any *other* session mark it unread, and
224 * opening a session clears its unread state.
226 private val _currentSession = MutableStateFlow<String?>(null)
227 open val currentSession: StateFlow<String?> = _currentSession.asStateFlow()
229 /** Session ids that received activity while not being viewed, mapped to the count
230 * of unread message events. A count (not just a presence set) lets the session
231 * list badge show "3 unread" instead of a bare dot, so the user can tell whether
232 * one reply or a whole burst arrived. */
233 private val _unread = MutableStateFlow<Map<String, Int>>(emptyMap())
234 open val unread: StateFlow<Map<String, Int>> = _unread.asStateFlow()
236 /** True when any assistant run is actively streaming across all sessions. Drives the
237 * disconnect confirmation on the session list (disconnecting mid-run kills it). */
238 private val _anyRunActive = MutableStateFlow(false)
239 open val anyRunActive: StateFlow<Boolean> = _anyRunActive.asStateFlow()
241 /** Latest task-plan progress text for the active run (e.g. "Step 2 of 5"), surfaced in
242 * the foreground-service notification so the user can see what the agent is working on
243 * without opening the app. Null when no plan is available or no run is active. */
244 private val _runProgressText = MutableStateFlow<String?>(null)
245 open val runProgressText: StateFlow<String?> = _runProgressText.asStateFlow()
247 /** Whether the device currently has network connectivity. Distinct from the SSE
248 * connection state so the UI can tell "you're offline" (device) from "server
249 * unreachable" (credentials/host). Updated by the [networkCallback]. */
250 private val _isOnline = MutableStateFlow(true)
251 open val isOnline: StateFlow<Boolean> = _isOnline.asStateFlow()
253 /** Whether a foreground Activity is currently showing. Distinct from [currentSession],
254 * which stays set while the app is merely backgrounded (the chat screen isn't disposed
255 * when the user locks their phone). Permission/completion notifications need the real
256 * foreground signal so they still fire for a session the user "has open" but has walked
257 * away from — the core of the run-in-the-background use case. Set from MainActivity's
259 private val _isForeground = MutableStateFlow(false)
260 open val isForeground: StateFlow<Boolean> = _isForeground.asStateFlow()
261 open fun setForeground(foreground: Boolean) { _isForeground.value = foreground }
263 /** True when the user is actively looking at [sessionId] (app foregrounded AND that
264 * session is the one on screen), so an in-app affordance (the permission dialog) is
265 * handling it and a notification would be redundant. */
266 private fun isActivelyViewing(sessionId: String): Boolean =
267 _isForeground.value && sessionId == _currentSession.value
269 open fun setCurrentSession(id: String?) {
270 _currentSession.value = id
272 _unread.update { it - id }
273 unreadMessageIds.remove(id)
275 SessionNotifications.cancel(it, id)
276 SessionNotifications.cancelPermission(it, id)
277 SessionNotifications.cancelError(it, id)
282 /** Clear unread badge and dedup set for a deleted session so its stale count doesn't
283 * linger in the session list badge or leak memory until the next connect/disconnect. */
284 open fun clearUnread(id: String) {
285 _unread.update { it - id }
286 unreadMessageIds.remove(id)
289 /** Mark a session as read (clear its unread badge) — the "Mark read" notification action. */
290 open fun markSessionRead(id: String) = clearUnread(id)
292 /** Clear every session's unread badge at once — the session list's "Mark all read" action. */
293 open fun clearAllUnread() {
294 _unread.update { emptyMap() }
297 /** Restore a session's unread badge after a failed server switch reconnects.
298 * Preserves the prior [count] so a session badged with "5 unread" before the
299 * switch attempt still shows 5 (not 1) after the restore. No-op for the session
300 * the user is currently viewing, since a viewed session is never badged. */
301 open fun restoreUnread(id: String, count: Int) {
302 if (id != _currentSession.value) {
303 _unread.update { it + (id to count) }
307 /** Resolve a localized string — view models reach resources through the container. */
308 open fun string(id: Int, vararg formatArgs: Any): String {
309 if (appContext == null) return ""
310 return if (formatArgs.isEmpty()) appContext.getString(id) else appContext.getString(id, *formatArgs)
314 * Convert a throwable into a user-facing message. Classifies the error by concrete
315 * type (network, timeout, HTTP status) rather than string-matching class names, so
316 * the message reflects what actually went wrong without leaking internal URLs/state.
318 open fun friendlyError(t: Throwable): String =
319 friendlyErrorFor(t, activeConnection.value?.profile?.baseUrl.orEmpty())
322 * Same classification as [friendlyError] but accepts an explicit base URL, so callers
323 * that aren't operating on the active connection (e.g. [probeServer]) can still
324 * produce a user-facing message with the right server address.
326 open fun friendlyErrorFor(t: Throwable, baseUrl: String): String =
327 when (classifyError(t)) {
328 ErrorKind.NOT_REACHABLE, ErrorKind.NETWORK ->
329 string(R.string.error_not_reachable, baseUrl)
330 ErrorKind.TIMEOUT -> string(R.string.error_timeout)
331 ErrorKind.SERVER -> string(R.string.error_server)
332 // Show only the HTTP status (e.g. 401, 404) — never the request URL,
333 // which a ClientRequestException carries in its message and which we
334 // promised not to leak (it can include auth or internal paths).
335 ErrorKind.CLIENT -> {
336 val code = responseStatusCode(t)
338 // Auth failures reuse the connection banner text: the request itself is
339 // fine, the credentials aren't, so the fix is re-editing the profile —
340 // not "request failed (401)" which implies an opaque program error.
341 401, 403 -> string(R.string.connection_failed)
342 404 -> string(R.string.error_not_found)
343 else -> code?.let { string(R.string.error_client_status, it) }
344 ?: string(R.string.error_generic)
347 ErrorKind.UNKNOWN -> string(R.string.error_generic)
351 * Drive the [RunForegroundService] from the process-lived [appScope] rather than from the
352 * Activity's composition. Previously the start/stop was wired to a
353 * `collectAsStateWithLifecycle` collector in the root composable, which pauses while the
354 * Activity is STOPPED — so a run *started while backgrounded* (e.g. via the notification
355 * inline-reply) never acquired foreground priority and its long-lived SSE stream could be
356 * killed by Doze, and a run that *ended* while backgrounded left the "working…"
357 * notification lingering until the app was reopened. Collecting here reacts to
358 * [anyRunActive] regardless of Activity state.
360 private fun observeRunForegroundService() {
361 val ctx = appContext ?: return
363 // Debounce so a run that starts then fails/idles almost immediately doesn't
364 // dispatch startForegroundService() then stopService() back-to-back — if the stop
365 // wins that race before the service calls startForeground(), Android raises the
366 // "did not then call startForeground()" crash ~5s later. Collapsing the window
367 // emits only the settled state; distinctUntilChanged drops a no-op re-emit when a
368 // brief true→false→true flurry settles back to the value already applied.
370 .debounce(NetworkConfig.runForegroundDebounceMs)
371 .distinctUntilChanged()
374 // When exactly one run is active, resolve its title (and id, for the
375 // Stop action) so the notification identifies which session is running
376 // and can be aborted from the notification. Multiple concurrent runs
377 // fall back to the generic title (naming one would be misleading).
378 val activeId = synchronized(activeRuns) {
379 activeRuns.toList().singleOrNull()
381 val title = activeId?.let { sid ->
382 runCatchingCancellable {
383 activeConnection.value?.repository?.listSessions()
384 ?.firstOrNull { it.id == sid }?.displayTitle
387 RunForegroundService.start(ctx, title, activeId, _runProgressText.value)
389 RunForegroundService.stop(ctx)
393 // Separate collector: update the FGS notification's progress text ("Step 2 of 5")
394 // as the agent works through its task plan. Re-sends the service intent with the
395 // updated progress, which the service uses to rebuild the notification.
398 .drop(1) // skip initial value — the anyRunActive collector handles the first start
399 .distinctUntilChanged()
400 .collect { progress ->
401 if (_anyRunActive.value) {
402 val activeId = synchronized(activeRuns) { activeRuns.toList().singleOrNull() }
403 val title = activeId?.let { sid ->
404 runCatchingCancellable {
405 activeConnection.value?.repository?.listSessions()
406 ?.firstOrNull { it.id == sid }?.displayTitle
409 RunForegroundService.start(ctx, title, activeId, progress)
416 * Reconcile active-run tracking on an SSE *reconnect*. A run that completes during a stream
417 * outage never delivers its `SessionIdle`, so its id would stay in [activeRuns] and pin
418 * [anyRunActive] (and the foreground service + "working…" notification) on indefinitely. On a
419 * reconnect we clear the tracking and let live streaming events re-populate it: a genuinely
420 * in-progress run re-adds itself within moments via [isLiveRunActivity], while a run that
421 * finished during the outage produces no further live events (its completion arrives only via
422 * the REST re-seed), so it correctly stays cleared. The first connect of a connection is not a
423 * reconnect; `connect()` already resets [activeRuns], so only subsequent Connected transitions
426 private fun observeRunReconcileOnReconnect() {
428 activeConnection.collectLatest { conn ->
429 if (conn == null) return@collectLatest
430 var hasConnectedBefore = false
431 conn.events.state.collect { state ->
432 if (state == EventStreamClient.ConnectionState.Connected) {
433 if (hasConnectedBefore) {
434 synchronized(activeRuns) {
436 _anyRunActive.value = false
439 hasConnectedBefore = true
447 * Flush the offline outbox whenever there's an active connection and the device is
448 * online (and on any manual [flushOutbox] nudge). Queued messages are only sent to the
449 * server they were composed for (matched by profile id), so a reconnect to a different
450 * server never misfires them against a session id that doesn't exist there. Uses a plain
451 * `collect` (not collectLatest) so an in-flight send is never cancelled mid-flight and
452 * send order is preserved; each message carries a stable idempotency key (its persistent
453 * outbox id) so even a re-send across flushes is deduplicated server-side.
455 private fun observeOutbox() {
457 combine(activeConnection, isOnline, _outboxFlushTrigger) { conn, online, _ -> conn to online }
458 .collect { (conn, online) ->
459 if (conn == null || !online) return@collect
465 private suspend fun flushOutboxNow(conn: OpencodeConnection) {
466 val pending = outboxStore.messages.value
467 .filter { it.profileId == conn.profile.id }
468 .sortedBy { it.createdAt }
469 if (pending.isEmpty()) return
470 _outboxSending.value = true
472 for (msg in pending) {
473 val result = runCatchingCancellable {
474 conn.repository.sendPrompt(
477 attachments = msg.attachments.map {
478 FilePromptPart(mime = it.mime, url = it.url, filename = it.filename)
482 // Reuse the persistent outbox id as a stable idempotency key so a
483 // later flush of a message whose earlier POST reached the server (but
484 // whose response was lost) is deduplicated server-side instead of
485 // starting a duplicate agent run.
486 idempotencyKey = msg.id,
489 Log.w("AppContainer", "Outbox flush failed for ${msg.sessionId}: ${safeExceptionSummary(it)}")
491 if (result.isSuccess) {
492 outboxStore.remove(msg.id)
495 // A permanently-undeliverable message (a non-408/429 4xx — e.g. the target
496 // session was deleted server-side, yielding a 404) must be dropped, not left at
497 // the head of the queue. The loop stops at the first failure to preserve send
498 // order, so a poison message would otherwise block every later queued message
499 // for this profile forever. Transient failures (offline again, 5xx, 408/429)
500 // keep the queue intact and retry on the next trigger.
501 // 401/403 (bad/expired credentials) are NOT permanent: the message itself is
502 // deliverable once the user fixes the profile. Dropping it would silently lose
503 // the user's composed offline prompt on a recoverable auth failure. They stay
504 // in the queue and retry on the next flush after auth is restored.
505 val status = result.exceptionOrNull()?.let { responseStatusCode(it) }
506 // Permanent = a non-408/429/401/403 4xx, matching withRetry's non-retryable rule
507 // except for auth failures, which are recoverable by re-editing the profile.
508 val permanent = when (status) {
509 null, 408, 429, 401, 403 -> false
516 "Dropping undeliverable outbox message ${msg.id} for ${msg.sessionId} (HTTP $status): ${string(R.string.outbox_dropped_text)}",
518 outboxStore.remove(msg.id)
519 // Previously this silent drop only hit logcat, so the user got no signal
520 // that their composed reply never landed. Surface it on the ERROR channel.
521 // postError is the only public entry point that posts there; it renders a
522 // fixed title/text and interpolates its title arg, so the localized
523 // outbox_dropped_title is passed as that arg (a user-facing message rather
524 // than an opaque session id). Tap routes back to the originating profile.
527 runCatchingCancellable {
528 SessionNotifications.postError(
530 string(R.string.outbox_dropped_title),
534 Log.w("AppContainer", "outbox drop notification failed: ${safeExceptionSummary(it)}")
542 _outboxSending.value = false
546 /** Release resources held for the process lifetime (network callback, app scope). */
547 open fun shutdown() {
548 // Cancel the app scope first so any coroutine holding the connection mutex
549 // (e.g. a connect() suspended on profileStore) is cancelled and releases the
550 // mutex. Otherwise the runBlocking below would deadlock waiting for a coroutine
551 // that can't be cancelled until after the runBlocking completes.
553 // Bounded runBlocking so a stuck mutex or a slow close() can't ANR the app.
554 // 2 seconds is generous: appScope.cancel() already triggered cancellation of
555 // any coroutine holding the mutex; we're just waiting for it to unwind.
556 kotlinx.coroutines.runBlocking {
557 withTimeoutOrNull(2_000) {
558 connectionMutex.withLock {
559 runCatching { _activeConnection.value?.close() }
560 _activeConnection.value = null
563 // If the timeout fired, force-clear the connection so it isn't left dangling.
564 _activeConnection.value = null
566 val cm = appContext?.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
567 networkCallback?.let { runCatching { cm?.unregisterNetworkCallback(it) } }
568 if (!skipInit) soy.iko.opencode.data.repo.CrashLogger.get(appContext!!).shutdown()
569 if (!skipInit) draftStore.shutdown()
573 * Watch the SSE bus for new message activity and badge any session that isn't
574 * currently open. This powers the unread dot on the session list so the user can
575 * tell which conversations got a reply while they were elsewhere.
577 * Uses [flatMapLatest] so a server switch (which replaces the active connection)
578 * re-subscribes to the new event stream and drops the old one.
580 @OptIn(ExperimentalCoroutinesApi::class)
581 private fun observeMessageActivity() {
583 // Retry the observer if it fails for any reason (transient flow exception,
584 // cancellation due to scope issues). Without this, a single failure would
585 // permanently disable unread tracking and completion notifications.
587 runCatchingCancellable {
589 .flatMapLatest { conn ->
590 // Capture the originating connection so notification coroutines
591 // resolve titles and route permission responses against THIS
592 // server, not whichever happens to be active when they resume.
593 // A server switch during a listSessions() round-trip would
594 // otherwise resolve the title (or POST the permission reply)
595 // against the new server, where the session/permission doesn't
596 // exist — showing an opaque id or silently losing the reply.
597 conn?.events?.events?.map { conn to it } ?: emptyFlow()
599 .collect { (conn, event) ->
600 // A permission request needs the user's approval. Post a heads-up
601 // notification (with Allow/Reject actions) unless the user is actively
602 // viewing this session, where the in-app permission dialog handles it.
603 if (event is PermissionUpdated) {
604 val psid = event.properties.sessionID.takeIf { it.isNotBlank() } ?: return@collect
605 if (!isActivelyViewing(psid)) appScope.launch { notifyPermission(event.properties, conn) }
608 // The permission was answered (in-app, from the notification, or
609 // auto-resolved): clear its heads-up notification.
610 if (event is PermissionReplied) {
611 val psid = event.properties.sessionID?.takeIf { it.isNotBlank() } ?: return@collect
612 appContext?.let { SessionNotifications.cancelPermission(it, psid) }
615 // A run failed. Clear its active-run tracking (an error isn't always
616 // followed by a SessionIdle) and notify unless it's being viewed.
617 if (event is SessionError) {
618 val esid = event.properties.sessionID ?: return@collect
619 val wasRunning = synchronized(activeRuns) {
620 activeRuns.remove(esid).also { removed ->
621 if (removed) _anyRunActive.value = activeRuns.isNotEmpty()
624 if (wasRunning && !isActivelyViewing(esid)) {
625 appScope.launch { notifySessionError(esid, conn.repository, conn.profile.id) }
629 // SessionIdle is not "message activity" (don't badge as unread)
630 // but signals a run finished — fire a completion notification if the
631 // session isn't currently being viewed and was actively streaming.
632 if (event is SessionIdle) {
633 val idleSid = event.properties.sessionID ?: return@collect
634 // Clear the run state unconditionally — even for the session on
635 // screen. Gating the removal on `idleSid != currentSession` (as
636 // this once did) leaks the id in activeRuns and pins anyRunActive
637 // true forever whenever a run finishes while being viewed. Only
638 // the completion *notification* is suppressed for the viewed one.
639 val wasRunning = synchronized(activeRuns) {
640 activeRuns.remove(idleSid).also { removed ->
641 if (removed) _anyRunActive.value = activeRuns.isNotEmpty()
644 // Clear the progress text when the run finishes.
645 _runProgressText.value = null
646 if (wasRunning && !isActivelyViewing(idleSid)) {
647 // Fire this off the collector's coroutine: notifySessionCompleted
648 // does a network listSessions() to resolve the title, and blocking
649 // the shared-event collector here would stall the SharedFlow,
650 // overflowing its DROP_OLDEST buffer and silently dropping live
651 // parts/updates for every other subscriber (e.g. the message reducer).
652 appScope.launch { notifySessionCompleted(idleSid, conn.repository, conn.profile.id) }
656 // Track the agent's task-plan progress (from todowrite tool calls) so the
657 // foreground-service notification can show "Step 2 of 5" instead of just a
658 // chronometer. The todowrite tool sends the full plan on every call, so the
659 // latest call's input is a complete snapshot.
660 if (event is MessagePartUpdated) {
661 val part = event.properties.part
662 if (part is ToolPart && part.tool.equals(TODO_WRITE_TOOL, ignoreCase = true)) {
663 val todos = parseTodos(part.state.inputElement())
664 if (todos.isNotEmpty()) {
665 val completed = todos.count { it.statusEnum() == TodoStatus.COMPLETED }
666 _runProgressText.value = appContext?.getString(
667 soy.iko.opencode.R.string.notif_running_progress,
674 val sid = sessionOf(event) ?: return@collect
675 // Re-read currentSession here, right before mutating the badge state:
676 // the user may have opened this session between when this event was
677 // enqueued and now (setCurrentSession clears its badge and dedup set).
678 // A stale earlier check would race a badge — and a dedup entry — back
679 // in for the session that's actually on screen.
680 if (sid != _currentSession.value) {
681 // Increment the unread count once per distinct *message*, not per
682 // event: a single reply emits many message.updated / message.part.
683 // updated events (one per streamed token, plus cost/token refreshes),
684 // which would otherwise inflate the badge into the dozens/hundreds.
685 val messageId = messageIdOfEvent(event)
686 val counted = unreadMessageIds.getOrPut(sid) {
687 java.util.Collections.synchronizedSet(mutableSetOf())
689 if (messageId == null || counted.add(messageId)) {
690 _unread.update { current ->
691 // Guard again inside the atomic update lambda so a retry
692 // (or an open that landed just now) can't reintroduce the
693 // badge for the session on screen.
694 if (sid == _currentSession.value) current
695 else current + (sid to (current[sid] ?: 0) + 1)
699 // Track sessions actively streaming so we know which idle events
700 // represent a finished run worth notifying about.
701 if (isLiveRunActivity(event)) {
702 // The add-under-lock is required both for thread-safety and for the
703 // access-order refresh (keeps a continuously-streaming session from
704 // being evicted), so it runs per event.
705 synchronized(activeRuns) {
707 // Derive the flag from the set state *inside* the same lock that
708 // guards the set, so a concurrent reconnect sweep (which clears
709 // both under this lock) can't interleave and leave the flag pinned
710 // true with an empty set — a stuck foreground service + "working…".
711 _anyRunActive.value = activeRuns.isNotEmpty()
714 } }.onFailure { Log.w("AppContainer", "Message activity observer failed, will retry: ${safeExceptionSummary(it)}") }
716 delay(NetworkConfig.observerRetryDelayMs)
721 /** True if [event] is live streaming activity worth tracking as an active run. A
722 * trailing message.updated that lands *after* the run finished (final cost/token totals,
723 * carrying a completion time) is NOT live streaming: with no further SessionIdle to
724 * follow it, re-adding the session would pin anyRunActive true indefinitely. */
725 private fun isLiveRunActivity(event: BusEvent): Boolean {
726 if (event !is MessagePartUpdated && event !is MessageUpdated) return false
727 // For MessageUpdated, only an in-flight *assistant* message counts as live activity.
728 // A UserMessage or UnknownMessage update (e.g. a metadata refresh of an existing user
729 // message) would otherwise add the session to activeRuns with no following SessionIdle
730 // to clear it — pinning anyRunActive true and the foreground notification indefinitely.
731 // Mirrors SessionRepository.isRunActivity's guard exactly.
732 if (event is MessageUpdated) {
733 val info = event.properties.info
734 return info is AssistantMessage && !info.isComplete && info.error == null
736 // A step-finish part is the trailing completion marker for a run (it carries the
737 // final cost/token totals) and arrives after the last stream delta. Like a completed
738 // message.updated, treating it as live would re-add the session to activeRuns with no
739 // following SessionIdle to clear it — pinning anyRunActive true indefinitely when such
740 // a part is replayed/reordered after the run already went idle.
741 val part = (event as MessagePartUpdated).properties.part
742 return part !is StepFinishPart
745 /** Session ids currently streaming an assistant run (best-effort, in-process). Backed by
746 * an access-order [java.util.LinkedHashMap] so re-adding an already-present id on each
747 * streaming event refreshes its recency (a plain LinkedHashSet.add is a no-op that does
748 * NOT reorder). Entries are drained by [SessionIdle] (and cleared wholesale on reconnect
749 * reconcile); we intentionally do NOT cap the set, because evicting the eldest would drop
750 * its completion notification when its SessionIdle eventually arrives — a busy server
751 * with many concurrent runs is still bounded by the server's own concurrency, so growth
752 * is not unbounded in practice. */
753 private val activeRuns: MutableSet<String> = java.util.Collections.synchronizedSet(
754 java.util.Collections.newSetFromMap(java.util.LinkedHashMap<String, Boolean>(16, 0.75f, true)),
757 /** Per non-viewed session, the set of message ids already reflected in its unread
758 * count. De-duplicates the many streaming events that share one message id so the
759 * badge counts messages, not deltas. Cleared when a session is viewed or on connect. */
760 private val unreadMessageIds = java.util.concurrent.ConcurrentHashMap<String, MutableSet<String>>()
762 /** Guards connect/disconnect so concurrent callers can't leak an old connection. */
763 private val connectionMutex = Mutex()
765 /** Only persist the `lastUsed` timestamp if it's older than this, to avoid
766 * a DataStore + encrypted-prefs write on every rapid reconnect. */
767 private companion object {
768 const val LAST_USED_SAVE_THRESHOLD_MS = 60_000L
771 /** Resolve the human-readable title for [sessionId] (best-effort), falling back to the id.
772 * Uses [originRepo] — the repository of the connection that emitted the event — so a
773 * server switch during the listSessions() round-trip doesn't resolve the title against
774 * an unrelated server (showing an opaque id, or worse, a colliding session's title). */
775 private suspend fun resolveSessionTitle(sessionId: String, originRepo: SessionRepository): String =
776 runCatchingCancellable {
777 originRepo.listSessions()?.firstOrNull { it.id == sessionId }?.displayTitle
778 }.getOrNull() ?: sessionId
780 /** Resolve the title for [sessionId] (best-effort) and post a completion notification.
781 * [originProfileId] is the id of the server that ran the session; it's embedded in the
782 * inline-reply action so a follow-up routes back to THIS server even after the user switches
783 * connections (mirrors [notifyPermission]). */
784 private suspend fun notifySessionCompleted(sessionId: String, originRepo: SessionRepository, originProfileId: String) {
785 val title = resolveSessionTitle(sessionId, originRepo)
786 // Re-check viewing right before posting: the call site checks !isActivelyViewing(sid)
787 // at event time, but resolveSessionTitle() suspends on a listSessions() round-trip.
788 // If the user opened the session in that gap, setCurrentSession already cancelled the
789 // prior notification — posting a fresh one here would leave a "session ready" heads-up
790 // lingering while they're already looking at the finished run.
791 if (isActivelyViewing(sessionId)) return
792 appContext?.let { SessionNotifications.postCompleted(it, sessionId, title, originProfileId) }
795 /** Post a heads-up permission notification for [permission], resolving its session title.
796 * [originConn] is the connection that emitted the permission event — its profile id is
797 * embedded in the notification so the receiver routes the response back to THIS server. */
798 private suspend fun notifyPermission(permission: Permission, originConn: OpencodeConnection) {
799 val sessionId = permission.sessionID.takeIf { it.isNotBlank() } ?: return
800 val title = resolveSessionTitle(sessionId, originConn.repository)
801 // Re-check viewing just before posting — see notifySessionCompleted for the race.
802 if (isActivelyViewing(sessionId)) return
803 appContext?.let { SessionNotifications.postPermission(it, permission, title, originConn.profile.id) }
806 /** Post an error notification for a failed background run. [originProfileId] is embedded
807 * in the tap intent so the session opens under the server that ran it, not whichever
808 * is active when the user taps — mirroring [notifySessionCompleted]. */
809 private suspend fun notifySessionError(sessionId: String, originRepo: SessionRepository, originProfileId: String) {
810 val title = resolveSessionTitle(sessionId, originRepo)
811 // Re-check viewing just before posting — see notifySessionCompleted for the race.
812 if (isActivelyViewing(sessionId)) return
813 appContext?.let { SessionNotifications.postError(it, sessionId, title, originProfileId) }
816 /** Respond to a permission from a notification action (Allow once / Always / Reject),
817 * off any UI. Runs on the process-lived app scope; [onDone] fires when the call resolves
818 * so the receiver's goAsync() result can finish. Its Boolean is true only when a live
819 * connection existed AND respondPermission succeeded — the receiver uses it to decide
820 * whether to dismiss the notification. A missing connection makes the respond a no-op,
821 * so it must report failure (not success) rather than let the notification vanish with
822 * the tool left unanswered.
824 * [profileId] is the id of the server that posted the permission request (embedded in
825 * the notification's PendingIntent). If it doesn't match the currently active
826 * connection's profile, the response is NOT sent: posting it to a different server
827 * would 404 (treated as "already resolved" success by respondPermission), dismissing
828 * the notification while the original server's tool stays paused forever. Reporting
829 * failure instead leaves the notification up so the user can switch back and retry. */
830 open fun respondToPermissionFromNotification(
832 permissionId: String,
833 response: PermissionResponse,
835 onDone: (Boolean) -> Unit,
838 // Guarantee onDone (so the BroadcastReceiver's goAsync() finishes) even if the
839 // scope is cancelled mid-call (process shutdown cancels appScope first). Without
840 // this, a cancelled scope leaves pending.finish() never called → a receiver ANR
841 // and the permission response silently lost.
844 success = runCatchingCancellable {
845 val conn = activeConnection.value ?: return@runCatchingCancellable false
846 // Route to the originating server only. A mismatch means the user
847 // switched servers while the permission was pending; sending the reply
848 // to the wrong server would be silently swallowed as a 404-success.
849 if (profileId != null && conn.profile.id != profileId) return@runCatchingCancellable false
850 conn.api.respondPermission(sessionId, permissionId, response)
853 Log.w("AppContainer", "notif permission respond failed: ${safeExceptionSummary(it)}")
854 }.getOrDefault(false)
861 /** Send a follow-up prompt from a notification's inline reply, off any UI. Uses the
862 * server's default model/agent. Routes through the durable [outboxStore] — the same path
863 * the in-app composer uses ([soy.iko.opencode.ui.chat.ChatViewModel.enqueueOffline]) —
864 * rather than a direct repository send, so a reply composed while disconnected survives to
865 * be flushed on reconnect instead of being silently dropped. [flushOutbox] nudges an
866 * immediate send when a connection is live, so an online reply still goes out now. [onDone]
867 * reports whether the reply was durably enqueued so the receiver only dismisses the
868 * notification on success. */
869 open fun sendPromptFromNotification(sessionId: String, text: String, originProfileId: String?, onDone: (Boolean) -> Unit) {
871 // Guarantee onDone (so the BroadcastReceiver's goAsync() finishes) even if the
872 // scope is cancelled mid-call — see respondToPermissionFromNotification above.
875 // Attribute to the profile that POSTED the notification (embedded in the reply
876 // intent) so the follow-up is always enqueued against the server that owns
877 // [sessionId] — never whichever server happens to be active now, which would POST
878 // A's session id to B, 404, and silently drop the reply. Fall back to the active/
879 // most-recent profile only for a legacy intent that predates the embedded id.
880 val profileId = originProfileId
881 ?: activeConnection.value?.profile?.id
882 ?: runCatchingCancellable { profileStore.profiles.first().firstOrNull()?.id }.getOrNull()
883 enqueued = if (profileId == null) {
886 runCatchingCancellable {
889 id = java.util.UUID.randomUUID().toString(),
890 profileId = profileId,
891 sessionId = sessionId,
894 createdAt = System.currentTimeMillis(),
898 Log.w("AppContainer", "notif reply enqueue failed: ${safeExceptionSummary(it)}")
901 if (enqueued) flushOutbox()
908 /** Abort the active run for [sessionId] from a notification Stop action. Routes to the
909 * originating profile (or the active connection) so the abort reaches the server that
910 * owns the session. Runs on the app scope so it survives the BroadcastReceiver's lifetime. */
911 open fun abortRunFromNotification(sessionId: String, originProfileId: String?, onDone: (Boolean) -> Unit) {
915 val conn = activeConnection.value?.takeIf {
916 originProfileId == null || it.profile.id == originProfileId
917 } ?: activeConnection.value
919 success = runCatchingCancellable { conn.repository.abort(sessionId) }.isSuccess
928 * Retry the last user prompt for [sessionId] from an error notification's "Retry last"
929 * action. Fetches the session's messages, re-sends the most recent user prompt, and
930 * cancels the error notification on success. Routes to the originating profile's
931 * connection (falling back to the active one) so the send reaches the server that owns
932 * the session. A no-op (reported as false) when there's no connection or no prior prompt.
934 open fun retryLastFromNotification(sessionId: String, originProfileId: String?, onDone: (Boolean) -> Unit) {
938 val conn = activeConnection.value?.takeIf {
939 originProfileId == null || it.profile.id == originProfileId
940 } ?: activeConnection.value
942 success = runCatchingCancellable {
943 val messages = conn.api.listMessages(sessionId)
944 val lastPrompt = messages
945 .lastOrNull { it.info is soy.iko.opencode.data.model.UserMessage }
947 ?.filterIsInstance<soy.iko.opencode.data.model.TextPart>()
948 ?.joinToString("\n\n") { it.text }
949 ?.takeIf { it.isNotBlank() }
950 if (lastPrompt != null) {
951 // model=null lets the server reuse the session's current model; we
952 // don't know the originally-selected model from a notification context.
953 conn.repository.sendPrompt(sessionId, lastPrompt, model = null)
966 /** Extract the session id an event pertains to, for message-activity events. */
967 private fun sessionOf(event: BusEvent): String? = sessionOfEvent(event)
970 * On cold start, transparently reconnect to the most recently used server so a
971 * returning user lands in their session list instead of the empty server screen.
973 private suspend fun autoConnect() {
974 val recent = runCatchingCancellable { profileStore.profiles.first() }
975 .getOrDefault(emptyList())
976 .firstOrNull() ?: return
977 // The profile read above suspends on DataStore. A manual connect() to a different
978 // server during that window establishes a connection the user explicitly chose; we must
979 // not clobber it by connecting to `recent` on top. connectIfIdle() checks "is anything
980 // active?" and connects atomically under connectionMutex, returning null if the cold-start
981 // slot was already taken — closing the TOCTOU window an unlocked null-check plus connect()'s
982 // unconditional close would leave. (disconnectIf below similarly guards the failed-ping
983 // teardown by identity.)
984 _reconnecting.value = true
985 var conn: OpencodeConnection? = null
987 val ok = runCatchingCancellable {
988 conn = connectIfIdle(recent) ?: return@runCatchingCancellable false
991 }.getOrDefault(false)
993 conn == null -> Unit // slot already taken by a manual connect — leave it be
994 ok -> _autoConnectDone.value = true
995 else -> disconnectIf(conn!!)
998 // Reset in a finally so a CancellationException (e.g. scope cancelled during
999 // shutdown) can't leave the flag pinned true — runCatchingCancellable rethrows it,
1000 // skipping the line below the try block.
1001 _reconnecting.value = false
1006 * When the device regains connectivity, nudge the active SSE stream to reconnect
1007 * immediately instead of waiting out the exponential backoff.
1009 private val networkCallback: ConnectivityManager.NetworkCallback? = registerNetworkMonitor()
1011 private fun registerNetworkMonitor(): ConnectivityManager.NetworkCallback? {
1012 if (appContext == null) return null
1013 val cm = appContext!!.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
1015 // Seed the online state from the current active network so the indicator is
1016 // correct on cold start rather than defaulting to "online" until a callback fires.
1017 _isOnline.value = runCatching { cm.activeNetwork != null }.getOrDefault(true)
1018 val callback = object : ConnectivityManager.NetworkCallback() {
1019 override fun onAvailable(network: Network) { _isOnline.value = true; activeConnection.value?.events?.triggerReconnect() }
1020 // A matched (internet-capable) network was lost. Don't eagerly flip offline: during a
1021 // Wi-Fi→cellular handoff onLost(WiFi) fires before onAvailable(cellular) and
1022 // cm.activeNetwork is transiently null in that window — flashing offline there would
1023 // skip outbox flushing and blink the indicator. Instead re-check activeNetwork after a
1024 // short grace period and only report offline if there's still no active network.
1026 // This MUST live in onLost, not onUnavailable: onUnavailable is delivered only for
1027 // requestNetwork() with a timeout, never for registerNetworkCallback() — relying on it
1028 // (as this once did) meant offline was never detected after the first onAvailable.
1029 override fun onLost(network: Network) {
1031 delay(NetworkConfig.networkOfflineGraceMs)
1032 _isOnline.value = runCatching { cm.activeNetwork != null }.getOrDefault(true)
1036 // Only match networks that actually provide internet — a capability-less request
1037 // fires onAvailable for transports that can't reach the server (and would trigger
1038 // spurious reconnects), so require NET_CAPABILITY_INTERNET.
1039 val request = NetworkRequest.Builder()
1040 .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
1042 runCatching { cm.registerNetworkCallback(request, callback) }
1043 .onFailure { Log.w("AppContainer", "Failed to register network callback", it) }
1047 /** Open (or re-open) a connection to [profile], replacing any current one.
1048 * Only persists the updated `lastUsed` timestamp if it's stale (older than a
1049 * minute), so rapid reconnect storms don't each trigger a DataStore write. */
1050 open suspend fun connect(profile: ServerProfile): OpencodeConnection =
1051 connectionMutex.withLock { connectLocked(profile) }
1053 /** Connect to the profile with [profileId], looking it up from [profileStore]. Returns
1054 * true on success, false if the profile wasn't found or connect threw. Used by the
1055 * notification body-tap path to route a session open back to the originating server. */
1056 open suspend fun connectByProfileId(profileId: String): Boolean {
1057 val profile = profileStore.profiles.first().find { it.id == profileId } ?: return false
1058 return runCatchingCancellable { connect(profile) }.isSuccess
1061 /** Connect to [profile] only if nothing is currently active, checking-and-connecting
1062 * atomically under [connectionMutex]. Returns null (touching nothing) when a connection is
1063 * already active, so autoConnect() can't clobber a manual connect() the user made during its
1064 * suspending profile read — closing the TOCTOU window an unlocked null-check would leave. */
1065 private suspend fun connectIfIdle(profile: ServerProfile): OpencodeConnection? =
1066 connectionMutex.withLock {
1067 if (_activeConnection.value != null) null else connectLocked(profile)
1070 /** Build and install a connection to [profile], replacing any current one. Caller MUST hold
1071 * [connectionMutex]. */
1072 private suspend fun connectLocked(profile: ServerProfile): OpencodeConnection {
1073 // Capture the previous profile id before closing, so we can preserve unread badges
1074 // when reconnecting to the same server — a brief network blip or manual reconnect
1075 // shouldn't erase "5 unread". Only a genuine server switch clears them.
1076 val previousProfileId = _activeConnection.value?.profile?.id
1077 _activeConnection.value?.close()
1078 _activeConnection.value = null
1079 // Mutate the run set and its derived flag together under activeRuns' monitor: the SSE
1080 // event handler adds ids under the same lock, so an unlocked clear() here could
1081 // interleave with an add()+flag=true and leave _anyRunActive pinned true with a torn-
1082 // down connection and no SessionIdle to clear it — a stuck foreground service.
1083 synchronized(activeRuns) {
1085 _anyRunActive.value = false
1087 if (previousProfileId != profile.id) {
1088 _unread.value = emptyMap()
1089 unreadMessageIds.clear()
1091 val now = System.currentTimeMillis()
1092 val resolved = profileStore.resolve(profile)
1093 val needsSave = (now - resolved.lastUsed) > LAST_USED_SAVE_THRESHOLD_MS
1094 val finalProfile = if (needsSave) resolved.copy(lastUsed = now) else resolved
1095 if (needsSave) profileStore.save(finalProfile)
1096 return OpencodeConnection(finalProfile, messageCacheStore).also { _activeConnection.value = it }
1099 open suspend fun disconnect() {
1100 // Acquire the connection mutex so disconnect is serialized with connect().
1101 // Previously this was fire-and-forget (launch on appScope), which raced with
1102 // a subsequent connect(): connect() could acquire the mutex, create a new
1103 // connection, and then the pending disconnect() would close it.
1104 connectionMutex.withLock { teardownActiveLocked() }
1107 /** Tear down [expected] only if it is still the active connection, with the identity
1108 * check performed *under* the mutex. A caller that checks `activeConnection.value === x`
1109 * itself and then calls [disconnect] has a TOCTOU race: a concurrent connect() to a
1110 * different server can replace the active connection in the gap, and [disconnect] would
1111 * then close that new connection. autoConnect()'s failed-ping teardown uses this. */
1112 private suspend fun disconnectIf(expected: OpencodeConnection) {
1113 connectionMutex.withLock {
1114 if (_activeConnection.value === expected) teardownActiveLocked()
1118 /** Close and reset the active connection. Caller MUST hold [connectionMutex]. */
1119 private suspend fun teardownActiveLocked() {
1120 _activeConnection.value?.close()
1121 _activeConnection.value = null
1122 // Reset run/unread state on disconnect too. With no SSE stream there's no
1123 // SessionIdle to drain activeRuns, so without this the working indicator
1124 // (anyRunActive) and unread badges would stay pinned until the next connect().
1125 // Mutate under activeRuns' monitor (see connect()) so a concurrent SSE add() can't
1126 // re-pin _anyRunActive after the clear.
1127 synchronized(activeRuns) {
1129 _anyRunActive.value = false
1131 _unread.value = emptyMap()
1132 unreadMessageIds.clear()
1135 /** Deferred profile deletions keyed by id, run on the process-lived [appScope] (not a
1136 * ViewModel scope) so navigating away during the undo window still commits the delete
1137 * instead of silently cancelling it — mirroring [CrashLogger.scheduleDelete]. */
1138 private val pendingProfileDeletes = java.util.concurrent.ConcurrentHashMap<String, Job>()
1141 * Schedule the profile [id] for deletion after [delayMs], cancellable via
1142 * [cancelProfileDelete] (the Undo action). [onDeleted]/[onError] run on [appScope]
1143 * after the delete resolves so the caller can clear its optimistic-hide state. A
1144 * repeated schedule for the same id replaces the prior timer.
1146 open fun scheduleProfileDelete(
1149 onDeleted: () -> Unit,
1150 onError: (Throwable) -> Unit,
1152 val job = appScope.launch {
1154 // Claim the delete by removing our own entry BEFORE committing it — the atomic
1155 // remove is the sync point with cancelProfileDelete. Losing the race (an undo
1156 // removed us first) means we skip, honouring cancel's `true`; deleting first
1157 // would let a cancel land in the gap and falsely report undo of a gone profile.
1158 // remove(id, thisJob) also no-ops for a stale job after a reschedule.
1159 val claimed = coroutineContext[Job]?.let { pendingProfileDeletes.remove(id, it) } == true
1160 if (!claimed) return@launch
1161 runCatchingCancellable { profileStore.delete(id) }
1162 .onSuccess { onDeleted() }
1163 .onFailure { onError(it) }
1165 pendingProfileDeletes[id]?.cancel()
1166 pendingProfileDeletes[id] = job
1169 /** Cancel a pending deferred profile delete (the Undo action). Returns true if it was
1170 * still pending (undo succeeded), false if the delete had already fired. */
1171 open fun cancelProfileDelete(id: String): Boolean {
1172 val job = pendingProfileDeletes.remove(id) ?: return false
1177 /** Deferred session deletions keyed by id, run on the process-lived [appScope] (not a
1178 * ViewModel scope) so navigating away during the undo window still commits the delete
1179 * instead of silently cancelling it (which would leave the session deleted-in-UI but
1180 * alive on the server, reappearing on the next refresh). */
1181 private val pendingSessionDeletes = java.util.concurrent.ConcurrentHashMap<String, Job>()
1183 /** One-shot events carrying the id of a session scheduled for deferred deletion from
1184 * outside the session list (e.g. from the chat screen's Delete). The SessionListScreen
1185 * collects these to show an Undo snackbar, mirroring its own VM-level undo events. */
1186 private val _externalSessionUndoEvents = MutableSharedFlow<String>(
1187 extraBufferCapacity = NetworkConfig.snackbarEventBufferCapacity,
1188 onBufferOverflow = kotlinx.coroutines.channels.BufferOverflow.DROP_OLDEST,
1190 val externalSessionUndoEvents: SharedFlow<String> = _externalSessionUndoEvents.asSharedFlow()
1192 /** Emit an undo event for a session delete initiated outside the session list (e.g.
1193 * from the chat screen). The SessionListScreen collects these to show an Undo snackbar. */
1194 fun emitExternalSessionUndo(sessionId: String) {
1195 _externalSessionUndoEvents.tryEmit(sessionId)
1199 * Schedule session [id] for deletion after [delayMs], cancellable via [cancelSessionDelete]
1200 * (the Undo action). Also clears the session's stored draft on a successful delete.
1201 * [onDeleted]/[onError] run on [appScope] after the delete resolves so the caller can update
1202 * its optimistic-hide state. A repeated schedule for the same id replaces the prior timer.
1204 * The owning connection is captured now, at schedule time — NOT re-resolved when the timer
1205 * fires. Otherwise a server switch during the undo window would send the delete to whichever
1206 * server happens to be active at fire time: the original session would never be deleted (its
1207 * row reappears on the next refresh) and, in the rare id-collision case, an unrelated session
1208 * on the new server could be deleted instead. A disconnect closes the captured connection's
1209 * HTTP client, so a delete deferred across a disconnect fails and is surfaced via [onError]
1210 * (the session cannot be deleted on a server we're no longer connected to). No-ops (as an
1211 * error) if there is no active connection at schedule time.
1213 open fun scheduleSessionDelete(
1216 onDeleted: () -> Unit,
1217 onError: (Throwable) -> Unit,
1219 val conn = _activeConnection.value
1220 val job = appScope.launch {
1222 // Claim the delete by removing our own entry BEFORE committing it — the atomic
1223 // remove is the sync point with cancelSessionDelete. Losing the race (an undo
1224 // removed us first) means we skip, honouring cancel's `true`; deleting first
1225 // would let a cancel land in the gap and falsely report undo of a gone session.
1226 // remove(id, thisJob) also no-ops for a stale job after a reschedule.
1227 val claimed = coroutineContext[Job]?.let { pendingSessionDeletes.remove(id, it) } == true
1228 if (!claimed) return@launch
1230 onError(IllegalStateException("No active connection"))
1232 runCatchingCancellable { conn.repository.deleteSession(id) }
1234 draftStore.remove(id)
1235 attachmentDraftStore.remove(id)
1236 sessionPrefsStore.forget(id)
1237 messageCacheStore.remove(conn.profile.id, id)
1238 outboxStore.removeForSession(id)
1242 .onFailure { onError(it) }
1245 pendingSessionDeletes[id]?.cancel()
1246 pendingSessionDeletes[id] = job
1249 /** Cancel a pending deferred session delete (the Undo action). Returns true if it was
1250 * still pending (undo succeeded), false if the delete had already fired. */
1251 open fun cancelSessionDelete(id: String): Boolean {
1252 val job = pendingSessionDeletes.remove(id) ?: return false
1258 * Probe a server URL (without credentials) to check reachability and detect whether
1259 * the server requires authentication. Builds a short-lived HTTP client with the same
1260 * base URL normalization as [connect] but no auth, calls the health endpoint, and
1261 * classifies the outcome:
1263 * - 2xx → [ProbeResult.Reachable] (no auth needed)
1264 * - 401/403 → [ProbeResult.NeedsAuth]
1265 * - anything else → [ProbeResult.Unreachable] with a user-facing message
1267 * The probe client is always closed afterwards so no resources linger. This does not
1268 * touch the active connection or the profile store.
1270 open suspend fun probeServer(baseUrl: String, requireHttps: Boolean, certPin: String?): ProbeResult {
1271 return probeWithProfile(ServerProfile(
1274 baseUrl = baseUrl.trim(),
1277 // Probe the same effective URL save()/connect() will use: HttpClientFactory upgrades
1278 // http->https and installs the cert pinner when these are set, so a probe without them
1279 // would validate a plain-http endpoint the real connection never talks to.
1280 requireHttps = requireHttps,
1286 * Probe a server URL *with* credentials to validate them. Returns true if the
1287 * server accepts the credentials (ping succeeds), false on 401/403, and throws
1288 * on other failures (unreachable, timeout) so the caller can surface a friendly
1289 * message. Mirrors [probeServer]'s short-lived-client pattern.
1291 open suspend fun probeWithCredentials(
1295 requireHttps: Boolean,
1298 val probeProfile = ServerProfile(
1301 baseUrl = baseUrl.trim(),
1302 username = username.trim().takeIf { it.isNotBlank() },
1303 password = password.trim().takeIf { it.isNotEmpty() },
1304 // Match the effective URL/pinning save()/connect() will use (see probeServer).
1305 requireHttps = requireHttps,
1308 val client = HttpClientFactory.create(probeProfile)
1310 val api = OpencodeApiClient(client)
1313 } catch (c: kotlinx.coroutines.CancellationException) {
1315 } catch (e: Exception) {
1316 val status = responseStatusCode(e)
1317 if (status == 401 || status == 403) false else throw e
1319 runCatching { client.close() }
1323 private suspend fun probeWithProfile(profile: ServerProfile): ProbeResult {
1324 val client = HttpClientFactory.create(profile)
1326 val api = OpencodeApiClient(client)
1328 ProbeResult.Reachable
1329 } catch (c: kotlinx.coroutines.CancellationException) {
1331 } catch (e: Exception) {
1332 val status = responseStatusCode(e)
1333 if (status == 401 || status == 403) {
1334 ProbeResult.NeedsAuth
1336 ProbeResult.Unreachable(friendlyErrorFor(e, profile.baseUrl))
1339 runCatching { client.close() }
1343 // Declared last so every property above is fully initialized before these observers start.
1344 // They only launch coroutines into appScope (Dispatchers.IO) — nothing here runs
1345 // synchronously during construction — so placing init at the end removes any dependence on
1346 // the textual declaration order of the fields those observers reference.
1349 NotificationChannels.create(appContext!!)
1350 observeMessageActivity()
1351 observeRunReconcileOnReconnect()
1352 observeRunForegroundService()
1353 // Flush right after load() populates the queue. observeOutbox()'s combine is keyed
1354 // on activeConnection/isOnline/_outboxFlushTrigger — NOT on outboxStore.messages — so
1355 // if connect() sets activeConnection before load() finishes reading a (possibly
1356 // multi-MB) outbox.json, the connect-triggered flush sees an empty queue and nothing
1357 // re-triggers it. Nudging the flush here guarantees a loaded queue is sent.
1359 runCatchingCancellable { outboxStore.load() }
1363 appScope.launch { autoConnect() }
1369 * Extract the session id a message-activity event pertains to, or null for events that
1370 * don't describe message activity. This drives the unread badge: any session that
1371 * receives activity while it isn't the currently-viewed one gets badged. Extracted as
1372 * a top-level `internal` function so the rule is unit-testable without an Android
1373 * [Context] (the surrounding [AppContainer] needs one).
1375internal fun sessionOfEvent(event: BusEvent): String? = when (event) {
1376 is MessageUpdated -> event.properties.info.sessionID
1377 is MessagePartUpdated -> event.properties.part.sessionID ?: event.properties.sessionID
1382 * The message id a message-activity event pertains to, used to de-duplicate the unread
1383 * badge so a reply's many streaming events count once. Null for events that carry no
1384 * message id (those fall back to being counted individually).
1386internal fun messageIdOfEvent(event: BusEvent): String? = when (event) {
1387 is MessageUpdated -> event.properties.info.id
1388 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