diff --git a/app/(tabs)/index.tsx b/app/(tabs)/index.tsx index 7c69237e7..94122b0bc 100644 --- a/app/(tabs)/index.tsx +++ b/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(null) @@ -527,7 +528,13 @@ export default function SessionsScreen() { testID="connection-status-bar" > - + {/* 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. */} + {activeConnection.name} diff --git a/package-lock.json b/package-lock.json index ef4a38bd5..04a57e259 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@opencode-ai/mobile", - "version": "0.4.16", + "version": "0.4.17", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@opencode-ai/mobile", - "version": "0.4.16", + "version": "0.4.17", "dependencies": { "@expo/vector-icons": "^15.0.3", "@gorhom/bottom-sheet": "5.2.8", @@ -23,6 +23,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", @@ -4966,6 +4967,16 @@ "react-native": "*" } }, + "node_modules/expo-network": { + "version": "8.0.8", + "resolved": "https://registry.npmjs.org/expo-network/-/expo-network-8.0.8.tgz", + "integrity": "sha512-dgrL8UHAmWofqeY4UEjWskCl/RoQAM0DG6PZR8xz2WZt+6aQEboQgFRXowCfhbKZ71d16sNuKXtwBEsp2DtdNw==", + "license": "MIT", + "peerDependencies": { + "expo": "*", + "react": "*" + } + }, "node_modules/expo-notifications": { "version": "0.32.17", "resolved": "https://registry.npmjs.org/expo-notifications/-/expo-notifications-0.32.17.tgz", diff --git a/package.json b/package.json index 2b7585591..1c2e694e7 100644 --- a/package.json +++ b/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", diff --git a/src/lib/sdk-v1.ts b/src/lib/sdk-v1.ts index a52dbd489..409e333e9 100644 --- a/src/lib/sdk-v1.ts +++ b/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 +} + +async function readWithIdleTimeout(reader: ReaderLike, ms: number): Promise<{ done?: boolean; value?: Uint8Array } | "idle"> { + let timer: ReturnType | 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)["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,9 +350,15 @@ 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, @@ -309,6 +366,9 @@ export function createV1Client(config: ClientConfig) { }) } } + if (yielded === 0) { + yield { type: SSE_LIVENESS_EVENT_TYPE, properties: {} } + } } } finally { reader.releaseLock() diff --git a/src/lib/sdk.ts b/src/lib/sdk.ts index f2978c321..b00f6f3c2 100644 --- a/src/lib/sdk.ts +++ b/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)["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() diff --git a/src/lib/session-list.integration.test.ts b/src/lib/session-list.integration.test.ts index 285837089..4bea187cf 100644 --- a/src/lib/session-list.integration.test.ts +++ b/src/lib/session-list.integration.test.ts @@ -8,11 +8,35 @@ // every session across every directory — WITHOUT the user picking a folder. // // Run: node --test src/lib/session-list.integration.test.ts +// +// This file also hosts the v2 SSE transport tests for #189/#186 (bottom): +// the events store itself (zustand + react-native AppState) cannot run under +// node --test, so the store's first-event/dedupe logic is covered by +// typecheck + review — these tests prove the /api/event endpoint the store +// reads from actually streams, heartbeats, and broadcasts over real HTTP. import { test, before, after } from "node:test" import assert from "node:assert/strict" +import { register } from "node:module" import { createMockOpencodeServer } from "../../tests/fixtures/mock-opencode-server.ts" import { loadSessionList, legacySessionQuery, type SessionListTransport } from "./session-list.ts" +import { SSEParser, SSE_LIVENESS_EVENT_TYPE } from "./sse.ts" +import type { Client } from "./sdk" + +// Bridge for the generator-level SSE tests below: sdk.ts / sdk-v1.ts import +// `expo/fetch`, which has no resolver outside Metro (its entry requires a +// `.ts` index CJS cannot find), so the SDK cannot be imported under plain +// `node --test` as-is. This resolve hook — registered before the dynamic +// `import("./sdk.ts")` in those tests — redirects `expo/fetch` to a module +// re-exporting the global fetch (WinterCG-compatible, streams +// ReadableStream, exactly what the SSE dial needs) and resolves the SDKs' +// extensionless Metro-style relative imports to their `.ts` files. No +// package.json / test-script change, no new files. `mock.module` cannot do +// this: it needs `--experimental-test-module-mocks`, which the test script +// does not pass. +register( + "data:text/javascript,export async function resolve(specifier, context, nextResolve) { if (specifier === 'expo/fetch') { return { url: 'data:text/javascript,export const fetch = globalThis.fetch;', shortCircuit: true }; } try { return await nextResolve(specifier, context); } catch (e) { if ((specifier.startsWith('./') || specifier.startsWith('../')) && !/\\.[a-z]+$/i.test(specifier)) { return nextResolve(specifier + '.ts', context); } throw e; } }", +) const PORT = 45071 let mock: ReturnType @@ -100,3 +124,194 @@ test("older servers (404 on /experimental/session) fall back to /session and sti // Sanity: the legacy query we sent is the exact one the old code sent. assert.equal(legacySessionQuery({ roots: true, limit: 50 }), "?roots=true&limit=50") }) + +// --- v2 SSE stream (/api/event) — liveness transport coverage (#189/#186) --- +// +// Transport-level only: subscribes over real HTTP, parses frames with the +// production SSEParser, and asserts the two properties the client's liveness +// logic depends on — (1) bytes keep arriving (the 45s idle watchdog in +// src/lib/sdk.ts / sdk-v1.ts treats ANY bytes as proof of life, so a stream +// that never emits bytes is indistinguishable from a dead one), and (2) +// lifecycle broadcasts actually reach subscribers (the events store flips +// `connected`/`transport` to live on the first received event, so a stream +// that never delivers an event would never read as live). +// +// The generator-level tests below (collectGeneratorEvents) go one layer up: +// they drive the REAL client.global.events generator — the exact code path +// the events store consumes — against the mock server, and assert a quiet +// stream yields the liveness-only event. Before the Finding 1 fix the +// generator yielded nothing on a quiet stream (server.connected is dropped +// by the translator, heartbeats are comments), so these tests fail on the +// old code: no yield within the wait, and an empty type list. + +async function collectSse( + url: string, + durationMs: number, + trigger?: () => Promise, +): Promise<{ raw: string; frames: Array> }> { + const controller = new AbortController() + const response = await fetch(url, { headers: { Accept: "text/event-stream" }, signal: controller.signal }) + assert.equal(response.status, 200, `GET ${url} must open the SSE stream`) + assert.match(response.headers.get("content-type") ?? "", /text\/event-stream/) + + const parser = new SSEParser() + const decoder = new TextDecoder() + const frames: Array> = [] + let raw = "" + const reader = response.body!.getReader() + const pump = (async () => { + try { + for (;;) { + const { done, value } = await reader.read() + if (done) break + const text = decoder.decode(value, { stream: true }) + raw += text + for (const data of parser.push(text)) frames.push(JSON.parse(data) as Record) + } + } catch { + // Expected: aborting the fetch below rejects the pending read. + } + })() + if (trigger) await trigger() + await new Promise((resolve) => setTimeout(resolve, durationMs)) + controller.abort() + await pump + return { raw, frames } +} + +test("v2 /api/event: stream opens, server.connected arrives, heartbeat bytes keep arriving", async () => { + const { raw, frames } = await collectSse(`${base}/api/event`, 4500) + + assert.ok( + frames.some((f) => f.type === "server.connected"), + "a server.connected data frame must arrive on subscribe", + ) + // Heartbeat comment frames every 2s — the bytes the idle watchdog watches + // for. If the bug were present (a stream that opens but never emits), this + // fails: no `: ping` in 4.5s of wall time. + assert.match(raw, /: ping/, "heartbeat comment frames must arrive within a few seconds") + assert.ok( + frames.every((f) => typeof f.type === "string" && "data" in f), + "v2 frames are {type, data} shaped (see V2EventTranslator)", + ) +}) + +test("v2 /api/event: execution lifecycle broadcasts reach subscribers in order", async () => { + const created = (await ( + await fetch(`${base}/session`, { method: "POST", headers: { "Content-Type": "application/json" }, body: "{}" }) + ).json()) as { id: string } + + const { frames } = await collectSse(`${base}/api/event`, 5000, async () => { + const res = await fetch(`${base}/session/${created.id}/prompt_async`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ parts: [{ type: "text", text: "hi" }] }), + }) + assert.equal(res.status, 200, "prompt_async must be accepted") + }) + + const types = frames.map((f) => f.type as string) + assert.ok(types.includes("session.execution.started"), "execution start must broadcast over /api/event") + assert.ok(types.includes("session.execution.succeeded"), "execution success must broadcast over /api/event") + assert.ok( + types.indexOf("session.execution.started") < types.indexOf("session.execution.succeeded"), + "started must precede succeeded", + ) + // The lifecycle frames carry the session they belong to. + const started = frames.find((f) => f.type === "session.execution.started") + assert.equal((started?.data as { sessionID?: string } | undefined)?.sessionID, created.id) +}) + +// --- Generator-level liveness (Finding 1) --------------------------------- +// +// Drive the REAL client.global.events generator — the exact code path the +// events store consumes — against the mock server. A quiet stream (subscribe +// frame + heartbeats only, no session events) must yield the liveness-only +// event: that yield is what flips the store to `live`, resets the backoff, +// and triggers the post-reconnect resync. Pre-fix the generator yielded +// nothing here, so the first assertion fails on the old code. + +async function collectGeneratorEvents( + client: Client, + durationMs: number, + trigger?: () => Promise, +): Promise { + const controller = new AbortController() + const gen = client.global.events(controller.signal) + const types: string[] = [] + let notifyFirst: () => void = () => {} + const firstYield = new Promise((resolve) => { + notifyFirst = resolve + }) + const pump = (async () => { + try { + for await (const event of gen) { + types.push(event.type) + if (types.length === 1) notifyFirst() + } + } catch { + // Expected: aborting the fetch below rejects the pending read. + } + })() + // Wait for the subscription to be established (first yield) before + // triggering: a broadcast sent before the dial completes is missed, and + // SSE does not replay. Times out loudly if the generator never yields — + // which is exactly Finding 1 on the old code. + await Promise.race([ + firstYield, + new Promise((_, reject) => + setTimeout(() => reject(new Error("generator never yielded — no liveness on a quiet stream")), 5000), + ), + ]) + if (trigger) await trigger() + await new Promise((resolve) => setTimeout(resolve, durationMs)) + controller.abort() + await Promise.race([pump, new Promise((resolve) => setTimeout(resolve, 2000))]) + return types +} + +test("v2 client.global.events: quiet stream yields liveness without any session event", async () => { + const { createClient } = await import("./sdk.ts") + const client = createClient({ baseUrl: base }) + const types = await collectGeneratorEvents(client, 3000) + + assert.ok(types.length >= 1, "generator must yield on a quiet stream (server.connected + heartbeats only)") + assert.ok( + types.every((t) => t === SSE_LIVENESS_EVENT_TYPE), + `quiet stream must yield only liveness, got: ${types.join(",")}`, + ) +}) + +test("v1 client.global.events: quiet stream yields liveness without any session event", async () => { + const { createClient } = await import("./sdk.ts") + const client = createClient({ baseUrl: base, api: "v1" }) + const types = await collectGeneratorEvents(client, 3000) + + assert.ok(types.length >= 1, "generator must yield on a quiet v1 stream (: connected comment + heartbeats only)") + assert.ok( + types.every((t) => t === SSE_LIVENESS_EVENT_TYPE), + `quiet v1 stream must yield only liveness, got: ${types.join(",")}`, + ) +}) + +test("v2 client.global.events: lifecycle events still flow through the generator", async () => { + const { createClient } = await import("./sdk.ts") + const client = createClient({ baseUrl: base }) + const created = (await ( + await fetch(`${base}/session`, { method: "POST", headers: { "Content-Type": "application/json" }, body: "{}" }) + ).json()) as { id: string } + + const types = await collectGeneratorEvents(client, 5000, async () => { + const res = await fetch(`${base}/session/${created.id}/prompt_async`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ parts: [{ type: "text", text: "hi" }] }), + }) + assert.equal(res.status, 200, "prompt_async must be accepted") + }) + + // execution.started/succeeded translate to session.status busy/idle. + assert.ok(types.includes("session.status"), `lifecycle must reach the generator, got: ${types.join(",")}`) + // And the liveness yield must not have swallowed them — both kinds flow. + assert.ok(types.includes(SSE_LIVENESS_EVENT_TYPE), `liveness must still be yielded, got: ${types.join(",")}`) +}) diff --git a/src/lib/sse.ts b/src/lib/sse.ts index d91fa1c4e..54862e6c6 100644 --- a/src/lib/sse.ts +++ b/src/lib/sse.ts @@ -2,6 +2,36 @@ // Extracted from sdk.ts so the chunk-buffering rules — which are easy to get wrong when // a single event is split across two network reads — are unit-testable without a socket. +/** + * Type of the liveness-only event the SDK event generators yield when a chunk + * arrives carrying no session event (subscribe frame, heartbeat comment). + * A healthy but idle server never emits a session event, so without this the + * events store could never observe that the stream is live (#189/#186 + * Finding 1). Carries no payload: the store treats it as proof of bytes on + * the wire and skips dispatch. Fits the shared `Event {type, properties}` + * shape, so both the v1 and v2 generators can yield it. + */ +export const SSE_LIVENESS_EVENT_TYPE = "sse.liveness" + +/** + * Bound on the SSE dial (headers must arrive within this long). The 45s idle + * watchdog only wraps `reader.read()` AFTER headers arrive, so a black-holed + * dial would otherwise park the reconnect loop forever — `attemptInFlight` + * never clears and `reconnectNow()` no-ops (Finding 2). The events store's + * stale-connecting rescue keys off this same constant, so the two bounds + * cannot drift apart. + */ +export const SSE_DIAL_TIMEOUT_MS = 15_000 + +/** + * A `live` stream with no byte for this long is stale — a half-open socket + * after a network change, where the 45s idle watchdog has not fired yet. + * The events store lets `reconnectNow()` rescue a stale-live stream at this + * bound instead of waiting out the watchdog. Two missed v2 heartbeats + * (15s interval); well under the 45s watchdog. + */ +export const SSE_STALE_LIVE_MS = 30_000 + export class SSEParser { private buffer = "" diff --git a/src/stores/connections.ts b/src/stores/connections.ts index da109fe9a..c1df528ae 100644 --- a/src/stores/connections.ts +++ b/src/stores/connections.ts @@ -96,6 +96,28 @@ async function resolveApi( } } +// Fetch /project/current + /path AFTER the client is published to the store, +// so the SSE stream (which keys off `client` appearing) is not serialized +// behind two metadata requests it doesn't need (#189). `isCurrent` guards +// against a stale response clobbering state after a server switch mid-fetch: +// the metadata only lands if this client is still the active one. +async function fetchMetadataBehind( + client: Client, + isCurrent: () => boolean, + apply: (project: Project | null, home: string | null) => void, +): Promise { + try { + const [proj, paths] = await Promise.all([ + client.project.current().catch(() => null), + client.path.get().catch(() => null), + ]) + if (!isCurrent()) return + apply(proj, paths?.home ?? null) + } catch { + // Server might be offline — metadata stays null, the client still works. + } +} + export const useConnections = create((set, get) => ({ connections: [], activeConnection: null, @@ -123,11 +145,11 @@ export const useConnections = create((set, get) => ({ // Create client for active connection let client: Client | null = null let base: ClientBase | null = null - let project: Project | null = null - let home: string | null = null if (active) { const password = await SecureStore.getItemAsync(`${PASSWORDS_PREFIX}${active.id}`) const auth = buildAuth(active.username, password) + // The resolveApi() probe stays BEFORE the publish: it chooses which + // client to build, so #189's split starts after the client exists. const { api, resolved } = await resolveApi(active, auth) const built = buildClient(active.url, active.directory, auth, api) client = built.client @@ -135,29 +157,29 @@ export const useConnections = create((set, get) => ({ // Persist a probe that was only just discovered, so the next launch // picks the same transport without probing again. if (resolved) await SecureStore.setItemAsync(CONNECTIONS_KEY, JSON.stringify(connections)) - // Fetch current project info and server paths - try { - const [proj, paths] = await Promise.all([ - client.project.current().catch(() => null), - client.path.get().catch(() => null), - ]) - project = proj - home = paths?.home || null - } catch { - // Server might be offline - } } + // Publish the client immediately — the SSE stream starts off this, and + // must not wait on the metadata fetches below (#189). set({ connections, activeConnection: active, client, clientBase: base, - currentProject: project, - serverHome: home, + currentProject: null, + serverHome: null, recentDirectories, isLoading: false, }) + + // Metadata fills in behind the publish, guarded so a stale response + // can't clobber state after a server switch mid-fetch. + if (active && client) { + const requestClient = client + void fetchMetadataBehind(requestClient, () => get().client === requestClient, (proj, home) => + set({ currentProject: proj, serverHome: home }), + ) + } } catch (error) { set({ error: "Failed to load connections", isLoading: false }) } @@ -185,9 +207,6 @@ export const useConnections = create((set, get) => ({ let base = get().clientBase let activeConnection = get().activeConnection - let project = get().currentProject - let serverHome = get().serverHome - if (newConnection.active) { activeConnection = newConnection const auth = buildAuth(newConnection.username, password) @@ -196,21 +215,23 @@ export const useConnections = create((set, get) => ({ client = built.client base = built.base - // Fetch server metadata so loadSessions can use clientForDirectory(serverHome) - // immediately after the connection is added (same as setActiveConnection does). - try { - const [proj, paths] = await Promise.all([ - client.project.current().catch(() => null), - client.path.get().catch(() => null), - ]) - project = proj - serverHome = paths?.home || null - } catch { - // Server might be unreachable; proceed without metadata - } + // Publish the client first so SSE can start; metadata fills in behind + // it (#189), guarded against a server switch mid-fetch. + set({ connections, activeConnection, client, clientBase: base, currentProject: null, serverHome: null }) + void fetchMetadataBehind(built.client, () => get().client === built.client, (proj, home) => + set({ currentProject: proj, serverHome: home }), + ) + return } - set({ connections, activeConnection, client, clientBase: base, currentProject: project, serverHome }) + set({ + connections, + activeConnection, + client, + clientBase: base, + currentProject: get().currentProject, + serverHome: get().serverHome, + }) }, removeConnection: async (id) => { @@ -253,8 +274,6 @@ export const useConnections = create((set, get) => ({ const active = connections.find((c) => c.id === id) || null let client: Client | null = null let base: ClientBase | null = null - let project: Project | null = null - let home: string | null = null if (active) { const password = await SecureStore.getItemAsync(`${PASSWORDS_PREFIX}${active.id}`) @@ -264,28 +283,29 @@ export const useConnections = create((set, get) => ({ client = built.client base = built.base - try { - const [proj, paths] = await Promise.all([ - client.project.current().catch(() => null), - client.path.get().catch(() => null), - ]) - project = proj - home = paths?.home || null - } catch { - // Server might be offline - } - // Update last connected time active.lastConnected = Date.now() await SecureStore.setItemAsync(CONNECTIONS_KEY, JSON.stringify(connections)) } - set({ connections, activeConnection: active, client, clientBase: base, currentProject: project, serverHome: home }) + // Publish the client immediately so SSE can start; metadata fills in + // behind it (#189), guarded against a server switch mid-fetch. + set({ connections, activeConnection: active, client, clientBase: base, currentProject: null, serverHome: null }) + // Breadcrumb at publish time, not after the metadata fetch: the old + // placement fired only if both metadata fetches resolved and the stale + // guard passed, so a slow/offline server lost the breadcrumb entirely + // (LOW 6). hasProject is unknown at publish — report false. addBreadcrumb({ category: "connection", message: active ? `active connection set: ${active.type}` : "active connection cleared", - data: { id: active?.id, type: active?.type, hasProject: Boolean(project) }, + data: { id: active?.id, type: active?.type, hasProject: false }, }) + if (active && client) { + const requestClient = client + void fetchMetadataBehind(requestClient, () => get().client === requestClient, (proj, home) => { + set({ currentProject: proj, serverHome: home }) + }) + } }, testConnection: async (connection, source, password) => { @@ -345,28 +365,19 @@ export const useConnections = create((set, get) => ({ const { api, resolved } = await resolveApi(active, auth) const built = buildClient(active.url, active.directory, auth, api) if (resolved) await SecureStore.setItemAsync(CONNECTIONS_KEY, JSON.stringify(connections)) - try { - const [project, paths] = await Promise.all([ - built.client.project.current().catch(() => null), - built.client.path.get().catch(() => null), - ]) - set({ - connections, - activeConnection: active, - client: built.client, - clientBase: built.base, - currentProject: project, - serverHome: paths?.home || null, - }) - } catch { - set({ - connections, - activeConnection: active, - client: built.client, - clientBase: built.base, - currentProject: null, - }) - } + // Publish the rebuilt client immediately; metadata fills in behind it + // (#189), guarded against a server switch mid-fetch. + set({ + connections, + activeConnection: active, + client: built.client, + clientBase: built.base, + currentProject: null, + serverHome: null, + }) + void fetchMetadataBehind(built.client, () => get().client === built.client, (proj, home) => + set({ currentProject: proj, serverHome: home }), + ) } else { set({ connections }) } diff --git a/src/stores/events.ts b/src/stores/events.ts index a6a1c3577..301078fe1 100644 --- a/src/stores/events.ts +++ b/src/stores/events.ts @@ -10,14 +10,29 @@ import { recordSuccessfulSession } from "../lib/store-review" import { isAuthError } from "../lib/api-error" import { isSessionActuallyIdle } from "../lib/session-status-reconcile" import { AppState } from "react-native" +import * as Network from "expo-network" import { mergePendingPermissions, normalizePendingPermission, normalizePermissionEvent, type PendingPermission } from "../lib/permissions" +import { SSE_DIAL_TIMEOUT_MS, SSE_LIVENESS_EVENT_TYPE, SSE_STALE_LIVE_MS } from "../lib/sse" import type { Client, Part, Session, Message } from "../lib/sdk" // Session status from the server type SessionStatus = { type: "idle" } | { type: "busy" } | { type: "retry"; attempt: number; message: string } +// Transport liveness, distinct from "a connection is selected" (#186). +// `connected` flips only when a byte actually arrives on the stream; the +// finer distinction lives here so the UI can't show green over a stream that +// never delivered anything. +export type TransportState = + /** No attempt in flight and nothing live. */ + | "idle" + /** Attempting; nothing received yet. NOT the same as connected. */ + | "connecting" + /** At least one event received on this attempt. */ + | "live" + interface EventsState { connected: boolean + transport: TransportState // Set when the last connection attempt failed with 401/403 — the server // rejected our credentials, not a transient network issue. The reconnect // loop stops retrying in this case (see connect()) since hammering a @@ -58,6 +73,21 @@ interface EventsState { let controller: AbortController | null = null let reconnectTimer: ReturnType | null = null +// True while a connect() attempt's stream loop is running. reconnectNow() +// (the single resume entry point) no-ops while this is set, so a foreground +// event racing a network-restore event cannot open a second stream. Guarded +// by attemptToken: connect() bumps the token, and a stale loop's finally only +// clears the flag when its token is still current — otherwise aborting an old +// attempt would clear the flag out from under the new one. +let attemptInFlight = false +let attemptToken = 0 +// Wall-clock time a byte last arrived on the current attempt (any yielded +// event — liveness-only included). Lets reconnectNow() tell a stale-live +// half-open stream from a fresh one (Finding 3). +let lastByteAt: number | null = null +// When the current attempt started dialling. Lets reconnectNow() rescue an +// attempt parked in a hung dial beyond the dial bound (Finding 2). +let attemptStartedAt: number | null = null // Sessions that emitted session.error since they last went busy. SessionStatus // has no error variant — an errored session still ends with a busy -> idle @@ -66,7 +96,6 @@ let reconnectTimer: ReturnType | null = null const erroredSessions = new Set() const RECONNECT_DELAYS_MS = [1000, 2000, 4000, 8000, 15000] as const -const STABLE_CONNECTION_MS = 10_000 const PROLONGED_DISCONNECT_MS = 30_000 // Re-fetch pending permissions and questions from the server for a session. @@ -174,6 +203,7 @@ async function resyncBusySessions() { export const useEvents = create((set, get) => ({ connected: false, + transport: "idle", authError: false, reconnectAttempts: 0, lastDisconnectAt: null, @@ -195,7 +225,15 @@ export const useEvents = create((set, get) => ({ controller = new AbortController() const currentController = controller - set({ connected: true, authError: false }) + const token = ++attemptToken + attemptInFlight = true + attemptStartedAt = Date.now() + lastByteAt = null + // `connected` is NOT set here: it flips only on the first received + // evidence below, so "connecting" never reads as live (#186). It IS + // cleared here, so a re-entry (manual Retry) cannot leave a stale + // `connected: true` behind while the transport is "connecting" (LOW 5). + set({ connected: false, transport: "connecting", authError: false }) console.log("[SSE] Connecting to event stream...") addBreadcrumb({ category: "sse", message: "connecting" }) @@ -208,11 +246,14 @@ export const useEvents = create((set, get) => ({ // failed retries can't re-arm the check on every attempt. const isReconnect = get().reconnectAttempts > 0 let resyncedAfterReconnect = false - const stableTimer = setTimeout(() => { - if (!currentController.signal.aborted) { - set({ reconnectAttempts: 0, lastDisconnectAt: null }) - } - }, STABLE_CONNECTION_MS) + // True once this attempt has demonstrated liveness. The backoff reset + // below keys off this, not a timer: a silently-failing connection used + // to clear its own backoff after 10s whether or not anything arrived. + // The SDK yields an `sse.liveness` event for heartbeat-only chunks, + // so a healthy but idle server demonstrates liveness without any + // session event (Finding 1); the idle watchdog already treats any + // bytes on the wire as liveness at the transport layer. + let sawLiveness = false const scheduleReconnect = (reason: unknown) => { if (reconnectScheduled || currentController.signal.aborted) return @@ -221,7 +262,7 @@ export const useEvents = create((set, get) => ({ const reconnectAttempts = state.reconnectAttempts + 1 const lastDisconnectAt = state.lastDisconnectAt ?? Date.now() const disconnectedFor = Date.now() - lastDisconnectAt - set({ connected: false, reconnectAttempts, lastDisconnectAt }) + set({ connected: false, transport: "connecting", reconnectAttempts, lastDisconnectAt }) if (disconnectedFor >= PROLONGED_DISCONNECT_MS) { notify({ @@ -253,17 +294,35 @@ export const useEvents = create((set, get) => ({ for await (const event of client.global.events(currentController.signal)) { if (currentController.signal.aborted) break + const payload = (event as any).payload || event + const type = payload.type as string + // Any yielded event — liveness-only or session — is proof of + // bytes on the wire. Record it first: reconnectNow() keys its + // stale-live rescue off this timestamp. + lastByteAt = Date.now() + + // First byte of evidence: the stream is genuinely live, not just + // dialled. This flips `connected`, marks the transport live, and + // resets the backoff — the only reset path (no timer). + if (!sawLiveness) { + sawLiveness = true + set({ connected: true, transport: "live", reconnectAttempts: 0, lastDisconnectAt: null }) + } + // The stream is genuinely live again (we're actually receiving // data, not just optimistically marked "connected") — resync once - // per reconnect, not on every event. + // per reconnect, not on every event. Runs on quiet reconnects + // too: liveness events reach this point, so no session event is + // needed to trigger the resync (Finding 1). if (isReconnect && !resyncedAfterReconnect) { resyncedAfterReconnect = true void resyncBusySessions() void refreshAllPending() } - const payload = (event as any).payload || event - const type = payload.type as string + // Liveness-only: no session payload to dispatch. + if (type === SSE_LIVENESS_EVENT_TYPE) continue + const props = payload.properties || {} switch (type) { @@ -496,12 +555,12 @@ export const useEvents = create((set, get) => ({ data: { status: err.status }, }) track(AnalyticsEvent.ConnectionFailed, { source: "sse", error_class: "unauthorized" }) - set({ connected: false, authError: true }) + set({ connected: false, transport: "idle", authError: true }) } else { scheduleReconnect(err) } } finally { - clearTimeout(stableTimer) + if (token === attemptToken) attemptInFlight = false if (currentController.signal.aborted) { console.log("[SSE] Disconnected (aborted)") } @@ -512,7 +571,27 @@ export const useEvents = create((set, get) => ({ reconnectNow: () => { const connState = useConnections.getState() if (!connState.client) return - console.log("[SSE] Foreground reconnect — re-establishing event stream") + // Deduplicated resume (#186): a foreground event and a network-restore + // event often fire together, and neither may tear down a live stream or + // race an attempt already dialling — both would open a second stream and + // double-handle every event. Two rescues pierce the dedupe: + // - a `live` stream gone stale (no byte for SSE_STALE_LIVE_MS) is a + // half-open socket network-restore must rescue (Finding 3); + // - a `connecting` attempt parked past the dial bound with no byte is a + // hung dial; connect() below aborts the old controller, unwinding it + // (Finding 2). A stalled "connecting" with no attempt in flight still + // resumes (attemptInFlight is false once the loop has exited, e.g. + // while a reconnect is waiting out its backoff). + const now = Date.now() + if (get().transport === "live") { + // `live` is only set after a byte arrives, so lastByteAt is non-null + // here; a null is treated as fresh (no rescue) — the safe direction. + if (lastByteAt === null || now - lastByteAt < SSE_STALE_LIVE_MS) return + } else if (attemptInFlight) { + const parkedFor = attemptStartedAt === null ? 0 : now - attemptStartedAt + if (lastByteAt !== null || parkedFor <= SSE_DIAL_TIMEOUT_MS) return + } + console.log("[SSE] Resume reconnect — re-establishing event stream") if (reconnectTimer) { clearTimeout(reconnectTimer) reconnectTimer = null @@ -546,10 +625,17 @@ export const useEvents = create((set, get) => ({ } controller?.abort() controller = null + // Invalidate any winding-down loop so its finally can't clear state that + // belongs to a later attempt. + attemptToken++ + attemptInFlight = false + lastByteAt = null + attemptStartedAt = null erroredSessions.clear() abortedSessions.clear() set({ connected: false, + transport: "idle", authError: false, reconnectAttempts: 0, lastDisconnectAt: null, @@ -563,4 +649,19 @@ export const useEvents = create((set, get) => ({ AppState.addEventListener("change", (state) => { if (state === "active") void refreshAllPending() -}) \ No newline at end of file +}) + +// Resume the SSE stream when the network comes back (#186). expo-network +// reports link state, not internet reachability: `isConnected` covers +// LAN/Tailscale servers that `isInternetReachable` would call offline, so +// only `isConnected` is consulted. Only the offline -> online transition +// resumes; reconnectNow() itself is deduped, so a simultaneous foreground +// event cannot open a second stream. expo-network is an Expo SDK module, so +// no @react-native-community/netinfo dependency is needed. +let wasNetworkConnected: boolean | null = null +Network.addNetworkStateListener((state) => { + const isConnected = state.isConnected ?? false + const regained = wasNetworkConnected === false && isConnected + wasNetworkConnected = isConnected + if (regained) useEvents.getState().reconnectNow() +}) diff --git a/tests/fixtures/mock-opencode-server.ts b/tests/fixtures/mock-opencode-server.ts index 29ab7f22c..7e1d14ece 100644 --- a/tests/fixtures/mock-opencode-server.ts +++ b/tests/fixtures/mock-opencode-server.ts @@ -16,6 +16,14 @@ // returns immediately; the actual reply is delivered as // `message.updated` + `message.part.updated` + `session.status` (idle) // events on the SSE stream (src/stores/events.ts). +// - GET /api/event is the v2 SSE stream: same framing as /global/event +// (heartbeat comment frames every 2s, `data:` frames) but v2-shaped raw +// events (`{type, data}` — see src/lib/v2-shape.ts V2EventTranslator): +// a `server.connected` frame on subscribe, plus +// `session.execution.started`/`succeeded` lifecycle frames when a prompt +// runs. Transport-level coverage for the SSE liveness work (#189/#186), +// not a full v2 content stream — message content still flows over the +// v1 endpoint only. // - The real server ALSO persists and broadcasts the USER's message. The // app relies on this: any `message.updated` event strips optimistic // `temp-` messages (src/stores/sessions.ts handleEvent), so if the mock @@ -218,6 +226,7 @@ export function createMockOpencodeServer(opts: MockServerOptions) { const sessions = new Map() const messagesBySession = new Map() const sseClients = new Set() + const sseV2Clients = new Set() if (seedDiff) { const now = Date.now() @@ -320,6 +329,56 @@ export function createMockOpencodeServer(opts: MockServerOptions) { } } + // v2-shaped broadcast ({type, data}) for /api/event subscribers. + function broadcastV2(type: string, data: Record) { + const line = `data: ${JSON.stringify({ type, data })}\n\n` + console.log(`[mock-opencode-server] broadcast v2 type=${type} clients=${sseV2Clients.size}`) + for (const res of sseV2Clients) { + try { + res.write(line) + } catch { + sseV2Clients.delete(res) + } + } + } + + // Shared SSE subscribe plumbing for the v1 (/global/event) and v2 + // (/api/event) streams: same headers, same 2s heartbeat comment frames, + // different first frame and client set. + function openSse( + req: http.IncomingMessage, + res: http.ServerResponse, + clients: Set, + firstFrame: string, + label: string, + ) { + res.writeHead(200, { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }) + res.write(firstFrame) + clients.add(res) + console.log(`[mock-opencode-server] SSE connect (${label}), clients=${clients.size}`) + + // Periodic heartbeat comment. If the client never sees even this, the + // silence is a transport problem (expo/fetch not streaming), not a + // broadcast bug — attributes issue #90 mode B. + const heartbeat = setInterval(() => { + try { + res.write(": ping\n\n") + } catch { + clearInterval(heartbeat) + } + }, 2000) + + req.on("close", () => { + clearInterval(heartbeat) + clients.delete(res) + console.log(`[mock-opencode-server] SSE disconnect (${label}), clients=${clients.size}`) + }) + } + function json(res: http.ServerResponse, status: number, body: unknown) { const data = JSON.stringify(body) res.writeHead(status, { @@ -375,6 +434,7 @@ export function createMockOpencodeServer(opts: MockServerOptions) { setTimeout(() => { broadcast("session.status", { sessionID, status: { type: "busy" } }) + broadcastV2("session.execution.started", { sessionID }) const now = Date.now() const messageID = randomUUID() @@ -400,6 +460,7 @@ export function createMockOpencodeServer(opts: MockServerOptions) { list.push({ info: { ...info, time: { created: now, completed: Date.now() } }, parts: [part] }) broadcast("session.status", { sessionID, status: { type: "idle" } }) + broadcastV2("session.execution.succeeded", { sessionID }) }, replyDelayMs) } @@ -496,31 +557,15 @@ export function createMockOpencodeServer(opts: MockServerOptions) { // SSE event stream — kept open for the lifetime of the connection. if (method === "GET" && path === "/global/event") { - res.writeHead(200, { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", - Connection: "keep-alive", - }) - res.write(": connected\n\n") - sseClients.add(res) - console.log(`[mock-opencode-server] SSE connect, clients=${sseClients.size}`) - - // Periodic heartbeat comment. If the client never sees even this, the - // silence is a transport problem (expo/fetch not streaming), not a - // broadcast bug — attributes issue #90 mode B. - const heartbeat = setInterval(() => { - try { - res.write(": ping\n\n") - } catch { - clearInterval(heartbeat) - } - }, 2000) + openSse(req, res, sseClients, ": connected\n\n", "v1") + return + } - req.on("close", () => { - clearInterval(heartbeat) - sseClients.delete(res) - console.log(`[mock-opencode-server] SSE disconnect, clients=${sseClients.size}`) - }) + // v2 SSE event stream (src/lib/sdk.ts createV2Client global.events). + // Same transport, v2-shaped frames: server.connected on subscribe, then + // session.execution.* lifecycle frames (see broadcastV2). + if (method === "GET" && path === "/api/event") { + openSse(req, res, sseV2Clients, `data: ${JSON.stringify({ type: "server.connected", data: {} })}\n\n`, "v2") return } @@ -645,7 +690,7 @@ export function createMockOpencodeServer(opts: MockServerOptions) { return new Promise((resolve) => server.listen(port, "0.0.0.0", () => resolve())) }, close(): Promise { - for (const res of sseClients) { + for (const res of [...sseClients, ...sseV2Clients]) { try { res.end() } catch { @@ -653,6 +698,7 @@ export function createMockOpencodeServer(opts: MockServerOptions) { } } sseClients.clear() + sseV2Clients.clear() return new Promise((resolve, reject) => { server.close((err) => (err ? reject(err) : resolve())) })