git.iko.soy/opencode-android
on 8bc2cce help sign in
opencode-android/app/src/main/java/soy/iko/opencode/di/AppContainer.kt
643 lines 33 kB history raw
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
8import android.util.Log
9import soy.iko.opencode.data.model.AssistantMessage
10import soy.iko.opencode.data.model.BusEvent
11import soy.iko.opencode.data.model.MessagePartUpdated
12import soy.iko.opencode.data.model.MessageUpdated
13import soy.iko.opencode.data.model.StepFinishPart
14import soy.iko.opencode.data.model.ServerProfile
15import soy.iko.opencode.data.model.SessionIdle
16import soy.iko.opencode.data.repo.DraftStore
17import soy.iko.opencode.data.repo.ErrorKind
18import soy.iko.opencode.data.repo.ProfileStore
19import soy.iko.opencode.data.repo.SettingsStore
20import soy.iko.opencode.data.repo.classifyError
21import soy.iko.opencode.data.repo.responseStatusCode
22import soy.iko.opencode.data.network.HttpClientFactory
23import soy.iko.opencode.data.network.NetworkConfig
24import soy.iko.opencode.data.network.OpencodeApiClient
25import soy.iko.opencode.notification.NotificationChannels
26import soy.iko.opencode.util.safeExceptionSummary
27import soy.iko.opencode.notification.SessionNotifications
28import soy.iko.opencode.R
29import soy.iko.opencode.util.runCatchingCancellable
30import kotlinx.coroutines.CoroutineScope
31import kotlinx.coroutines.Dispatchers
32import kotlinx.coroutines.ExperimentalCoroutinesApi
33import kotlinx.coroutines.SupervisorJob
34import kotlinx.coroutines.cancel
35import kotlinx.coroutines.delay
36import kotlinx.coroutines.flow.MutableStateFlow
37import kotlinx.coroutines.flow.StateFlow
38import kotlinx.coroutines.flow.asStateFlow
39import kotlinx.coroutines.flow.emptyFlow
40import kotlinx.coroutines.flow.first
41import kotlinx.coroutines.Job
42import kotlinx.coroutines.flow.flatMapLatest
43import kotlinx.coroutines.flow.update
44import kotlinx.coroutines.isActive
45import kotlinx.coroutines.launch
46import kotlinx.coroutines.sync.Mutex
47import kotlinx.coroutines.sync.withLock
48import kotlinx.coroutines.withTimeoutOrNull
49import java.util.concurrent.atomic.AtomicBoolean
51/**
52 * Result of probing a server URL to check connectivity and whether authentication
53 * is required, without persisting a profile or opening a long-lived connection.
54 */
55sealed class ProbeResult {
56 /** Server is reachable and accepts unauthenticated requests. */
57 object Reachable : ProbeResult()
59 /** Server is reachable but rejected the probe with an authentication error. */
60 object NeedsAuth : ProbeResult()
62 /** Server could not be reached (host unresolved, connection refused, timeout, etc.). */
63 data class Unreachable(val error: String) : ProbeResult()
66/**
67 * Hand-written service locator held by [soy.iko.opencode.OpencodeApp]. Owns the
68 * process-wide singletons and the currently active [OpencodeConnection].
69 */
70open class AppContainer private constructor(
71 private val appContext: Context?,
72 private val skipInit: Boolean,
73 @Suppress("unused") private val testMode: Boolean,
74) {
75 constructor(context: Context) : this(context.applicationContext, false, false)
76 protected constructor() : this(null, true, true)
78 private val appScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
79 open val profileStore: ProfileStore by lazy { ProfileStore(appContext!!) }
80 open val settingsStore: SettingsStore by lazy { SettingsStore(appContext!!) }
81 open val draftStore: DraftStore by lazy { DraftStore(appContext!!, appScope) }
83 private val _activeConnection = MutableStateFlow<OpencodeConnection?>(null)
84 open val activeConnection: StateFlow<OpencodeConnection?> = _activeConnection.asStateFlow()
86 /** True while a background auto-reconnect to the most-recent server is in flight. */
87 private val _reconnecting = MutableStateFlow(false)
88 open val reconnecting: StateFlow<Boolean> = _reconnecting.asStateFlow()
90 /**
91 * Set once a startup auto-reconnect succeeds. The server list collects this and
92 * (once) navigates straight to the session list so a returning user skips it.
93 * [consumeAutoConnect] guards against re-firing on screen re-entry.
94 */
95 private val _autoConnectDone = MutableStateFlow(false)
96 open val autoConnectDone: StateFlow<Boolean> = _autoConnectDone.asStateFlow()
97 private val autoConnectConsumed = AtomicBoolean(false)
98 open fun consumeAutoConnect(): Boolean = autoConnectConsumed.compareAndSet(false, true)
100 /** Text shared into the app (ACTION_SEND), prefilled into a session draft. */
101 private val _pendingShare = MutableStateFlow<String?>(null)
102 open val pendingShare: StateFlow<String?> = _pendingShare.asStateFlow()
103 open fun setPendingShare(text: String) { _pendingShare.value = text }
104 open fun consumePendingShare(): String? {
105 val current = _pendingShare.value ?: return null
106 return if (_pendingShare.compareAndSet(current, null)) current else null
107 }
109 /**
110 * A session id to open from an external trigger (a notification tap or a deep link).
111 * The nav host consumes it once a connection is active.
112 */
113 private val _pendingOpenSession = MutableStateFlow<String?>(null)
114 open val pendingOpenSession: StateFlow<String?> = _pendingOpenSession.asStateFlow()
115 open fun requestOpenSession(id: String) { _pendingOpenSession.value = id }
116 open fun consumePendingOpenSession(): String? {
117 val current = _pendingOpenSession.value ?: return null
118 return if (_pendingOpenSession.compareAndSet(current, null)) current else null
119 }
121 /**
122 * The session the user is currently viewing (or null when not in a chat). Drives the
123 * unread tracker: messages arriving for any *other* session mark it unread, and
124 * opening a session clears its unread state.
125 */
126 private val _currentSession = MutableStateFlow<String?>(null)
127 open val currentSession: StateFlow<String?> = _currentSession.asStateFlow()
129 /** Session ids that received activity while not being viewed, mapped to the count
130 * of unread message events. A count (not just a presence set) lets the session
131 * list badge show "3 unread" instead of a bare dot, so the user can tell whether
132 * one reply or a whole burst arrived. */
133 private val _unread = MutableStateFlow<Map<String, Int>>(emptyMap())
134 open val unread: StateFlow<Map<String, Int>> = _unread.asStateFlow()
136 /** True when any assistant run is actively streaming across all sessions. Drives the
137 * disconnect confirmation on the session list (disconnecting mid-run kills it). */
138 private val _anyRunActive = MutableStateFlow(false)
139 open val anyRunActive: StateFlow<Boolean> = _anyRunActive.asStateFlow()
141 /** Whether the device currently has network connectivity. Distinct from the SSE
142 * connection state so the UI can tell "you're offline" (device) from "server
143 * unreachable" (credentials/host). Updated by the [networkCallback]. */
144 private val _isOnline = MutableStateFlow(true)
145 open val isOnline: StateFlow<Boolean> = _isOnline.asStateFlow()
147 open fun setCurrentSession(id: String?) {
148 _currentSession.value = id
149 if (id != null) {
150 _unread.update { it - id }
151 unreadMessageIds.remove(id)
152 appContext?.let { SessionNotifications.cancel(it, id) }
153 }
154 }
156 /** Restore a session's unread badge after a failed server switch reconnects.
157 * Preserves the prior [count] so a session badged with "5 unread" before the
158 * switch attempt still shows 5 (not 1) after the restore. No-op for the session
159 * the user is currently viewing, since a viewed session is never badged. */
160 open fun restoreUnread(id: String, count: Int) {
161 if (id != _currentSession.value) {
162 _unread.update { it + (id to count) }
163 }
164 }
166 /** Resolve a localized string — view models reach resources through the container. */
167 open fun string(id: Int, vararg formatArgs: Any): String {
168 if (appContext == null) return ""
169 return if (formatArgs.isEmpty()) appContext.getString(id) else appContext.getString(id, *formatArgs)
170 }
172 /**
173 * Convert a throwable into a user-facing message. Classifies the error by concrete
174 * type (network, timeout, HTTP status) rather than string-matching class names, so
175 * the message reflects what actually went wrong without leaking internal URLs/state.
176 */
177 open fun friendlyError(t: Throwable): String =
178 friendlyErrorFor(t, activeConnection.value?.profile?.baseUrl.orEmpty())
180 /**
181 * Same classification as [friendlyError] but accepts an explicit base URL, so callers
182 * that aren't operating on the active connection (e.g. [probeServer]) can still
183 * produce a user-facing message with the right server address.
184 */
185 open fun friendlyErrorFor(t: Throwable, baseUrl: String): String =
186 when (classifyError(t)) {
187 ErrorKind.NOT_REACHABLE, ErrorKind.NETWORK ->
188 string(R.string.error_not_reachable, baseUrl)
189 ErrorKind.TIMEOUT -> string(R.string.error_timeout)
190 ErrorKind.SERVER -> string(R.string.error_server)
191 // Show only the HTTP status (e.g. 401, 404) — never the request URL,
192 // which a ClientRequestException carries in its message and which we
193 // promised not to leak (it can include auth or internal paths).
194 ErrorKind.CLIENT -> responseStatusCode(t)?.let { string(R.string.error_client_status, it) }
195 ?: string(R.string.error_generic)
196 ErrorKind.UNKNOWN -> string(R.string.error_generic)
197 }
199 init {
200 if (!skipInit) {
201 NotificationChannels.create(appContext!!)
202 observeMessageActivity()
203 appScope.launch { autoConnect() }
204 }
205 }
207 /** Release resources held for the process lifetime (network callback, app scope). */
208 open fun shutdown() {
209 // Cancel the app scope first so any coroutine holding the connection mutex
210 // (e.g. a connect() suspended on profileStore) is cancelled and releases the
211 // mutex. Otherwise the runBlocking below would deadlock waiting for a coroutine
212 // that can't be cancelled until after the runBlocking completes.
213 appScope.cancel()
214 // Bounded runBlocking so a stuck mutex or a slow close() can't ANR the app.
215 // 2 seconds is generous: appScope.cancel() already triggered cancellation of
216 // any coroutine holding the mutex; we're just waiting for it to unwind.
217 kotlinx.coroutines.runBlocking {
218 withTimeoutOrNull(2_000) {
219 connectionMutex.withLock {
220 runCatching { _activeConnection.value?.close() }
221 _activeConnection.value = null
222 }
223 }
224 // If the timeout fired, force-clear the connection so it isn't left dangling.
225 _activeConnection.value = null
226 }
227 val cm = appContext?.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
228 networkCallback?.let { runCatching { cm?.unregisterNetworkCallback(it) } }
229 if (!skipInit) soy.iko.opencode.data.repo.CrashLogger.get(appContext!!).shutdown()
230 if (!skipInit) draftStore.shutdown()
231 }
233 /**
234 * Watch the SSE bus for new message activity and badge any session that isn't
235 * currently open. This powers the unread dot on the session list so the user can
236 * tell which conversations got a reply while they were elsewhere.
237 *
238 * Uses [flatMapLatest] so a server switch (which replaces the active connection)
239 * re-subscribes to the new event stream and drops the old one.
240 */
241 @OptIn(ExperimentalCoroutinesApi::class)
242 private fun observeMessageActivity() {
243 appScope.launch {
244 // Retry the observer if it fails for any reason (transient flow exception,
245 // cancellation due to scope issues). Without this, a single failure would
246 // permanently disable unread tracking and completion notifications.
247 while (isActive) {
248 runCatchingCancellable {
249 activeConnection
250 .flatMapLatest { conn -> conn?.events?.events ?: emptyFlow() }
251 .collect { event ->
252 // SessionIdle is not "message activity" (don't badge as unread)
253 // but signals a run finished — fire a completion notification if the
254 // session isn't currently being viewed and was actively streaming.
255 if (event is SessionIdle) {
256 val idleSid = event.properties.sessionID ?: return@collect
257 // Clear the run state unconditionally — even for the session on
258 // screen. Gating the removal on `idleSid != currentSession` (as
259 // this once did) leaks the id in activeRuns and pins anyRunActive
260 // true forever whenever a run finishes while being viewed. Only
261 // the completion *notification* is suppressed for the viewed one.
262 val wasRunning = synchronized(activeRuns) {
263 activeRuns.remove(idleSid).also { removed ->
264 if (removed) _anyRunActive.value = activeRuns.isNotEmpty()
265 }
266 }
267 if (wasRunning && idleSid != _currentSession.value) {
268 // Fire this off the collector's coroutine: notifySessionCompleted
269 // does a network listSessions() to resolve the title, and blocking
270 // the shared-event collector here would stall the SharedFlow,
271 // overflowing its DROP_OLDEST buffer and silently dropping live
272 // parts/updates for every other subscriber (e.g. the message reducer).
273 appScope.launch { notifySessionCompleted(idleSid) }
274 }
275 return@collect
276 }
277 val sid = sessionOf(event) ?: return@collect
278 // Re-read currentSession here, right before mutating the badge state:
279 // the user may have opened this session between when this event was
280 // enqueued and now (setCurrentSession clears its badge and dedup set).
281 // A stale earlier check would race a badge — and a dedup entry — back
282 // in for the session that's actually on screen.
283 if (sid != _currentSession.value) {
284 // Increment the unread count once per distinct *message*, not per
285 // event: a single reply emits many message.updated / message.part.
286 // updated events (one per streamed token, plus cost/token refreshes),
287 // which would otherwise inflate the badge into the dozens/hundreds.
288 val messageId = messageIdOfEvent(event)
289 val counted = unreadMessageIds.getOrPut(sid) {
290 java.util.Collections.synchronizedSet(mutableSetOf())
291 }
292 if (messageId == null || counted.add(messageId)) {
293 _unread.update { current ->
294 // Guard again inside the atomic update lambda so a retry
295 // (or an open that landed just now) can't reintroduce the
296 // badge for the session on screen.
297 if (sid == _currentSession.value) current
298 else current + (sid to (current[sid] ?: 0) + 1)
299 }
300 }
301 }
302 // Track sessions actively streaming so we know which idle events
303 // represent a finished run worth notifying about.
304 if (isLiveRunActivity(event)) {
305 synchronized(activeRuns) {
306 if (activeRuns.size >= activeRunsLimit) {
307 activeRuns.remove(activeRuns.iterator().next())
308 }
309 activeRuns.add(sid)
310 }
311 _anyRunActive.value = true
312 }
313 } }.onFailure { Log.w("AppContainer", "Message activity observer failed, will retry: ${safeExceptionSummary(it)}") }
314 if (!isActive) break
315 delay(NetworkConfig.observerRetryDelayMs)
316 }
317 }
318 }
320 /** True if [event] is live streaming activity worth tracking as an active run. A
321 * trailing message.updated that lands *after* the run finished (final cost/token totals,
322 * carrying a completion time) is NOT live streaming: with no further SessionIdle to
323 * follow it, re-adding the session would pin anyRunActive true indefinitely. */
324 private fun isLiveRunActivity(event: BusEvent): Boolean {
325 if (event !is MessagePartUpdated && event !is MessageUpdated) return false
326 val info = (event as? MessageUpdated)?.properties?.info
327 if (info is AssistantMessage && info.isComplete) return false
328 // A step-finish part is the trailing completion marker for a run (it carries the
329 // final cost/token totals) and arrives after the last stream delta. Like a completed
330 // message.updated, treating it as live would re-add the session to activeRuns with no
331 // following SessionIdle to clear it — pinning anyRunActive true indefinitely when such
332 // a part is replayed/reordered after the run already went idle.
333 val part = (event as? MessagePartUpdated)?.properties?.part
334 if (part is StepFinishPart) return false
335 return true
336 }
338 /** Session ids currently streaming an assistant run (best-effort, in-process). */
339 private val activeRuns: MutableSet<String> = java.util.Collections.synchronizedSet(mutableSetOf())
341 /** Per non-viewed session, the set of message ids already reflected in its unread
342 * count. De-duplicates the many streaming events that share one message id so the
343 * badge counts messages, not deltas. Cleared when a session is viewed or on connect. */
344 private val unreadMessageIds = java.util.concurrent.ConcurrentHashMap<String, MutableSet<String>>()
346 /** Upper bound on [activeRuns] to prevent unbounded growth if SessionIdle never arrives. */
347 private val activeRunsLimit = NetworkConfig.activeRunsLimit
349 /** Guards connect/disconnect so concurrent callers can't leak an old connection. */
350 private val connectionMutex = Mutex()
352 /** Only persist the `lastUsed` timestamp if it's older than this, to avoid
353 * a DataStore + encrypted-prefs write on every rapid reconnect. */
354 private companion object {
355 const val LAST_USED_SAVE_THRESHOLD_MS = 60_000L
356 }
358 /** Resolve the title for [sessionId] (best-effort) and post a completion notification. */
359 private suspend fun notifySessionCompleted(sessionId: String) {
360 val conn = activeConnection.value ?: return
361 val title = runCatchingCancellable {
362 conn.repository.listSessions().firstOrNull { it.id == sessionId }?.displayTitle
363 }.getOrNull() ?: sessionId
364 appContext?.let { SessionNotifications.postCompleted(it, sessionId, title) }
365 }
367 /** Extract the session id an event pertains to, for message-activity events. */
368 private fun sessionOf(event: BusEvent): String? = sessionOfEvent(event)
370 /**
371 * On cold start, transparently reconnect to the most recently used server so a
372 * returning user lands in their session list instead of the empty server screen.
373 */
374 private suspend fun autoConnect() {
375 val recent = runCatchingCancellable { profileStore.profiles.first() }
376 .getOrDefault(emptyList())
377 .firstOrNull() ?: return
378 _reconnecting.value = true
379 val ok = runCatchingCancellable {
380 val conn = connect(recent)
381 conn.api.ping()
382 }.isSuccess
383 _reconnecting.value = false
384 if (ok) _autoConnectDone.value = true else disconnect()
385 }
387 /**
388 * When the device regains connectivity, nudge the active SSE stream to reconnect
389 * immediately instead of waiting out the exponential backoff.
390 */
391 private val networkCallback: ConnectivityManager.NetworkCallback? = registerNetworkMonitor()
393 private fun registerNetworkMonitor(): ConnectivityManager.NetworkCallback? {
394 if (appContext == null) return null
395 val cm = appContext!!.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
396 ?: return null
397 // Seed the online state from the current active network so the indicator is
398 // correct on cold start rather than defaulting to "online" until a callback fires.
399 _isOnline.value = runCatching { cm.activeNetwork != null }.getOrDefault(true)
400 val callback = object : ConnectivityManager.NetworkCallback() {
401 override fun onAvailable(network: Network) { _isOnline.value = true; activeConnection.value?.events?.triggerReconnect() }
402 override fun onLost(network: Network) { _isOnline.value = cm.activeNetwork != null }
403 }
404 // Only match networks that actually provide internet — a capability-less request
405 // fires onAvailable for transports that can't reach the server (and would trigger
406 // spurious reconnects), so require NET_CAPABILITY_INTERNET.
407 val request = NetworkRequest.Builder()
408 .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
409 .build()
410 runCatching { cm.registerNetworkCallback(request, callback) }
411 .onFailure { Log.w("AppContainer", "Failed to register network callback", it) }
412 return callback
413 }
415 /** Open (or re-open) a connection to [profile], replacing any current one.
416 * Only persists the updated `lastUsed` timestamp if it's stale (older than a
417 * minute), so rapid reconnect storms don't each trigger a DataStore write. */
418 open suspend fun connect(profile: ServerProfile): OpencodeConnection =
419 connectionMutex.withLock {
420 _activeConnection.value?.close()
421 _activeConnection.value = null
422 activeRuns.clear()
423 _anyRunActive.value = false
424 _unread.value = emptyMap()
425 unreadMessageIds.clear()
426 val now = System.currentTimeMillis()
427 val resolved = profileStore.resolve(profile)
428 val needsSave = (now - resolved.lastUsed) > LAST_USED_SAVE_THRESHOLD_MS
429 val finalProfile = if (needsSave) resolved.copy(lastUsed = now) else resolved
430 if (needsSave) profileStore.save(finalProfile)
431 OpencodeConnection(finalProfile).also { _activeConnection.value = it }
432 }
434 open suspend fun disconnect() {
435 // Acquire the connection mutex so disconnect is serialized with connect().
436 // Previously this was fire-and-forget (launch on appScope), which raced with
437 // a subsequent connect(): connect() could acquire the mutex, create a new
438 // connection, and then the pending disconnect() would close it.
439 connectionMutex.withLock {
440 _activeConnection.value?.close()
441 _activeConnection.value = null
442 // Reset run/unread state on disconnect too. With no SSE stream there's no
443 // SessionIdle to drain activeRuns, so without this the working indicator
444 // (anyRunActive) and unread badges would stay pinned until the next connect().
445 activeRuns.clear()
446 _anyRunActive.value = false
447 _unread.value = emptyMap()
448 unreadMessageIds.clear()
449 }
450 }
452 /** Deferred profile deletions keyed by id, run on the process-lived [appScope] (not a
453 * ViewModel scope) so navigating away during the undo window still commits the delete
454 * instead of silently cancelling it — mirroring [CrashLogger.scheduleDelete]. */
455 private val pendingProfileDeletes = java.util.concurrent.ConcurrentHashMap<String, Job>()
457 /**
458 * Schedule the profile [id] for deletion after [delayMs], cancellable via
459 * [cancelProfileDelete] (the Undo action). [onDeleted]/[onError] run on [appScope]
460 * after the delete resolves so the caller can clear its optimistic-hide state. A
461 * repeated schedule for the same id replaces the prior timer.
462 */
463 open fun scheduleProfileDelete(
464 id: String,
465 delayMs: Long,
466 onDeleted: () -> Unit,
467 onError: (Throwable) -> Unit,
468 ) {
469 val job = appScope.launch {
470 delay(delayMs)
471 // Claim the delete by removing our own entry BEFORE committing it — the atomic
472 // remove is the sync point with cancelProfileDelete. Losing the race (an undo
473 // removed us first) means we skip, honouring cancel's `true`; deleting first
474 // would let a cancel land in the gap and falsely report undo of a gone profile.
475 // remove(id, thisJob) also no-ops for a stale job after a reschedule.
476 val claimed = coroutineContext[Job]?.let { pendingProfileDeletes.remove(id, it) } == true
477 if (!claimed) return@launch
478 runCatchingCancellable { profileStore.delete(id) }
479 .onSuccess { onDeleted() }
480 .onFailure { onError(it) }
481 }
482 pendingProfileDeletes.put(id, job)?.cancel()
483 }
485 /** Cancel a pending deferred profile delete (the Undo action). Returns true if it was
486 * still pending (undo succeeded), false if the delete had already fired. */
487 open fun cancelProfileDelete(id: String): Boolean {
488 val job = pendingProfileDeletes.remove(id) ?: return false
489 job.cancel()
490 return true
491 }
493 /** Deferred session deletions keyed by id, run on the process-lived [appScope] (not a
494 * ViewModel scope) so navigating away during the undo window still commits the delete
495 * instead of silently cancelling it (which would leave the session deleted-in-UI but
496 * alive on the server, reappearing on the next refresh). */
497 private val pendingSessionDeletes = java.util.concurrent.ConcurrentHashMap<String, Job>()
499 /**
500 * Schedule session [id] for deletion after [delayMs], cancellable via [cancelSessionDelete]
501 * (the Undo action). Also clears the session's stored draft on a successful delete.
502 * [onDeleted]/[onError] run on [appScope] after the delete resolves so the caller can update
503 * its optimistic-hide state. A repeated schedule for the same id replaces the prior timer.
504 *
505 * The owning connection is captured now, at schedule time — NOT re-resolved when the timer
506 * fires. Otherwise a server switch during the undo window would send the delete to whichever
507 * server happens to be active at fire time: the original session would never be deleted (its
508 * row reappears on the next refresh) and, in the rare id-collision case, an unrelated session
509 * on the new server could be deleted instead. A disconnect closes the captured connection's
510 * HTTP client, so a delete deferred across a disconnect fails and is surfaced via [onError]
511 * (the session cannot be deleted on a server we're no longer connected to). No-ops (as an
512 * error) if there is no active connection at schedule time.
513 */
514 open fun scheduleSessionDelete(
515 id: String,
516 delayMs: Long,
517 onDeleted: () -> Unit,
518 onError: (Throwable) -> Unit,
519 ) {
520 val conn = _activeConnection.value
521 val job = appScope.launch {
522 delay(delayMs)
523 // Claim the delete by removing our own entry BEFORE committing it — the atomic
524 // remove is the sync point with cancelSessionDelete. Losing the race (an undo
525 // removed us first) means we skip, honouring cancel's `true`; deleting first
526 // would let a cancel land in the gap and falsely report undo of a gone session.
527 // remove(id, thisJob) also no-ops for a stale job after a reschedule.
528 val claimed = coroutineContext[Job]?.let { pendingSessionDeletes.remove(id, it) } == true
529 if (!claimed) return@launch
530 if (conn == null) {
531 onError(IllegalStateException("No active connection"))
532 } else {
533 runCatchingCancellable { conn.repository.deleteSession(id) }
534 .onSuccess {
535 draftStore.remove(id)
536 onDeleted()
537 }
538 .onFailure { onError(it) }
539 }
540 }
541 pendingSessionDeletes.put(id, job)?.cancel()
542 }
544 /** Cancel a pending deferred session delete (the Undo action). Returns true if it was
545 * still pending (undo succeeded), false if the delete had already fired. */
546 open fun cancelSessionDelete(id: String): Boolean {
547 val job = pendingSessionDeletes.remove(id) ?: return false
548 job.cancel()
549 return true
550 }
552 /**
553 * Probe a server URL (without credentials) to check reachability and detect whether
554 * the server requires authentication. Builds a short-lived HTTP client with the same
555 * base URL normalization as [connect] but no auth, calls the health endpoint, and
556 * classifies the outcome:
557 *
558 * - 2xx → [ProbeResult.Reachable] (no auth needed)
559 * - 401/403 → [ProbeResult.NeedsAuth]
560 * - anything else → [ProbeResult.Unreachable] with a user-facing message
561 *
562 * The probe client is always closed afterwards so no resources linger. This does not
563 * touch the active connection or the profile store.
564 */
565 open suspend fun probeServer(baseUrl: String): ProbeResult {
566 return probeWithProfile(ServerProfile(
567 id = "probe",
568 label = "",
569 baseUrl = baseUrl.trim(),
570 username = null,
571 password = null,
572 ))
573 }
575 /**
576 * Probe a server URL *with* credentials to validate them. Returns true if the
577 * server accepts the credentials (ping succeeds), false on 401/403, and throws
578 * on other failures (unreachable, timeout) so the caller can surface a friendly
579 * message. Mirrors [probeServer]'s short-lived-client pattern.
580 */
581 open suspend fun probeWithCredentials(baseUrl: String, username: String, password: String): Boolean {
582 val probeProfile = ServerProfile(
583 id = "probe-auth",
584 label = "",
585 baseUrl = baseUrl.trim(),
586 username = username.trim().takeIf { it.isNotBlank() },
587 password = password.trim().takeIf { it.isNotEmpty() },
588 )
589 val client = HttpClientFactory.create(probeProfile)
590 return try {
591 val api = OpencodeApiClient(client)
592 api.ping()
593 true
594 } catch (e: Exception) {
595 val status = responseStatusCode(e)
596 if (status == 401 || status == 403) false else throw e
597 } finally {
598 runCatching { client.close() }
599 }
600 }
602 private suspend fun probeWithProfile(profile: ServerProfile): ProbeResult {
603 val client = HttpClientFactory.create(profile)
604 return try {
605 val api = OpencodeApiClient(client)
606 api.ping()
607 ProbeResult.Reachable
608 } catch (e: Exception) {
609 val status = responseStatusCode(e)
610 if (status == 401 || status == 403) {
611 ProbeResult.NeedsAuth
612 } else {
613 ProbeResult.Unreachable(friendlyErrorFor(e, profile.baseUrl))
614 }
615 } finally {
616 runCatching { client.close() }
617 }
618 }
621/**
622 * Extract the session id a message-activity event pertains to, or null for events that
623 * don't describe message activity. This drives the unread badge: any session that
624 * receives activity while it isn't the currently-viewed one gets badged. Extracted as
625 * a top-level `internal` function so the rule is unit-testable without an Android
626 * [Context] (the surrounding [AppContainer] needs one).
627 */
628internal fun sessionOfEvent(event: BusEvent): String? = when (event) {
629 is MessageUpdated -> event.properties.info.sessionID
630 is MessagePartUpdated -> event.properties.part.sessionID ?: event.properties.sessionID
631 else -> null
634/**
635 * The message id a message-activity event pertains to, used to de-duplicate the unread
636 * badge so a reply's many streaming events count once. Null for events that carry no
637 * message id (those fall back to being counted individually).
638 */
639internal fun messageIdOfEvent(event: BusEvent): String? = when (event) {
640 is MessageUpdated -> event.properties.info.id
641 is MessagePartUpdated -> event.properties.messageID ?: event.properties.part.messageID
642 else -> null