@@ -24,16 +24,19 @@ import soy.iko.opencode.data.repo.BackupManager
24import soy.iko.opencode.data.repo.DraftStore
24import soy.iko.opencode.data.repo.DraftStore
25import soy.iko.opencode.data.repo.ErrorKind
25import soy.iko.opencode.data.repo.ErrorKind
26import soy.iko.opencode.data.repo.MessageCacheStore
26import soy.iko.opencode.data.repo.MessageCacheStore
27+import soy.iko.opencode.data.repo.OutboxMessage
27import soy.iko.opencode.data.repo.OutboxStore
28import soy.iko.opencode.data.repo.OutboxStore
28import soy.iko.opencode.data.repo.SessionPrefsStore
29import soy.iko.opencode.data.repo.SessionPrefsStore
29import soy.iko.opencode.data.repo.ProfileStore
30import soy.iko.opencode.data.repo.ProfileStore
30import soy.iko.opencode.data.repo.SettingsStore
31import soy.iko.opencode.data.repo.SettingsStore
31import soy.iko.opencode.data.repo.classifyError
32import soy.iko.opencode.data.repo.classifyError
32import soy.iko.opencode.data.repo.responseStatusCode
33import soy.iko.opencode.data.repo.responseStatusCode
34+import soy.iko.opencode.data.network.EventStreamClient
33import soy.iko.opencode.data.network.HttpClientFactory
35import soy.iko.opencode.data.network.HttpClientFactory
34import soy.iko.opencode.data.network.NetworkConfig
36import soy.iko.opencode.data.network.NetworkConfig
35import soy.iko.opencode.data.network.OpencodeApiClient
37import soy.iko.opencode.data.network.OpencodeApiClient
36import soy.iko.opencode.notification.NotificationChannels
38import soy.iko.opencode.notification.NotificationChannels
39+import soy.iko.opencode.notification.RunForegroundService
37import soy.iko.opencode.util.safeExceptionSummary
40import soy.iko.opencode.util.safeExceptionSummary
38import soy.iko.opencode.notification.SessionNotifications
41import soy.iko.opencode.notification.SessionNotifications
39import soy.iko.opencode.R
42import soy.iko.opencode.R
@@ -47,7 +50,10 @@ import kotlinx.coroutines.delay
47import kotlinx.coroutines.flow.MutableStateFlow
50import kotlinx.coroutines.flow.MutableStateFlow
48import kotlinx.coroutines.flow.StateFlow
51import kotlinx.coroutines.flow.StateFlow
49import kotlinx.coroutines.flow.asStateFlow
52import kotlinx.coroutines.flow.asStateFlow
53+import kotlinx.coroutines.flow.collectLatest
50import kotlinx.coroutines.flow.combine
54import kotlinx.coroutines.flow.combine
55+import kotlinx.coroutines.flow.debounce
56+import kotlinx.coroutines.flow.distinctUntilChanged
51import kotlinx.coroutines.flow.emptyFlow
57import kotlinx.coroutines.flow.emptyFlow
52import kotlinx.coroutines.flow.first
58import kotlinx.coroutines.flow.first
53import kotlinx.coroutines.Job
59import kotlinx.coroutines.Job
@@ -215,6 +221,7 @@ open class AppContainer private constructor(
215 appContext?.let {
221 appContext?.let {
216 SessionNotifications.cancel(it, id)
222 SessionNotifications.cancel(it, id)
217 SessionNotifications.cancelPermission(it, id)
223 SessionNotifications.cancelPermission(it, id)
224+ SessionNotifications.cancelError(it, id)
@@ -262,13 +269,62 @@ open class AppContainer private constructor(
262 ErrorKind.UNKNOWN -> string(R.string.error_generic)
269 ErrorKind.UNKNOWN -> string(R.string.error_generic)
266− if (!skipInit) {
273+ * Drive the [RunForegroundService] from the process-lived [appScope] rather than from the
267− NotificationChannels.create(appContext!!)
274+ * Activity's composition. Previously the start/stop was wired to a
268− observeMessageActivity()
275+ * `collectAsStateWithLifecycle` collector in the root composable, which pauses while the
269− appScope.launch { runCatchingCancellable { outboxStore.load() } }
276+ * Activity is STOPPED — so a run *started while backgrounded* (e.g. via the notification
270− observeOutbox()
277+ * inline-reply) never acquired foreground priority and its long-lived SSE stream could be
278+ * killed by Doze, and a run that *ended* while backgrounded left the "working…"
279+ * notification lingering until the app was reopened. Collecting here reacts to
280+ * [anyRunActive] regardless of Activity state.
282+ private fun observeRunForegroundService() {
283+ val ctx = appContext ?: return
285+ // Debounce so a run that starts then fails/idles almost immediately doesn't
286+ // dispatch startForegroundService() then stopService() back-to-back — if the stop
287+ // wins that race before the service calls startForeground(), Android raises the
288+ // "did not then call startForeground()" crash ~5s later. Collapsing the window
289+ // emits only the settled state; distinctUntilChanged drops a no-op re-emit when a
290+ // brief true→false→true flurry settles back to the value already applied.
292+ .debounce(NetworkConfig.runForegroundDebounceMs)
293+ .distinctUntilChanged()
294+ .collect { active ->
295+ if (active) RunForegroundService.start(ctx) else RunForegroundService.stop(ctx)
301+ * Reconcile active-run tracking on an SSE *reconnect*. A run that completes during a stream
302+ * outage never delivers its `SessionIdle`, so its id would stay in [activeRuns] and pin
303+ * [anyRunActive] (and the foreground service + "working…" notification) on indefinitely. On a
304+ * reconnect we clear the tracking and let live streaming events re-populate it: a genuinely
305+ * in-progress run re-adds itself within moments via [isLiveRunActivity], while a run that
306+ * finished during the outage produces no further live events (its completion arrives only via
307+ * the REST re-seed), so it correctly stays cleared. The first connect of a connection is not a
308+ * reconnect; `connect()` already resets [activeRuns], so only subsequent Connected transitions
309+ * trigger the sweep.
311+ private fun observeRunReconcileOnReconnect() {
271− appScope.launch { autoConnect() }
312+ appScope.launch {
313+ activeConnection.collectLatest { conn ->
314+ if (conn == null) return@collectLatest
315+ var hasConnectedBefore = false
316+ conn.events.state.collect { state ->
317+ if (state == EventStreamClient.ConnectionState.Connected) {
318+ if (hasConnectedBefore) {
319+ synchronized(activeRuns) {
321+ _anyRunActive.value = false
324+ hasConnectedBefore = true
@@ -277,9 +333,9 @@ open class AppContainer private constructor(
277 * online (and on any manual [flushOutbox] nudge). Queued messages are only sent to the
333 * online (and on any manual [flushOutbox] nudge). Queued messages are only sent to the
278 * server they were composed for (matched by profile id), so a reconnect to a different
334 * server they were composed for (matched by profile id), so a reconnect to a different
279 * server never misfires them against a session id that doesn't exist there. Uses a plain
335 * server never misfires them against a session id that doesn't exist there. Uses a plain
280− * `collect` (not collectLatest) so an in-flight send is never cancelled mid-flight — a
336+ * `collect` (not collectLatest) so an in-flight send is never cancelled mid-flight and
281− * cancelled+retried POST could double-send since each attempt carries a fresh idempotency
337+ * send order is preserved; each message carries a stable idempotency key (its persistent
282− * key.
338+ * outbox id) so even a re-send across flushes is deduplicated server-side.
284 private fun observeOutbox() {
340 private fun observeOutbox() {
285 appScope.launch {
341 appScope.launch {
@@ -299,7 +355,7 @@ open class AppContainer private constructor(
299 _outboxSending.value = true
355 _outboxSending.value = true
301 for (msg in pending) {
357 for (msg in pending) {
302− val sent = runCatchingCancellable {
358+ val result = runCatchingCancellable {
303 conn.repository.sendPrompt(
359 conn.repository.sendPrompt(
304 msg.sessionId,
360 msg.sessionId,
305 msg.text,
361 msg.text,
@@ -308,14 +364,38 @@ open class AppContainer private constructor(
309 model = msg.model,
365 model = msg.model,
310 agent = msg.agent,
366 agent = msg.agent,
367+ // Reuse the persistent outbox id as a stable idempotency key so a
368+ // later flush of a message whose earlier POST reached the server (but
369+ // whose response was lost) is deduplicated server-side instead of
370+ // starting a duplicate agent run.
371+ idempotencyKey = msg.id,
312 }.onFailure {
373 }.onFailure {
313 Log.w("AppContainer", "Outbox flush failed for ${msg.sessionId}: ${safeExceptionSummary(it)}")
374 Log.w("AppContainer", "Outbox flush failed for ${msg.sessionId}: ${safeExceptionSummary(it)}")
314− }.isSuccess
376+ if (result.isSuccess) {
315− // Stop on the first failure — the connection likely dropped again; the
377+ outboxStore.remove(msg.id)
380+ // A permanently-undeliverable message (a non-408/429 4xx — e.g. the target
381+ // session was deleted server-side, yielding a 404) must be dropped, not left at
382+ // the head of the queue. The loop stops at the first failure to preserve send
383+ // order, so a poison message would otherwise block every later queued message
384+ // for this profile forever. Transient failures (offline again, 5xx, 408/429)
316− // remaining messages stay queued and retry on the next trigger, preserving
385+ // keep the queue intact and retry on the next trigger.
317− // send order.
386+ val status = result.exceptionOrNull()?.let { responseStatusCode(it) }
387+ // Permanent = a non-408/429 4xx, matching withRetry's non-retryable rule.
388+ val permanent = when (status) {
389+ null, 408, 429 -> false
394+ Log.w("AppContainer", "Dropping undeliverable outbox message ${msg.id} for ${msg.sessionId} (HTTP $status)")
318− if (sent) outboxStore.remove(msg.id) else break
395+ outboxStore.remove(msg.id)
320 } finally {
400 } finally {
321 _outboxSending.value = false
401 _outboxSending.value = false
@@ -451,15 +531,18 @@ open class AppContainer private constructor(
451 if (isLiveRunActivity(event)) {
531 if (isLiveRunActivity(event)) {
452 // The add-under-lock is required both for thread-safety and for the
532 // The add-under-lock is required both for thread-safety and for the
453 // LRU access-order refresh (keeps a continuously-streaming session
533 // LRU access-order refresh (keeps a continuously-streaming session
454− // from being evicted), so it runs per event. Only the flag write is
534+ // from being evicted), so it runs per event.
455− // skippable once it's already set.
456 synchronized(activeRuns) {
535 synchronized(activeRuns) {
457 if (activeRuns.size >= activeRunsLimit) {
536 if (activeRuns.size >= activeRunsLimit) {
458 activeRuns.remove(activeRuns.iterator().next())
537 activeRuns.remove(activeRuns.iterator().next())
460 activeRuns.add(sid)
539 activeRuns.add(sid)
540+ // Derive the flag from the set state *inside* the same lock that
541+ // guards the set, so a concurrent reconnect sweep (which clears
542+ // both under this lock) can't interleave and leave the flag pinned
543+ // true with an empty set — a stuck foreground service + "working…".
544+ _anyRunActive.value = activeRuns.isNotEmpty()
462− if (!_anyRunActive.value) _anyRunActive.value = true
464 } }.onFailure { Log.w("AppContainer", "Message activity observer failed, will retry: ${safeExceptionSummary(it)}") }
547 } }.onFailure { Log.w("AppContainer", "Message activity observer failed, will retry: ${safeExceptionSummary(it)}") }
465 if (!isActive) break
548 if (!isActive) break
@@ -486,8 +569,15 @@ open class AppContainer private constructor(
486 return true
569 return true
489− /** Session ids currently streaming an assistant run (best-effort, in-process). */
572+ /** Session ids currently streaming an assistant run (best-effort, in-process). Backed by
573+ * an access-order [java.util.LinkedHashMap] so re-adding an already-present id on each
574+ * streaming event refreshes its recency (a plain LinkedHashSet.add is a no-op that does
575+ * NOT reorder). This keeps a long, continuously-streaming session from being evicted as
576+ * the eldest once [activeRunsLimit] other sessions appear — which would drop its
577+ * completion notification when its SessionIdle finally arrives. */
490− private val activeRuns: MutableSet<String> = java.util.Collections.synchronizedSet(mutableSetOf())
578+ private val activeRuns: MutableSet<String> = java.util.Collections.synchronizedSet(
579+ java.util.Collections.newSetFromMap(java.util.LinkedHashMap<String, Boolean>(16, 0.75f, true)),
492 /** Per non-viewed session, the set of message ids already reflected in its unread
582 /** Per non-viewed session, the set of message ids already reflected in its unread
493 * count. De-duplicates the many streaming events that share one message id so the
583 * count. De-duplicates the many streaming events that share one message id so the
@@ -533,29 +623,63 @@ open class AppContainer private constructor(
534 /** Respond to a permission from a notification action (Allow once / Always / Reject),
624 /** Respond to a permission from a notification action (Allow once / Always / Reject),
535 * off any UI. Runs on the process-lived app scope; [onDone] fires when the call resolves
625 * off any UI. Runs on the process-lived app scope; [onDone] fires when the call resolves
536− * so the receiver's goAsync() result can finish. No-ops if the connection is gone. */
626+ * so the receiver's goAsync() result can finish. Its Boolean is true only when a live
627+ * connection existed AND respondPermission succeeded — the receiver uses it to decide
628+ * whether to dismiss the notification. A missing connection makes the respond a no-op,
629+ * so it must report failure (not success) rather than let the notification vanish with
630+ * the tool left unanswered. */
537 open fun respondToPermissionFromNotification(
631 open fun respondToPermissionFromNotification(
538 sessionId: String,
632 sessionId: String,
539 permissionId: String,
633 permissionId: String,
540 response: PermissionResponse,
634 response: PermissionResponse,
541− onDone: () -> Unit,
635+ onDone: (Boolean) -> Unit,
543 appScope.launch {
637 appScope.launch {
544− runCatchingCancellable {
638+ val success = runCatchingCancellable {
639+ val api = activeConnection.value?.api ?: return@runCatchingCancellable false
545− activeConnection.value?.api?.respondPermission(sessionId, permissionId, response)
640+ api.respondPermission(sessionId, permissionId, response)
546− }.onFailure { Log.w("AppContainer", "notif permission respond failed: ${safeExceptionSummary(it)}") }
643+ Log.w("AppContainer", "notif permission respond failed: ${safeExceptionSummary(it)}")
644+ }.getOrDefault(false)
547− onDone()
645+ onDone(success)
551 /** Send a follow-up prompt from a notification's inline reply, off any UI. Uses the
649 /** Send a follow-up prompt from a notification's inline reply, off any UI. Uses the
552− * server's default model/agent. [onDone] fires when the send resolves. */
650+ * server's default model/agent. Routes through the durable [outboxStore] — the same path
651+ * the in-app composer uses ([soy.iko.opencode.ui.chat.ChatViewModel.enqueueOffline]) —
652+ * rather than a direct repository send, so a reply composed while disconnected survives to
653+ * be flushed on reconnect instead of being silently dropped. [flushOutbox] nudges an
654+ * immediate send when a connection is live, so an online reply still goes out now. [onDone]
655+ * reports whether the reply was durably enqueued so the receiver only dismisses the
656+ * notification on success. */
553− open fun sendPromptFromNotification(sessionId: String, text: String, onDone: () -> Unit) {
657+ open fun sendPromptFromNotification(sessionId: String, text: String, onDone: (Boolean) -> Unit) {
554 appScope.launch {
658 appScope.launch {
555− runCatchingCancellable {
659+ // Attribute to the active (or most-recent) profile so a reconnect to a different
660+ // server doesn't misfire this against a session id that only exists elsewhere.
556− activeConnection.value?.repository?.sendPrompt(sessionId, text, model = null)
661+ val profileId = activeConnection.value?.profile?.id
662+ ?: runCatchingCancellable { profileStore.profiles.first().firstOrNull()?.id }.getOrNull()
663+ val enqueued = if (profileId == null) {
666+ runCatchingCancellable {
667+ outboxStore.enqueue(
669+ id = java.util.UUID.randomUUID().toString(),
670+ profileId = profileId,
671+ sessionId = sessionId,
674+ createdAt = System.currentTimeMillis(),
557− }.onFailure { Log.w("AppContainer", "notif reply send failed: ${safeExceptionSummary(it)}") }
678+ Log.w("AppContainer", "notif reply enqueue failed: ${safeExceptionSummary(it)}")
681+ if (enqueued) flushOutbox()
558− onDone()
682+ onDone(enqueued)
@@ -571,12 +695,22 @@ open class AppContainer private constructor(
571 .getOrDefault(emptyList())
695 .getOrDefault(emptyList())
572 .firstOrNull() ?: return
696 .firstOrNull() ?: return
573 _reconnecting.value = true
697 _reconnecting.value = true
698+ var conn: OpencodeConnection? = null
574 val ok = runCatchingCancellable {
699 val ok = runCatchingCancellable {
575− val conn = connect(recent)
700+ conn = connect(recent)
576− conn.api.ping()
701+ conn!!.api.ping()
577 }.isSuccess
702 }.isSuccess
578 _reconnecting.value = false
703 _reconnecting.value = false
579− if (ok) _autoConnectDone.value = true else disconnect()
705+ _autoConnectDone.value = true
708+ // Only tear down the connection we created: a manual connect to a different server
709+ // during the ping window may have already replaced it, and disconnect() closes
710+ // whatever is currently active — which would silently drop that new connection.
711+ // disconnectIf re-checks identity under the mutex to close the check-then-close race.
712+ if (created != null) disconnectIf(created)
@@ -631,19 +765,33 @@ open class AppContainer private constructor(
631 // Previously this was fire-and-forget (launch on appScope), which raced with
765 // Previously this was fire-and-forget (launch on appScope), which raced with
632 // a subsequent connect(): connect() could acquire the mutex, create a new
766 // a subsequent connect(): connect() could acquire the mutex, create a new
633 // connection, and then the pending disconnect() would close it.
767 // connection, and then the pending disconnect() would close it.
768+ connectionMutex.withLock { teardownActiveLocked() }
771+ /** Tear down [expected] only if it is still the active connection, with the identity
772+ * check performed *under* the mutex. A caller that checks `activeConnection.value === x`
773+ * itself and then calls [disconnect] has a TOCTOU race: a concurrent connect() to a
774+ * different server can replace the active connection in the gap, and [disconnect] would
775+ * then close that new connection. autoConnect()'s failed-ping teardown uses this. */
776+ private suspend fun disconnectIf(expected: OpencodeConnection) {
634 connectionMutex.withLock {
777 connectionMutex.withLock {
635− _activeConnection.value?.close()
636− _activeConnection.value = null
778+ if (_activeConnection.value === expected) teardownActiveLocked()
637− // Reset run/unread state on disconnect too. With no SSE stream there's no
638− // SessionIdle to drain activeRuns, so without this the working indicator
639− // (anyRunActive) and unread badges would stay pinned until the next connect().
641− _anyRunActive.value = false
642− _unread.value = emptyMap()
643− unreadMessageIds.clear()
782+ /** Close and reset the active connection. Caller MUST hold [connectionMutex]. */
783+ private suspend fun teardownActiveLocked() {
784+ _activeConnection.value?.close()
785+ _activeConnection.value = null
786+ // Reset run/unread state on disconnect too. With no SSE stream there's no
787+ // SessionIdle to drain activeRuns, so without this the working indicator
788+ // (anyRunActive) and unread badges would stay pinned until the next connect().
790+ _anyRunActive.value = false
791+ _unread.value = emptyMap()
792+ unreadMessageIds.clear()
647 /** Deferred profile deletions keyed by id, run on the process-lived [appScope] (not a
795 /** Deferred profile deletions keyed by id, run on the process-lived [appScope] (not a
648 * ViewModel scope) so navigating away during the undo window still commits the delete
796 * ViewModel scope) so navigating away during the undo window still commits the delete
649 * instead of silently cancelling it — mirroring [CrashLogger.scheduleDelete]. */
797 * instead of silently cancelling it — mirroring [CrashLogger.scheduleDelete]. */
@@ -815,6 +963,22 @@ open class AppContainer private constructor(
815 runCatching { client.close() }
963 runCatching { client.close() }
967+ // Declared last so every property above is fully initialized before these observers start.
968+ // They only launch coroutines into appScope (Dispatchers.IO) — nothing here runs
969+ // synchronously during construction — so placing init at the end removes any dependence on
970+ // the textual declaration order of the fields those observers reference.
973+ NotificationChannels.create(appContext!!)
974+ observeMessageActivity()
975+ observeRunReconcileOnReconnect()
976+ observeRunForegroundService()
977+ appScope.launch { runCatchingCancellable { outboxStore.load() } }
979+ appScope.launch { autoConnect() }