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