@@ -67,6 +67,41 @@ class EventStreamClient(
6767 /** Request an immediate reconnect, skipping any in-progress backoff. */
6868 fun triggerReconnect() { reconnectSignal.trySend(Unit) }
6969
70+ /**
71+ * Read SSE events from [events], racing each read against an idle-timeout watchdog.
72+ * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
73+ * clean stream close or scope cancellation.
74+ */
75+ private suspend fun readSseEvents(
76+ scope: CoroutineScope,
77+ events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
78+ send: suspend (BusEvent) -> Unit,
79+ ) {
80+ while (scope.isActive) {
81+ // The OkHttp engine has readTimeout=0 (the SSE stream must never time out on
82+ // read), so a socket the peer never closed would otherwise hang until OS TCP
83+ // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
84+ // within the idle timeout, treat the connection as half-open and drop it.
85+ val result: ChannelResult<ServerSentEvent>? = select {
86+ events.onReceiveCatching { it }
87+ onTimeout(NetworkConfig.sseIdleTimeoutMs) { null }
88+ }
89+ if (result == null) throw IOException("SSE idle timeout, reconnecting")
90+ val sse = result.getOrNull() ?: break
91+ val data = sse.data ?: continue
92+ val event = try {
93+ OpencodeJson.decodeFromString(BusEvent.serializer(), data)
94+ } catch (e: kotlinx.serialization.SerializationException) {
95+ Log.d("EventStream", "Skipping unparseable SSE event", e)
96+ continue
97+ }
98+ if (_state.value != ConnectionState.Connected) {
99+ _state.value = ConnectionState.Connected
100+ }
101+ send(event)
102+ }
103+ }
104+
70105 private fun stream(): Flow<BusEvent> = channelFlow {
71106 val scope = this
72107 // Drain any stale reconnect signal from a previous collection cycle so it
@@ -86,10 +121,13 @@ class EventStreamClient(
86121 request = {
87122 timeout {
88123 requestTimeoutMillis = Long.MAX_VALUE
89− // Use a finite socket timeout so a server that accepts TCP but
90− // never sends SSE headers doesn't hang indefinitely. Once events
91− // are flowing, the idle-timeout watchdog handles half-open detection.
92− socketTimeoutMillis = NetworkConfig.sseHandshakeTimeoutMs
124+ // Disable the socket (read) timeout entirely for the SSE stream.
125+ // A finite value would kill the connection whenever the server
126+ // is quiet for longer than that (idle sessions, slow agent runs),
127+ // causing a reconnect storm. Half-open detection is handled by
128+ // the idle-timeout watchdog below plus OkHttp's pingInterval on
129+ // HTTP/2 connections.
130+ socketTimeoutMillis = Long.MAX_VALUE
93131 }
94132 },
95133 ) {
@@ -102,35 +140,7 @@ class EventStreamClient(
102140 // element can be raced against an idle timeout via select.
103141 val events = incoming.produceIn(scope)
104142 try {
105− while (isActive) {
106− // The OkHttp engine has readTimeout=0 (the SSE stream must
107− // never time out on read), so a socket the peer never closed
108− // would otherwise hang until OS TCP keepalive kicks in —
109− // minutes to hours on stock Linux. If nothing arrives within
110− // the idle timeout, treat the connection as half-open and
111− // drop it so the outer loop reconnects.
112− val result: ChannelResult<ServerSentEvent>? = select {
113− events.onReceiveCatching { it }
114− onTimeout(NetworkConfig.sseIdleTimeoutMs) { null }
115− }
116− // Timeout: drop and reconnect.
117− if (result == null) throw IOException("SSE idle timeout, reconnecting")
118− // Stream closed cleanly: exit to reconnect without a warning log.
119− val sse = result.getOrNull() ?: break
120− val data = sse.data ?: continue
121− val event = try {
122− OpencodeJson.decodeFromString(BusEvent.serializer(), data)
123− } catch (e: kotlinx.serialization.SerializationException) {
124− Log.d("EventStream", "Skipping unparseable SSE event", e)
125− continue
126− }
127− // Any successfully decoded event means the stream is live;
128− // don't wait for a ServerConnected event to show "Connected".
129− if (_state.value != ConnectionState.Connected) {
130− _state.value = ConnectionState.Connected
131− }
132− send(event)
133− }
143+ readSseEvents(scope, events) { send(it) }
134144 } finally {
135145 events.cancel()
136146 }
@@ -138,9 +148,14 @@ class EventStreamClient(
138148 } catch (c: CancellationException) {
139149 throw c
140150 } catch (e: Exception) {
151+ // If the scope was cancelled (e.g. connection close), the closed-client
152+ // exception is expected — don't log a spurious "stream error" warning.
141153 if (isSseAuthFailure(e)) {
142− Log.w("EventStream", "SSE auth failed, will retry with backoff", e)
143− } else {
154+ // Credentials are wrong — retrying won't help. Log and stop.
155+ Log.w("EventStream", "SSE auth failed (401/403), stopping retries", e)
156+ _state.value = ConnectionState.Disconnected
157+ break
158+ } else if (isActive) {
144159 Log.w("EventStream", "SSE stream error, will retry", e)
145160 }
146161 }