fix: SSE startup + liveness (#189, #186)

app/(tabs)/index.tsx

@@ -186,6 +186,7 @@ export default function SessionsScreen() {
     recentDirectories,
   } = useConnections()
   const authError = useEvents((s) => s.authError)
+  const transport = useEvents((s) => s.transport)
   const reconnect = useEvents((s) => s.connect)
   const loadCatalog = useCatalog((s) => s.load)
   const dirSheetRef = useRef<BottomSheet>(null)
@@ -527,7 +528,13 @@ export default function SessionsScreen() {
         testID="connection-status-bar"
       >
         <View style={styles.connectionInfo}>
-          <View style={[styles.connectionDot, { backgroundColor: "#22c55e" }]} testID="connection-status-dot" />
+          {/* Green only when the SSE transport is verified live (#186):
+              amber while connecting, idle, or reconnecting — the dot used to
+              be hardcoded green regardless of stream health. */}
+          <View
+            style={[styles.connectionDot, { backgroundColor: transport === "live" ? "#22c55e" : "#f59e0b" }]}
+            testID="connection-status-dot"
+          />
           <Text style={[styles.connectionName, isDark && styles.textDark]} numberOfLines={1}>
             {activeConnection.name}
           </Text>

package.json

@@ -28,6 +28,7 @@
     "expo-linking": "~8.0.11",
     "expo-local-authentication": "~17.0.8",
     "expo-localization": "~17.0.9",
+    "expo-network": "~8.0.8",
     "expo-notifications": "~0.32.16",
     "expo-router": "~6.0.22",
     "expo-secure-store": "~15.0.8",

src/lib/sdk-v1.ts

@@ -12,7 +12,7 @@
 // expo/fetch provides WinterCG-compliant fetch with ReadableStream support for SSE
 import { fetch as expoFetch } from "expo/fetch"
 import { buildRequestHeaders } from "./headers"
-import { SSEParser } from "./sse"
+import { SSEParser, SSE_DIAL_TIMEOUT_MS, SSE_LIVENESS_EVENT_TYPE } from "./sse"
 import { apiErrorFor } from "./api-error"
 import { loadSessionList } from "./session-list"
 import type { FileRoot } from "./file-roots"
@@ -183,6 +183,32 @@ export interface HealthResponse {
 
 const REQUEST_TIMEOUT_MS = 30_000
 
+// Minimal shape of a stream reader we need for the idle watchdog (expo/fetch
+// returns its own ReadableStream reader type; structural typing avoids fighting
+// with it). Mirrors src/lib/sdk.ts — the v1/v2 clients are deliberately kept
+// as separate modules, so the wrapper is duplicated rather than shared.
+// The v1 server's heartbeat interval is unknown, but any stream silent for
+// 45s is dead either way: without this the read parks forever on a half-open
+// socket (#186), exactly the failure the v2 wrapper already guards.
+const SSE_IDLE_TIMEOUT_MS = 45_000
+
+type ReaderLike = {
+  read: () => Promise<{ done?: boolean; value?: Uint8Array }>
+  cancel?: () => Promise<void>
+}
+
+async function readWithIdleTimeout(reader: ReaderLike, ms: number): Promise<{ done?: boolean; value?: Uint8Array } | "idle"> {
+  let timer: ReturnType<typeof setTimeout> | undefined
+  // Swallow the read rejection: if the fetch is aborted while this race is
+  // settling, the losing read promise must not become an unhandled rejection.
+  const read = reader.read().catch(() => ({ done: true, value: undefined }))
+  try {
+    return await Promise.race([read, new Promise<"idle">((resolve) => { timer = setTimeout(() => resolve("idle"), ms) })])
+  } finally {
+    clearTimeout(timer)
+  }
+}
+
 // Thrown by request() on a non-2xx response. Carries the HTTP status so
 // callers can distinguish e.g. 404 (older server, endpoint missing) from
 // other failures without parsing the message string.
@@ -275,8 +301,27 @@ export function createV1Client(config: ClientConfig) {
         // Remove Content-Type for SSE (it's text/event-stream)
         delete (headers as Record<string, string>)["Content-Type"]
 
-        // Must use expo/fetch for ReadableStream support on native
-        const response = await expoFetch(url, { headers, signal })
+        // Bound the dial — same shape as the v2 client (see sdk.ts): the
+        // 45s idle watchdog only wraps reader.read() AFTER headers arrive,
+        // so a black-holed dial would park the reconnect loop forever
+        // (Finding 2).
+        const dialController = new AbortController()
+        const dialTimer = setTimeout(() => dialController.abort(), SSE_DIAL_TIMEOUT_MS)
+        const onParentAbort = () => dialController.abort()
+        signal?.addEventListener("abort", onParentAbort)
+        let response: Response
+        try {
+          // Must use expo/fetch for ReadableStream support on native
+          response = await expoFetch(url, { headers, signal: dialController.signal })
+        } catch (err) {
+          if (dialController.signal.aborted && !signal?.aborted) {
+            throw new Error(`SSE dial timed out after ${SSE_DIAL_TIMEOUT_MS / 1000}s — server not responding`)
+          }
+          throw err
+        } finally {
+          clearTimeout(dialTimer)
+          signal?.removeEventListener("abort", onParentAbort)
+        }
         if (!response.ok || !response.body) {
           throw apiErrorFor(response.status, `Failed to connect to event stream: ${response.status}`)
         }
@@ -288,7 +333,13 @@ export function createV1Client(config: ClientConfig) {
         let receivedFirstByte = false
         try {
           while (true) {
-            const { done, value } = await reader.read()
+            const chunk = await readWithIdleTimeout(reader as ReaderLike, SSE_IDLE_TIMEOUT_MS)
+            if (chunk === "idle") {
+              console.warn(`[SSE] no data for ${SSE_IDLE_TIMEOUT_MS / 1000}s — dropping dead stream`)
+              void reader.cancel?.().catch(() => {})
+              throw new Error("Event stream idle — no heartbeat received")
+            }
+            const { done, value } = chunk
             if (done) {
               console.log("[SSE] stream ended")
               break
@@ -299,16 +350,25 @@ export function createV1Client(config: ClientConfig) {
               console.log(`[SSE] first byte received (${value?.byteLength ?? 0} bytes)`)
             }
 
+            // Liveness-only yield for heartbeat-only chunks — same shape as
+            // v2 (see sdk.ts): the v1 subscribe frame is a `: connected`
+            // comment and heartbeats are comments, so a quiet v1 server
+            // would otherwise never yield anything either (Finding 1).
+            let yielded = 0
             for (const data of parser.push(decoder.decode(value, { stream: true }))) {
               try {
                 yield JSON.parse(data)
+                yielded++
               } catch (err) {
                 console.warn("[SSE] Failed to parse event", {
                   length: data.length,
                   error: err instanceof Error ? err.message : String(err),
                 })
               }
             }
+            if (yielded === 0) {
+              yield { type: SSE_LIVENESS_EVENT_TYPE, properties: {} }
+            }
           }
         } finally {
           reader.releaseLock()

src/lib/sdk.ts

@@ -11,7 +11,7 @@
 // unit-testable outside RN.
 import { fetch as expoFetch } from "expo/fetch"
 import { buildRequestHeaders } from "./headers"
-import { SSEParser } from "./sse"
+import { SSEParser, SSE_DIAL_TIMEOUT_MS, SSE_LIVENESS_EVENT_TYPE } from "./sse"
 import { apiErrorFor, isAuthStatus } from "./api-error"
 import { loadSessionList } from "./session-list"
 import type { FileRoot } from "./file-roots"
@@ -362,8 +362,29 @@ function createV2Client(config: ClientConfig) {
         // Remove Content-Type for SSE (it's text/event-stream)
         delete (headers as Record<string, string>)["Content-Type"]
 
-        // Must use expo/fetch for ReadableStream support on native
-        const response = await expoFetch(url, { headers, signal })
+        // Bound the dial: the 45s idle watchdog only wraps reader.read()
+        // AFTER headers arrive, so a black-holed dial would park the
+        // reconnect loop forever — the loop's finally never runs,
+        // attemptInFlight never clears, and reconnectNow() no-ops (#186
+        // Finding 2). A dedicated timeout controller aborts the dial; the
+        // parent signal still aborts it too (connect()/disconnect()).
+        const dialController = new AbortController()
+        const dialTimer = setTimeout(() => dialController.abort(), SSE_DIAL_TIMEOUT_MS)
+        const onParentAbort = () => dialController.abort()
+        signal?.addEventListener("abort", onParentAbort)
+        let response: Response
+        try {
+          // Must use expo/fetch for ReadableStream support on native
+          response = await expoFetch(url, { headers, signal: dialController.signal })
+        } catch (err) {
+          if (dialController.signal.aborted && !signal?.aborted) {
+            throw new Error(`SSE dial timed out after ${SSE_DIAL_TIMEOUT_MS / 1000}s — server not responding`)
+          }
+          throw err
+        } finally {
+          clearTimeout(dialTimer)
+          signal?.removeEventListener("abort", onParentAbort)
+        }
         if (!response.ok || !response.body) {
           throw apiErrorFor(response.status, `Failed to connect to event stream: ${response.status}`)
         }
@@ -393,11 +414,19 @@ function createV2Client(config: ClientConfig) {
               console.log(`[SSE] first byte received (${value?.byteLength ?? 0} bytes)`)
             }
 
+            // Yield a liveness-only event when this chunk carried no session
+            // event: on a quiet server the only traffic is the subscribe
+            // frame (server.connected, dropped by the translator) and
+            // heartbeat comments (never parsed as data). Without this the
+            // events store never sees evidence the stream is live, so a
+            // healthy idle server would read amber forever (Finding 1).
+            let yielded = 0
             for (const data of parser.push(decoder.decode(value, { stream: true }))) {
               try {
                 const raw = JSON.parse(data) as V2RawEvent
                 for (const translated of translator.translate(raw)) {
                   yield translated
+                  yielded++
                 }
               } catch (err) {
                 console.warn("[SSE] Failed to parse event", {
@@ -406,6 +435,9 @@ function createV2Client(config: ClientConfig) {
                 })
               }
             }
+            if (yielded === 0) {
+              yield { type: SSE_LIVENESS_EVENT_TYPE, properties: {} }
+            }
           }
         } finally {
           reader.releaseLock()