fix(services/bidi): gate reconnect reset on stream health, add exponential backoff

apps/flipcash/app/src/main/kotlin/com/flipcash/app/internal/startup/BugsnagErrorReporter.kt

@@ -1,12 +1,17 @@
 package com.flipcash.app.internal.startup
 
 import com.bugsnag.android.Bugsnag
+import com.bugsnag.android.Severity
 import com.getcode.utils.ErrorReporter
 
 class BugsnagErrorReporter : ErrorReporter {
     override fun report(error: Throwable, cause: Throwable, isNotifiable: Boolean) {
-        if (!isNotifiable) return
         if (!Bugsnag.isStarted()) return
-        Bugsnag.notify(error)
+        Bugsnag.notify(error) { event ->
+            // WARNING = needs active attention (also drives the Slack filter);
+            // INFO = recorded for reference only, excluded from Slack.
+            event.severity = if (isNotifiable) Severity.WARNING else Severity.INFO
+            true
+        }
     }
 }

services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/services/EventStreamingService.kt

@@ -73,6 +73,7 @@ internal class EventStreamingService @Inject constructor(
             reconnectOnCancelled = true,
             reconnectOnAborted = true,
             reconnectDelayMs = 1_000L,
+            maxReconnectDelayMs = 30_000L,
             onError = { onError(it) },
             responseHandler = { response, sendRequest ->
                 when (response.typeCase) {

services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt

@@ -13,6 +13,10 @@ import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.launch
 import kotlin.coroutines.cancellation.CancellationException
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TimeMark
+import kotlin.time.TimeSource
 
 fun <Request, Response, StreamRef> openBidirectionalStream(
     streamRef: StreamRef,
@@ -27,7 +31,13 @@ fun <Request, Response, StreamRef> openBidirectionalStream(
     reconnectOnCancelled: Boolean = false,
     reconnectOnAborted: Boolean = false,
     reconnectDelayMs: Long = 300L,
+    maxReconnectDelayMs: Long = 30_000L,
     maxReconnectAttempts: Int = 8,
+    // Minimum time a stream must stay connected (past first response) before we
+    // treat it as healthy and reset the backoff counter. Prevents an
+    // activate-then-immediately-fail stream from resetting forever.
+    healthyThreshold: Duration = 30.seconds,
+    timeSource: TimeSource = TimeSource.Monotonic,
     onReconnectAttempt: ((attempt: Int, reason: Throwable?) -> Unit)? = null
 ) where StreamRef : BidirectionalStreamReference<*, *> {
 
@@ -55,8 +65,9 @@ fun <Request, Response, StreamRef> openBidirectionalStream(
         var attempt = 0
 
         while (attempt++ <= maxReconnectAttempts) {
-            if (attempt > 1) {
-                delay(reconnectDelayMs)
+            val backoffMs = computeBackoffMs(attempt, reconnectDelayMs, maxReconnectDelayMs)
+            if (backoffMs > 0) {
+                delay(backoffMs)
             }
 
             trace(tag = tag, message = "Opening bidirectional stream (attempt $attempt)")
@@ -84,6 +95,7 @@ fun <Request, Response, StreamRef> openBidirectionalStream(
             }
 
             var activated = false
+            var activationMark: TimeMark? = null
 
             val collectionJob = launch {
                 responseFlow
@@ -96,6 +108,7 @@ fun <Request, Response, StreamRef> openBidirectionalStream(
                     .collect { response ->
                         if (!activated) {
                             activated = true
+                            activationMark = timeSource.markNow()
                             streamRef.activateStream()
                             trace(tag = tag, message = "Stream activated on first response")
                         }
@@ -154,14 +167,21 @@ fun <Request, Response, StreamRef> openBidirectionalStream(
 
             val code = error.grpcStatusCode()
             if (isRetryable(error)) {
+                val healthy = activationMark?.elapsedNow()?.let { it >= healthyThreshold } ?: false
                 trace(
                     tag = tag,
-                    message = "Stream failed (${code?.name ?: error.javaClass.simpleName}) → reconnecting"
+                    message = "Stream failed (${code?.name ?: error.javaClass.simpleName}) " +
+                        "→ reconnecting (attempt $attempt, healthy=$healthy)"
                 )
                 onReconnectAttempt?.invoke(attempt, error)
-                // Reset attempt counter on successful activation — only count
-                // consecutive failures, not total lifetime attempts.
-                if (activated) attempt = 0
+                // Reset the backoff counter ONLY when the stream was genuinely
+                // healthy (stayed connected past healthyThreshold). A stream that
+                // activates then immediately fails (e.g. ABORTED "stream already
+                // exists") must count toward maxReconnectAttempts so the loop
+                // terminates instead of hammering the server.
+                // Reset to 0: after a healthy stream dies, reconnect immediately
+                // (computeBackoffMs returns 0 for attempt=1).
+                if (healthy) attempt = 0
                 continue
             }
 
@@ -189,3 +209,14 @@ internal fun Throwable.grpcStatusCode(): Status.Code? = when (this) {
     is StatusException -> status.code
     else -> null
 }
+
+/**
+ * Exponential backoff with a hard cap. Returns 0 for the first attempt (no delay)
+ * and for a zero base delay (used in tests). Doubles per consecutive failed attempt.
+ */
+internal fun computeBackoffMs(attempt: Int, baseDelayMs: Long, maxDelayMs: Long): Long {
+    if (attempt <= 1 || baseDelayMs <= 0) return 0L
+    val exponent = (attempt - 2).coerceIn(0, 16)
+    val scaled = baseDelayMs shl exponent
+    return scaled.coerceAtMost(maxDelayMs)
+}

services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt

@@ -1,15 +1,19 @@
-@file:OptIn(ExperimentalCoroutinesApi::class)
+@file:OptIn(ExperimentalCoroutinesApi::class, kotlin.time.ExperimentalTime::class)
 
 package com.getcode.opencode.internal.bidi
 
+import io.grpc.Status
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.channels.ClosedSendChannelException
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TestTimeSource
 
 class OpenBidirectionalStreamTest {
 
@@ -89,4 +93,116 @@ class OpenBidirectionalStreamTest {
 
         streamRef.destroy()
     }
+
+    @Test
+    fun `computeBackoffMs grows exponentially and caps at max`() {
+        assertEquals(0L, computeBackoffMs(attempt = 1, baseDelayMs = 1000, maxDelayMs = 30_000))
+        assertEquals(1000L, computeBackoffMs(attempt = 2, baseDelayMs = 1000, maxDelayMs = 30_000))
+        assertEquals(2000L, computeBackoffMs(attempt = 3, baseDelayMs = 1000, maxDelayMs = 30_000))
+        assertEquals(4000L, computeBackoffMs(attempt = 4, baseDelayMs = 1000, maxDelayMs = 30_000))
+        assertEquals(8000L, computeBackoffMs(attempt = 5, baseDelayMs = 1000, maxDelayMs = 30_000))
+        assertEquals(30_000L, computeBackoffMs(attempt = 20, baseDelayMs = 1000, maxDelayMs = 30_000))
+        // Zero base (used by tests) must never delay.
+        assertEquals(0L, computeBackoffMs(attempt = 5, baseDelayMs = 0, maxDelayMs = 30_000))
+    }
+
+    /**
+     * Regression for the infinite ABORTED reconnect loop (Bugsnag 6a4528ca).
+     * The stream activates (emits one response) then aborts with ABORTED on every
+     * attempt. Because it never stays healthy past `healthyThreshold`, the attempt
+     * counter must NOT reset, so the loop terminates at maxReconnectAttempts.
+     *
+     * Without the fix (reset on mere activation), this loops forever and the test
+     * times out.
+     */
+    @Test
+    fun `activate-then-abort stream stops at maxReconnectAttempts`() = runTest {
+        var attemptCount = 0
+        val errors = mutableListOf<Throwable>()
+        val timeSource = TestTimeSource() // stays at 0 → never healthy
+
+        val streamRef = BidirectionalStreamReference<String, String>(this, "test-stream")
+        streamRef.retain()
+
+        openBidirectionalStream<String, String, BidirectionalStreamReference<String, String>>(
+            streamRef = streamRef,
+            apiCall = { requestFlow ->
+                attemptCount++
+                flow {
+                    // RENDEZVOUS channel: requestChannel.send() in the coordinator suspends until
+                    // a consumer is ready. Call first() here to rendezvous so send can proceed.
+                    requestFlow.first()
+                    emit("activation") // activates the stream (first response)
+                    throw Status.ABORTED.withDescription("stream already exists").asRuntimeException()
+                }
+            },
+            initialRequest = { "req" },
+            responseHandler = { _: String, _: (String) -> Unit -> },
+            onError = { errors.add(it) },
+            reconnectOnAborted = true,
+            maxReconnectAttempts = 3,
+            reconnectDelayMs = 0,
+            healthyThreshold = 30.seconds,
+            timeSource = timeSource,
+        )
+
+        advanceUntilIdle()
+
+        // maxReconnectAttempts=3 → 4 total attempts, then give up.
+        assertEquals(4, attemptCount, "Should stop after maxReconnectAttempts, got $attemptCount")
+        assertTrue(
+            errors.any { it is IllegalStateException },
+            "Should terminate with max-attempts IllegalStateException"
+        )
+
+        streamRef.destroy()
+    }
+
+    /**
+     * A stream that stays healthy (survives past healthyThreshold) between failures
+     * must reset the backoff counter and keep reconnecting well beyond
+     * maxReconnectAttempts, until a non-retryable error stops it.
+     */
+    @Test
+    fun `healthy stream resets backoff counter across failures`() = runTest {
+        var attemptCount = 0
+        val errors = mutableListOf<Throwable>()
+        val timeSource = TestTimeSource()
+
+        val streamRef = BidirectionalStreamReference<String, String>(this, "test-stream")
+        streamRef.retain()
+
+        openBidirectionalStream<String, String, BidirectionalStreamReference<String, String>>(
+            streamRef = streamRef,
+            apiCall = { requestFlow ->
+                attemptCount++
+                flow {
+                    // RENDEZVOUS channel: requestChannel.send() in the coordinator suspends until
+                    // a consumer is ready. Call first() here to rendezvous so send can proceed.
+                    requestFlow.first()
+                    emit("activation")            // activates + marks activation time
+                    timeSource += 31.seconds      // stream stayed healthy
+                    if (attemptCount >= 5) {
+                        throw IllegalArgumentException("stop") // non-retryable → terminate
+                    }
+                    throw Status.ABORTED.asRuntimeException() // healthy abort → reset counter
+                }
+            },
+            initialRequest = { "req" },
+            responseHandler = { _: String, _: (String) -> Unit -> },
+            onError = { errors.add(it) },
+            reconnectOnAborted = true,
+            maxReconnectAttempts = 2, // would stop at 3 attempts WITHOUT resets
+            reconnectDelayMs = 0,
+            healthyThreshold = 30.seconds,
+            timeSource = timeSource,
+        )
+
+        advanceUntilIdle()
+
+        assertEquals(5, attemptCount, "Healthy resets should let it survive past maxReconnectAttempts")
+        assertTrue(errors.last() is IllegalArgumentException)
+
+        streamRef.destroy()
+    }
 }