From 99a510718714f85a20b9d3b1cb4b2249e9b01eb9 Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Sun, 9 Aug 2026 16:38:12 -0600 Subject: [PATCH 1/4] Reconnect SSE stream after fatal EventSource errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EventSource only auto-reconnects after transient failures. A fatal one — a non-200 response or wrong content-type, e.g. loading the page while the server is mid-restart — puts it in CLOSED permanently, per spec. use-sse.ts didn't handle that: onerror only recorded a metric, and the dead instance stayed in eventSourceRef, so openSSE() early-returned forever. Only a tab hide/show or an authState change cleared the ref, so a visible desktop tab silently lost every realtime update — agent status, terminal-state banner, media/review refreshes, injection-hold badge — while the app looked healthy. onerror now branches on readyState: CONNECTING means the browser is retrying on its own and is left alone; CLOSED means the instance is dead, so it's dropped from the ref and reopened on a capped backoff (1s doubling to 30s). The backoff resets when the stream delivers an event (the server sends a snapshot on every connect) and when the tab is foregrounded. No heartbeat watchdog: the server's keepalives are SSE comments, which EventSource never surfaces to onmessage, so a client-side liveness timer couldn't tell a hung proxy from a genuinely idle stream. Co-Authored-By: Claude Opus 5 (1M context) --- apps/web/src/hooks/use-sse.test.ts | 136 ++++++++++++++++++++++++++++- apps/web/src/hooks/use-sse.ts | 49 +++++++++++ 2 files changed, 183 insertions(+), 2 deletions(-) diff --git a/apps/web/src/hooks/use-sse.test.ts b/apps/web/src/hooks/use-sse.test.ts index 3b7000b7..ae091e09 100644 --- a/apps/web/src/hooks/use-sse.test.ts +++ b/apps/web/src/hooks/use-sse.test.ts @@ -1,6 +1,8 @@ // @vitest-environment jsdom -import { QueryClient } from "@tanstack/react-query"; -import { describe, expect, it, vi } from "vitest"; +import { createElement, type ReactNode } from "react"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { act, renderHook } from "@testing-library/react"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { agentDiffQueryKey } from "@/hooks/use-agent-diff"; import { diffStatsQueryKey } from "@/hooks/use-agent-diff-stats"; @@ -11,6 +13,7 @@ import { applyAgentUpsert, applyDiffStateChanged, applyReviewCreated, + useSSE, } from "./use-sse"; function agent( @@ -81,3 +84,132 @@ describe("review submission SSE state", () => { }); }); }); + +/** + * Minimal EventSource stand-in. Real EventSource only auto-reconnects after + * transient failures, so the tests drive `readyState` explicitly to + * distinguish "browser is retrying" (CONNECTING) from "browser gave up" + * (CLOSED) — the latter is the case the hook has to recover from itself. + */ +class FakeEventSource { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSED = 2; + static instances: FakeEventSource[] = []; + + readyState = FakeEventSource.CONNECTING; + onmessage: ((event: MessageEvent) => void) | null = null; + onerror: ((event: Event) => void) | null = null; + + constructor(readonly url: string) { + FakeEventSource.instances.push(this); + } + + close(): void { + this.readyState = FakeEventSource.CLOSED; + } + + /** Simulate the server delivering an event (the hook's success signal). */ + emit(payload: unknown): void { + this.readyState = FakeEventSource.OPEN; + this.onmessage?.( + new MessageEvent("message", { data: JSON.stringify(payload) }) + ); + } + + /** Simulate a failure; `fatal` mirrors a non-200 / bad-content-type response. */ + fail(fatal: boolean): void { + this.readyState = fatal + ? FakeEventSource.CLOSED + : FakeEventSource.CONNECTING; + this.onerror?.(new Event("error")); + } +} + +describe("useSSE reconnect", () => { + let originalEventSource: typeof globalThis.EventSource; + + function renderSSE() { + const queryClient = new QueryClient(); + return renderHook(() => useSSE("authenticated"), { + wrapper: ({ children }: { children: ReactNode }) => + createElement(QueryClientProvider, { client: queryClient }, children), + }); + } + + beforeEach(() => { + vi.useFakeTimers(); + FakeEventSource.instances = []; + originalEventSource = globalThis.EventSource; + globalThis.EventSource = FakeEventSource as unknown as typeof EventSource; + }); + + afterEach(() => { + globalThis.EventSource = originalEventSource; + vi.useRealTimers(); + }); + + it("reopens the stream after a fatal error the browser will not retry", () => { + renderSSE(); + expect(FakeEventSource.instances).toHaveLength(1); + + act(() => FakeEventSource.instances[0].fail(true)); + // Nothing yet — the retry is scheduled, not immediate. + expect(FakeEventSource.instances).toHaveLength(1); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + }); + + it("leaves transient errors to the browser's own retry", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(false)); + act(() => void vi.advanceTimersByTime(60_000)); + + expect(FakeEventSource.instances).toHaveLength(1); + }); + + it("backs off exponentially while reconnects keep failing", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => FakeEventSource.instances[1].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + // Second delay is 2s, so 1s is not enough. + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + + it("resets the backoff once the stream delivers again", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + // A snapshot arrives — the connection is healthy again. + act(() => + FakeEventSource.instances[1].emit({ type: "snapshot", agents: [] }) + ); + + act(() => FakeEventSource.instances[1].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + + it("stops retrying after unmount", () => { + const { unmount } = renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + unmount(); + act(() => void vi.advanceTimersByTime(60_000)); + + expect(FakeEventSource.instances).toHaveLength(1); + }); +}); diff --git a/apps/web/src/hooks/use-sse.ts b/apps/web/src/hooks/use-sse.ts index 7d558aca..4978efa8 100644 --- a/apps/web/src/hooks/use-sse.ts +++ b/apps/web/src/hooks/use-sse.ts @@ -20,6 +20,10 @@ import { type ReleaseInfoSnapshot, } from "@/hooks/use-cached-release-info"; +/** Backoff bounds for self-driven reconnects after a fatal EventSource error. */ +const INITIAL_RECONNECT_DELAY_MS = 1_000; +const MAX_RECONNECT_DELAY_MS = 30_000; + type UiEvent = | { type: "snapshot"; agents: Agent[] } | { type: "agent.upsert"; agent: Agent } @@ -150,7 +154,20 @@ export function useSSE(authState: AuthState): void { const eventSourceRef = useRef(null); useEffect(() => { + // EventSource only auto-reconnects after *transient* failures. A fatal + // one (non-200 response, wrong content-type — e.g. hitting the server + // mid-restart) moves it to CLOSED permanently, and nothing here noticed: + // the dead instance stayed in `eventSourceRef`, so `openSSE` early- + // returned forever and a visible tab silently lost every realtime update. + // We drive our own capped backoff for that case. + let reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; + let reconnectTimer: ReturnType | null = null; + const handleSSEMessage = (event: MessageEvent) => { + // Any delivered event means the stream is healthy again — the server + // sends a snapshot on every connect, so this doubles as the success + // signal for the backoff. + reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; try { recordSSEEvent(); const payload = JSON.parse(event.data) as UiEvent; @@ -350,8 +367,27 @@ export function useSSE(authState: AuthState): void { } catch {} }; + const cancelReconnect = () => { + if (reconnectTimer !== null) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } + }; + + const scheduleReconnect = () => { + if (reconnectTimer !== null) return; + const delay = reconnectDelayMs; + reconnectDelayMs = Math.min(reconnectDelayMs * 2, MAX_RECONNECT_DELAY_MS); + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + if (document.hidden) return; + openSSE(); + }, delay); + }; + const openSSE = () => { if (eventSourceRef.current) return; + cancelReconnect(); const source = new EventSource("/api/v1/events", { withCredentials: true, }); @@ -359,10 +395,20 @@ export function useSSE(authState: AuthState): void { source.onmessage = handleSSEMessage; source.onerror = () => { recordSSEReconnect(); + // CONNECTING means the browser is retrying on its own — leave it be. + // CLOSED means it has given up; the instance is dead, so drop it and + // retry ourselves. + if (source.readyState !== EventSource.CLOSED) return; + source.close(); + if (eventSourceRef.current === source) { + eventSourceRef.current = null; + } + scheduleReconnect(); }; }; const closeSSE = () => { + cancelReconnect(); if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; @@ -377,6 +423,9 @@ export function useSSE(authState: AuthState): void { if (document.hidden || authState !== "authenticated") { closeSSE(); } else { + // Foregrounding is a fresh start — don't inherit a backed-off delay + // from whatever killed the previous connection. + reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; openSSE(); } }; From da47c6b5684e5567dc439291926f6ab2b50a824b Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Sun, 9 Aug 2026 20:35:07 -0600 Subject: [PATCH 2/4] Gate the SSE backoff reset on a connection that stayed up MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both reviewers independently found the same hole: the reset was keyed on "any delivered event", but the server writes the snapshot unconditionally the moment it accepts the connection (events-routes.ts:42-48), so a delivered event only proves the *connect* succeeded — not that the stream is healthy. That made a no-backoff loop reachable by the exact scenario this PR targets: the browser's own transient retry lands on a half-started server that accepts and sends a snapshot, the stream dies, the next retry hits the 502 window and goes CLOSED — but the snapshot already reset us to the 1s floor. A server restarting in a loop kept every open tab cycling near the floor, so the cap never engaged in the case it exists for, and each cycle carried a full agent-list re-render plus the invalidation burst. The reset now requires the connection to have lasted STABLE_CONNECTION_MS (10s) first. It stays keyed on a delivered event rather than a bare timer, so a hung proxy holding the socket open still can't clear the backoff. Also fixes the test harness: this vitest config sets neither globals nor setupFiles, so RTL's auto-cleanup never registered and every rendered hook stayed mounted for the rest of the file. The new hidden-tab test sees 7 connections instead of 2 without the explicit cleanup() — six leaked hooks answering the same visibilitychange. Tests: flapping server keeps backing off; tab hidden with a retry pending does not reconnect behind the user's back and recovers exactly once on foreground. Co-Authored-By: Claude Opus 5 (1M context) --- apps/web/src/hooks/use-sse.test.ts | 67 ++++++++++++++++++++++++++++-- apps/web/src/hooks/use-sse.ts | 24 +++++++++-- 2 files changed, 84 insertions(+), 7 deletions(-) diff --git a/apps/web/src/hooks/use-sse.test.ts b/apps/web/src/hooks/use-sse.test.ts index ae091e09..9f0852ca 100644 --- a/apps/web/src/hooks/use-sse.test.ts +++ b/apps/web/src/hooks/use-sse.test.ts @@ -1,7 +1,7 @@ // @vitest-environment jsdom import { createElement, type ReactNode } from "react"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; -import { act, renderHook } from "@testing-library/react"; +import { act, cleanup, renderHook } from "@testing-library/react"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { agentDiffQueryKey } from "@/hooks/use-agent-diff"; @@ -128,6 +128,7 @@ class FakeEventSource { describe("useSSE reconnect", () => { let originalEventSource: typeof globalThis.EventSource; + let hiddenValue = false; function renderSSE() { const queryClient = new QueryClient(); @@ -137,14 +138,30 @@ describe("useSSE reconnect", () => { }); } + /** Flip tab visibility the way a browser does — state change then event. */ + function setHidden(value: boolean) { + hiddenValue = value; + document.dispatchEvent(new Event("visibilitychange")); + } + beforeEach(() => { vi.useFakeTimers(); FakeEventSource.instances = []; + hiddenValue = false; + Object.defineProperty(document, "hidden", { + configurable: true, + get: () => hiddenValue, + }); originalEventSource = globalThis.EventSource; globalThis.EventSource = FakeEventSource as unknown as typeof EventSource; }); afterEach(() => { + // This config sets neither `globals: true` nor `setupFiles`, so React + // Testing Library's auto-cleanup never registers — without an explicit + // call, every hook stays mounted (listeners and all) for the rest of the + // file and a leaked instance answers the next test's document events. + cleanup(); globalThis.EventSource = originalEventSource; vi.useRealTimers(); }); @@ -186,23 +203,67 @@ describe("useSSE reconnect", () => { expect(FakeEventSource.instances).toHaveLength(3); }); - it("resets the backoff once the stream delivers again", () => { + it("resets the backoff once a connection has delivered and stayed up", () => { renderSSE(); act(() => FakeEventSource.instances[0].fail(true)); act(() => void vi.advanceTimersByTime(1_000)); expect(FakeEventSource.instances).toHaveLength(2); - // A snapshot arrives — the connection is healthy again. + // The reconnect holds long enough to count as healthy, then delivers. + act(() => void vi.advanceTimersByTime(10_000)); act(() => FakeEventSource.instances[1].emit({ type: "snapshot", agents: [] }) ); + // Backoff is back at the floor, so 1s is enough for the next retry. act(() => FakeEventSource.instances[1].fail(true)); act(() => void vi.advanceTimersByTime(1_000)); expect(FakeEventSource.instances).toHaveLength(3); }); + it("keeps backing off when a flapping server delivers only on connect", () => { + // The server writes a snapshot the instant it accepts the connection, so a + // short-lived connect delivers an event without ever being healthy. That + // must not clear the backoff, or a flapping server pins us at the floor. + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + // Snapshot lands immediately on connect, then the stream dies right away. + act(() => + FakeEventSource.instances[1].emit({ type: "snapshot", agents: [] }) + ); + act(() => FakeEventSource.instances[1].fail(true)); + + // The delay kept doubling: 2s, so 1s must not be enough. + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + + it("recovers when the tab is hidden while a retry is pending", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => setHidden(true)); + + // Hiding cancels the pending retry, so nothing reconnects behind the + // user's back — the timer's own `document.hidden` guard never gets to run. + act(() => void vi.advanceTimersByTime(60_000)); + expect(FakeEventSource.instances).toHaveLength(1); + + // Foregrounding is the escape hatch: reopen at once, and only once. + act(() => setHidden(false)); + expect(FakeEventSource.instances).toHaveLength(2); + act(() => void vi.advanceTimersByTime(60_000)); + expect(FakeEventSource.instances).toHaveLength(2); + }); + it("stops retrying after unmount", () => { const { unmount } = renderSSE(); diff --git a/apps/web/src/hooks/use-sse.ts b/apps/web/src/hooks/use-sse.ts index 4978efa8..3020b17e 100644 --- a/apps/web/src/hooks/use-sse.ts +++ b/apps/web/src/hooks/use-sse.ts @@ -23,6 +23,8 @@ import { /** Backoff bounds for self-driven reconnects after a fatal EventSource error. */ const INITIAL_RECONNECT_DELAY_MS = 1_000; const MAX_RECONNECT_DELAY_MS = 30_000; +/** How long a connection must last before it counts as healthy for the backoff. */ +const STABLE_CONNECTION_MS = 10_000; type UiEvent = | { type: "snapshot"; agents: Agent[] } @@ -162,12 +164,25 @@ export function useSSE(authState: AuthState): void { // We drive our own capped backoff for that case. let reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; let reconnectTimer: ReturnType | null = null; + let connectedAt = 0; const handleSSEMessage = (event: MessageEvent) => { - // Any delivered event means the stream is healthy again — the server - // sends a snapshot on every connect, so this doubles as the success - // signal for the backoff. - reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; + // Clearing the backoff needs proof the stream is actually healthy, and a + // delivered event alone isn't that: the server writes a snapshot the + // moment it accepts the connection, so the first one only proves the + // *connect* succeeded. Resetting on it lets a flapping server pin every + // tab at the 1s floor forever — connect, snapshot, reset, drop, repeat — + // so the cap never engages in exactly the case it exists for. Require + // the connection to have lasted a while first. + // + // Still keyed on a delivered event rather than a bare timer, so a hung + // proxy that holds the socket open without sending anything can't clear + // the backoff. The tradeoff: a connection that stays healthy but silent + // for its whole life fails with an elevated delay (capped at 30s, and + // reset on tab foreground) — acceptable for a stream this chatty. + if (Date.now() - connectedAt >= STABLE_CONNECTION_MS) { + reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; + } try { recordSSEEvent(); const payload = JSON.parse(event.data) as UiEvent; @@ -391,6 +406,7 @@ export function useSSE(authState: AuthState): void { const source = new EventSource("/api/v1/events", { withCredentials: true, }); + connectedAt = Date.now(); eventSourceRef.current = source; source.onmessage = handleSSEMessage; source.onerror = () => { From 912bd6673c72e36b220f8f2f0098f8fd20d18359 Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Sun, 9 Aug 2026 20:51:15 -0600 Subject: [PATCH 3/4] Give the SSE connection and its backoff one owner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Architecture review follow-ups. No behavior change. The backoff reset lived at the top of handleSSEMessage, which is otherwise pure event dispatch — it depends only on queryClient/jotaiStore and is the obvious thing to lift out of this hook later. Putting lifecycle mutation there meant the dispatch concern closed over backoff state and couldn't move without dragging it along. It's now noteStreamActivity(), wrapped at the onmessage assignment site alongside the rest of the backoff mutation. The EventSource lived in a component-lifetime useRef while the three variables describing the same connection were effect-scoped locals — one state machine stored two ways, where establishing that the ref can't outlive the locals took non-local reasoning. The ref is never read during render and never wanted across effect re-runs, so it's now a local like the rest, and the mutual exclusion of "live source" and "pending timer" is stated instead of implied. connectedAt -> connectStartedAt: it's stamped at construction, so it marks when the connect was initiated, not when it was established. The old name made `Date.now() - connectedAt >= STABLE_CONNECTION_MS` read as exact when it's generous by the establishment time. STABLE_CONNECTION_MS's rationale moves to its declaration and cites the server code it depends on, so whoever tunes or deletes the constant reads the why at the same place as the what. Tests adopt the sibling stream test's vi.stubGlobal/unstubAllGlobals, and document.hidden is now a restored spy rather than a permanent redefine. Co-Authored-By: Claude Opus 5 (1M context) --- apps/web/src/hooks/use-sse.test.ts | 12 ++-- apps/web/src/hooks/use-sse.ts | 90 ++++++++++++++++++------------ 2 files changed, 57 insertions(+), 45 deletions(-) diff --git a/apps/web/src/hooks/use-sse.test.ts b/apps/web/src/hooks/use-sse.test.ts index 9f0852ca..6171d0fc 100644 --- a/apps/web/src/hooks/use-sse.test.ts +++ b/apps/web/src/hooks/use-sse.test.ts @@ -127,7 +127,6 @@ class FakeEventSource { } describe("useSSE reconnect", () => { - let originalEventSource: typeof globalThis.EventSource; let hiddenValue = false; function renderSSE() { @@ -148,12 +147,8 @@ describe("useSSE reconnect", () => { vi.useFakeTimers(); FakeEventSource.instances = []; hiddenValue = false; - Object.defineProperty(document, "hidden", { - configurable: true, - get: () => hiddenValue, - }); - originalEventSource = globalThis.EventSource; - globalThis.EventSource = FakeEventSource as unknown as typeof EventSource; + vi.stubGlobal("EventSource", FakeEventSource); + vi.spyOn(document, "hidden", "get").mockImplementation(() => hiddenValue); }); afterEach(() => { @@ -162,7 +157,8 @@ describe("useSSE reconnect", () => { // call, every hook stays mounted (listeners and all) for the rest of the // file and a leaked instance answers the next test's document events. cleanup(); - globalThis.EventSource = originalEventSource; + vi.unstubAllGlobals(); + vi.restoreAllMocks(); vi.useRealTimers(); }); diff --git a/apps/web/src/hooks/use-sse.ts b/apps/web/src/hooks/use-sse.ts index 3020b17e..aa0c921f 100644 --- a/apps/web/src/hooks/use-sse.ts +++ b/apps/web/src/hooks/use-sse.ts @@ -1,4 +1,4 @@ -import { useEffect, useRef } from "react"; +import { useEffect } from "react"; import { type QueryClient, useQueryClient } from "@tanstack/react-query"; import { useStore } from "jotai"; import { @@ -23,7 +23,23 @@ import { /** Backoff bounds for self-driven reconnects after a fatal EventSource error. */ const INITIAL_RECONNECT_DELAY_MS = 1_000; const MAX_RECONNECT_DELAY_MS = 30_000; -/** How long a connection must last before it counts as healthy for the backoff. */ +/** + * How long a connection must last before a delivered event counts as proof + * that it is healthy. + * + * A delivered event on its own proves only that the *connect* succeeded: the + * server writes the snapshot the moment it accepts the connection (see + * `sendUiSnapshot` in apps/server/src/routes/agents/events-routes.ts, called + * immediately after accept). Clearing the backoff on that lets a flapping + * server pin every tab at the 1s floor forever — connect, snapshot, reset, + * drop, repeat — so the cap never engages in the case it exists for. + * + * The reset stays keyed on a delivered event rather than a bare timer, so a + * hung proxy holding the socket open without sending anything can't clear the + * backoff either. Tradeoff: a connection that stays healthy but silent for its + * whole life fails with an elevated delay (capped at MAX_RECONNECT_DELAY_MS, + * and cleared on tab foreground) — acceptable for a stream this chatty. + */ const STABLE_CONNECTION_MS = 10_000; type UiEvent = @@ -153,36 +169,23 @@ export function applyReviewCreated( export function useSSE(authState: AuthState): void { const queryClient = useQueryClient(); const jotaiStore = useStore(); - const eventSourceRef = useRef(null); - useEffect(() => { // EventSource only auto-reconnects after *transient* failures. A fatal // one (non-200 response, wrong content-type — e.g. hitting the server // mid-restart) moves it to CLOSED permanently, and nothing here noticed: - // the dead instance stayed in `eventSourceRef`, so `openSSE` early- - // returned forever and a visible tab silently lost every realtime update. - // We drive our own capped backoff for that case. + // the dead instance stayed put, so `openSSE` early-returned forever and a + // visible tab silently lost every realtime update. We drive our own capped + // backoff for that case. + // + // The connection and its backoff are one state machine, so they share one + // lifetime: all of it is effect-scoped and torn down together. Nothing + // here is read during render, so none of it needs to be a ref. + let source: EventSource | null = null; let reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; let reconnectTimer: ReturnType | null = null; - let connectedAt = 0; + let connectStartedAt = 0; const handleSSEMessage = (event: MessageEvent) => { - // Clearing the backoff needs proof the stream is actually healthy, and a - // delivered event alone isn't that: the server writes a snapshot the - // moment it accepts the connection, so the first one only proves the - // *connect* succeeded. Resetting on it lets a flapping server pin every - // tab at the 1s floor forever — connect, snapshot, reset, drop, repeat — - // so the cap never engages in exactly the case it exists for. Require - // the connection to have lasted a while first. - // - // Still keyed on a delivered event rather than a bare timer, so a hung - // proxy that holds the socket open without sending anything can't clear - // the backoff. The tradeoff: a connection that stays healthy but silent - // for its whole life fails with an elevated delay (capped at 30s, and - // reset on tab foreground) — acceptable for a stream this chatty. - if (Date.now() - connectedAt >= STABLE_CONNECTION_MS) { - reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; - } try { recordSSEEvent(); const payload = JSON.parse(event.data) as UiEvent; @@ -389,6 +392,14 @@ export function useSSE(authState: AuthState): void { } }; + /** A delivered event clears the backoff, but only once the connection has + * proven it can last — see STABLE_CONNECTION_MS. */ + const noteStreamActivity = () => { + if (Date.now() - connectStartedAt >= STABLE_CONNECTION_MS) { + reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; + } + }; + const scheduleReconnect = () => { if (reconnectTimer !== null) return; const delay = reconnectDelayMs; @@ -401,23 +412,28 @@ export function useSSE(authState: AuthState): void { }; const openSSE = () => { - if (eventSourceRef.current) return; + // A live source and a pending retry are mutually exclusive: a retry is + // only ever scheduled after the source is dropped. + if (source) return; cancelReconnect(); - const source = new EventSource("/api/v1/events", { + const opened = new EventSource("/api/v1/events", { withCredentials: true, }); - connectedAt = Date.now(); - eventSourceRef.current = source; - source.onmessage = handleSSEMessage; - source.onerror = () => { + connectStartedAt = Date.now(); + source = opened; + opened.onmessage = (event) => { + noteStreamActivity(); + handleSSEMessage(event); + }; + opened.onerror = () => { recordSSEReconnect(); // CONNECTING means the browser is retrying on its own — leave it be. // CLOSED means it has given up; the instance is dead, so drop it and // retry ourselves. - if (source.readyState !== EventSource.CLOSED) return; - source.close(); - if (eventSourceRef.current === source) { - eventSourceRef.current = null; + if (opened.readyState !== EventSource.CLOSED) return; + opened.close(); + if (source === opened) { + source = null; } scheduleReconnect(); }; @@ -425,9 +441,9 @@ export function useSSE(authState: AuthState): void { const closeSSE = () => { cancelReconnect(); - if (eventSourceRef.current) { - eventSourceRef.current.close(); - eventSourceRef.current = null; + if (source) { + source.close(); + source = null; } }; From 24ced6898e9cce404b4bcf8988c6e881acea201a Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Sun, 9 Aug 2026 21:09:45 -0600 Subject: [PATCH 4/4] Measure connection age, not EventSource instance age MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second architecture review caught a semantics gap in the stability gate. `connectStartedAt` was stamped only in `openSSE`, but the browser reconnects transiently under the *same* instance without re-running it. So a connect-time snapshot delivered by an internal retry passed the 10s check whenever the instance happened to be old — clearing the backoff on exactly the event the gate exists to discount. Reachable trace: elevated backoff -> our reconnect creates an instance -> it survives 10s (possibly silent) -> mixed flapping starts (transient drop, internal retry, snapshot, fatal 502) -> the internal-retry snapshot resets the delay and the fatal that follows starts at 1s instead of the remembered one. Bounded (each cycle costs 10s+ of instance life) but the code no longer meant what STABLE_CONNECTION_MS said. `open` fires on every establishment, including internal retries, so stamping there makes the gate measure true connection age. The stamp in `openSSE` stays as the pre-open fallback: generous by the establishment latency, but never 0 — a zero would make the check trivially true. Renamed `connectionAliveSince` to match what it now holds. Also closes a defensive asymmetry in `onerror`: the `source === opened` identity check guarded the null-out but not `scheduleReconnect()`, so the (unreachable) stale path would have scheduled a retry beside a live connection and doubled the delay for nothing. The whole branch is behind the check now, with the unreachability stated. Co-Authored-By: Claude Opus 5 (1M context) --- apps/web/src/hooks/use-sse.test.ts | 39 ++++++++++++++++++++++++++++++ apps/web/src/hooks/use-sse.ts | 29 +++++++++++++++++----- 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/apps/web/src/hooks/use-sse.test.ts b/apps/web/src/hooks/use-sse.test.ts index 6171d0fc..f6d0c4e1 100644 --- a/apps/web/src/hooks/use-sse.test.ts +++ b/apps/web/src/hooks/use-sse.test.ts @@ -98,6 +98,7 @@ class FakeEventSource { static instances: FakeEventSource[] = []; readyState = FakeEventSource.CONNECTING; + onopen: ((event: Event) => void) | null = null; onmessage: ((event: MessageEvent) => void) | null = null; onerror: ((event: Event) => void) | null = null; @@ -109,6 +110,15 @@ class FakeEventSource { this.readyState = FakeEventSource.CLOSED; } + /** + * Simulate the connection establishing. Fires on the browser's own internal + * retries too, which is what separates connection age from instance age. + */ + open(): void { + this.readyState = FakeEventSource.OPEN; + this.onopen?.(new Event("open")); + } + /** Simulate the server delivering an event (the hook's success signal). */ emit(payload: unknown): void { this.readyState = FakeEventSource.OPEN; @@ -242,6 +252,35 @@ describe("useSSE reconnect", () => { expect(FakeEventSource.instances).toHaveLength(3); }); + it("measures connection age, not instance age, across an internal retry", () => { + // The browser reconnects transiently under the *same* instance without + // re-running openSSE. A snapshot from that fresh connection must not clear + // the backoff just because the instance itself is old. + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + const instance = FakeEventSource.instances[1]; + act(() => instance.open()); + act(() => void vi.advanceTimersByTime(10_000)); + + // Transient drop, then the browser re-establishes on its own and the + // server's connect-time snapshot lands on a seconds-old connection. + act(() => instance.fail(false)); + act(() => instance.open()); + act(() => instance.emit({ type: "snapshot", agents: [] })); + + // Backoff must still be elevated: 2s, so 1s is not enough. + act(() => instance.fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + it("recovers when the tab is hidden while a retry is pending", () => { renderSSE(); diff --git a/apps/web/src/hooks/use-sse.ts b/apps/web/src/hooks/use-sse.ts index aa0c921f..44ff82d1 100644 --- a/apps/web/src/hooks/use-sse.ts +++ b/apps/web/src/hooks/use-sse.ts @@ -183,7 +183,9 @@ export function useSSE(authState: AuthState): void { let source: EventSource | null = null; let reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; let reconnectTimer: ReturnType | null = null; - let connectStartedAt = 0; + /** When the current connection last established — or, before it has + * opened, when we started attempting it. */ + let connectionAliveSince = 0; const handleSSEMessage = (event: MessageEvent) => { try { @@ -395,7 +397,7 @@ export function useSSE(authState: AuthState): void { /** A delivered event clears the backoff, but only once the connection has * proven it can last — see STABLE_CONNECTION_MS. */ const noteStreamActivity = () => { - if (Date.now() - connectStartedAt >= STABLE_CONNECTION_MS) { + if (Date.now() - connectionAliveSince >= STABLE_CONNECTION_MS) { reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; } }; @@ -419,8 +421,20 @@ export function useSSE(authState: AuthState): void { const opened = new EventSource("/api/v1/events", { withCredentials: true, }); - connectStartedAt = Date.now(); + // No `open` yet, so measure from the attempt. Generous by the + // establishment latency, but never 0 — a zero here would make the + // staleness check trivially true and silently clear the backoff. + connectionAliveSince = Date.now(); source = opened; + // `open` fires on every establishment, including the browser's own + // internal retries after a transient drop. Those never re-run `openSSE`, + // so without this the gate would measure the age of the *instance* + // rather than of the connection, and a connect-time snapshot delivered + // by an internal retry would clear the backoff — the exact event the + // gate exists to discount. + opened.onopen = () => { + connectionAliveSince = Date.now(); + }; opened.onmessage = (event) => { noteStreamActivity(); handleSSEMessage(event); @@ -432,9 +446,12 @@ export function useSSE(authState: AuthState): void { // retry ourselves. if (opened.readyState !== EventSource.CLOSED) return; opened.close(); - if (source === opened) { - source = null; - } + // `close()` aborts queued dispatches, so a replaced instance can't + // reach here in a conformant browser. Keep the whole branch behind the + // identity check anyway: scheduling a retry beside a live connection + // would double the delay for nothing. + if (source !== opened) return; + source = null; scheduleReconnect(); }; };