@@ -4,10 +4,11 @@ import soy.iko.opencode.data.model.BusEvent
44import android.util.Log
55import io.ktor.client.HttpClient
66import io.ktor.client.plugins.sse.sse
7+import io.ktor.sse.ServerSentEvent
78import kotlinx.coroutines.CancellationException
89import kotlinx.coroutines.CoroutineScope
910import kotlinx.coroutines.channels.Channel
10−import kotlinx.coroutines.delay
11+import kotlinx.coroutines.channels.ChannelResult
1112import kotlinx.coroutines.flow.Flow
1213import kotlinx.coroutines.flow.MutableStateFlow
1314import kotlinx.coroutines.flow.SharedFlow
@@ -15,10 +16,12 @@ import kotlinx.coroutines.flow.StateFlow
1516import kotlinx.coroutines.flow.SharingStarted
1617import kotlinx.coroutines.flow.asStateFlow
1718import kotlinx.coroutines.flow.channelFlow
19+import kotlinx.coroutines.flow.produceIn
1820import kotlinx.coroutines.flow.shareIn
1921import kotlinx.coroutines.isActive
2022import kotlinx.coroutines.selects.onTimeout
2123import kotlinx.coroutines.selects.select
24+import java.io.IOException
2225
2326/**
2427 * Owns the single long-lived `GET /event` SSE subscription and exposes it as a hot
@@ -52,22 +55,45 @@ class EventStreamClient(
5255 fun triggerReconnect() { reconnectSignal.trySend(Unit) }
5356
5457 private fun stream(): Flow<BusEvent> = channelFlow {
58+ val scope = this
5559 var backoffMs = INITIAL_BACKOFF_MS
5660 while (isActive) {
5761 _state.value = ConnectionState.Connecting
5862 try {
5963 client.sse("event") {
6064 backoffMs = INITIAL_BACKOFF_MS
61− incoming.collect { sse ->
62− val data = sse.data ?: return@collect
63− val event = runCatching { OpencodeJson.decodeFromString(BusEvent.serializer(), data) }
64− .getOrNull() ?: return@collect
65− // Any successfully decoded event means the stream is live;
66− // don't wait for a ServerConnected event to show "Connected".
67− if (_state.value != ConnectionState.Connected) {
68− _state.value = ConnectionState.Connected
65+ // `incoming` is a cold Flow; bridge it to a ReceiveChannel so each
66+ // element can be raced against an idle timeout via select.
67+ val events = incoming.produceIn(scope)
68+ try {
69+ while (isActive) {
70+ // The OkHttp engine is configured with readTimeout=0 (the SSE
71+ // stream must never time out on read), so a socket the peer never
72+ // closed would otherwise hang until the OS TCP keepalive kicks in
73+ // — minutes to hours on stock Linux. If nothing arrives within
74+ // IDLE_TIMEOUT_MS, treat the connection as half-open and drop it
75+ // so the outer loop reconnects.
76+ val result: ChannelResult<ServerSentEvent>? = select {
77+ events.onReceiveCatching { it }
78+ onTimeout(IDLE_TIMEOUT_MS) { null }
79+ }
80+ // Timeout: drop and reconnect.
81+ if (result == null) throw IOException("SSE idle timeout, reconnecting")
82+ // Stream closed cleanly: exit to reconnect without a warning log.
83+ val sse = result.getOrNull() ?: break
84+ val data = sse.data ?: continue
85+ val event = runCatching {
86+ OpencodeJson.decodeFromString(BusEvent.serializer(), data)
87+ }.getOrNull() ?: continue
88+ // Any successfully decoded event means the stream is live;
89+ // don't wait for a ServerConnected event to show "Connected".
90+ if (_state.value != ConnectionState.Connected) {
91+ _state.value = ConnectionState.Connected
92+ }
93+ send(event)
6994 }
70− send(event)
95+ } finally {
96+ events.cancel()
7197 }
7298 }
7399 } catch (c: CancellationException) {
@@ -91,5 +117,7 @@ class EventStreamClient(
91117 private const val TAG = "EventStream"
92118 const val INITIAL_BACKOFF_MS = 500L
93119 const val MAX_BACKOFF_MS = 10_000L
120+ /** Max gap between events before a silent/half-open connection is dropped. */
121+ const val IDLE_TIMEOUT_MS = 90_000L
94122 }
95123}