diff --git a/tests/e2e_ui/chat/test_stream_tab_limit_banner.py b/tests/e2e_ui/chat/test_stream_tab_limit_banner.py new file mode 100644 index 0000000000..c402be3854 --- /dev/null +++ b/tests/e2e_ui/chat/test_stream_tab_limit_banner.py @@ -0,0 +1,76 @@ +"""E2E: the too-many-tabs banner appears when open streams fill the HTTP pool. + +Each tab with a conversation open holds one long-lived +``GET /v1/sessions/{id}/stream`` SSE request. Browsers cap HTTP/1.1 +connections at ~6 per origin and share that budget across every tab in the +profile, so once ~6 conversations are open the held streams occupy every slot +and unrelated requests queue behind them — the app appears hung with nothing +explaining why. The banner makes that cause visible. + +Why several pages in ONE browser context: Web Locks (which back the count) are +scoped to a browsing-context group, and pages in one Playwright context share a +lock manager — the same scope as real tabs in one browser profile. Separate +contexts are isolated from each other and would each count only themselves. + +The same conversation opened N times is deliberate and realistic: every tab runs +its own store and stream pump, so N tabs on one session hold N streams and +consume N connections, exactly as N tabs on N sessions would. + +A failure here means one of: + +- The lock registry stopped counting held streams + (``web/src/lib/streamTabRegistry.ts``), or ``startStreamPump`` stopped + acquiring/releasing a slot for the stream's lifetime. +- The banner's threshold or render logic regressed + (``web/src/components/StreamTabLimitBanner.tsx``), or it fell out of the + standalone root in ``web/src/main.tsx``. +- The HTTP/1.1 gate started suppressing the banner on the local dev server + (which serves HTTP/1.1, so the cap genuinely applies). +""" + +from __future__ import annotations + +from playwright.sync_api import Browser, expect + +_COMPOSER = "Ask the agent anything…" +# Mirrors WARN_AT_TABS in web/src/components/StreamTabLimitBanner.tsx: the +# banner fires while the app still works, one slot before the pool is full. +_WARN_AT_TABS = 5 + + +def test_banner_warns_once_open_tabs_threaten_the_connection_pool( + browser: Browser, + seeded_session: tuple[str, str], +) -> None: + """Opening enough conversation tabs surfaces the warning; closing clears it. + + :param browser: Playwright session-scoped browser. One context stands in + for one browser profile, whose tabs share both the connection pool and + the Web Locks scope. + :param seeded_session: ``(base_url, session_id)`` from the fixture. + """ + base_url, session_id = seeded_session + context = browser.new_context() + try: + pages = [] + for _ in range(_WARN_AT_TABS): + page = context.new_page() + page.goto(f"{base_url}/c/{session_id}") + # Wait for the composer before opening the next tab: the stream (and + # so the lock) is only held once the conversation has actually bound. + expect(page.get_by_placeholder(_COMPOSER)).to_be_visible(timeout=30_000) + pages.append(page) + + # Assert on the last tab: its own slot acquisition refreshes the count + # immediately, so it doesn't wait on the peer-tab poll interval. + banner = pages[-1].get_by_role("status").filter(has_text="conversation open") + expect(banner).to_be_visible(timeout=30_000) + expect(banner).to_contain_text(f"{_WARN_AT_TABS} tabs have a conversation open") + + # Closing a tab releases its stream — and its connection — so the + # warning must retire itself rather than persist after the user has + # already acted on it. + pages[0].close() + expect(banner).not_to_be_visible(timeout=30_000) + finally: + context.close() diff --git a/web/src/components/StreamTabLimitBanner.test.tsx b/web/src/components/StreamTabLimitBanner.test.tsx new file mode 100644 index 0000000000..ebfdb76791 --- /dev/null +++ b/web/src/components/StreamTabLimitBanner.test.tsx @@ -0,0 +1,78 @@ +import { fireEvent, render, screen } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +const { mockUseStreamTabCount, mockLowLimit } = vi.hoisted(() => ({ + mockUseStreamTabCount: vi.fn(), + mockLowLimit: vi.fn(), +})); + +vi.mock("@/hooks/useStreamTabCount", () => ({ + useStreamTabCount: mockUseStreamTabCount, +})); +vi.mock("@/lib/streamTabRegistry", () => ({ + connectionHasLowStreamLimit: mockLowLimit, +})); + +import { StreamTabLimitBanner } from "./StreamTabLimitBanner"; + +/** Default to the case where the connection cap actually binds (HTTP/1.1). */ +function setup(tabCount: number, lowLimit = true): void { + mockUseStreamTabCount.mockReturnValue(tabCount); + mockLowLimit.mockReturnValue(lowLimit); +} + +afterEach(() => { + vi.clearAllMocks(); +}); + +describe("StreamTabLimitBanner", () => { + it("stays hidden while tab count is below the warning threshold", () => { + setup(4); + const { container } = render(); + expect(container).toBeEmptyDOMElement(); + }); + + it("warns once enough tabs hold a stream to threaten the connection pool", () => { + setup(5); + render(); + expect(screen.getByRole("status")).toHaveTextContent("5 tabs have a conversation open"); + }); + + it("stays hidden on multiplexed connections where the cap does not bind", () => { + // HTTP/2 / HTTP/3: N streams share one connection, so there is nothing to + // warn about and warning anyway would be a false alarm in production. + setup(8, false); + const { container } = render(); + expect(container).toBeEmptyDOMElement(); + }); + + it("hides after dismissal", () => { + setup(5); + render(); + fireEvent.click(screen.getByRole("button", { name: "Dismiss" })); + expect(screen.queryByRole("status")).not.toBeInTheDocument(); + }); + + it("re-warns when the situation worsens after a dismissal", () => { + setup(5); + const { rerender } = render(); + fireEvent.click(screen.getByRole("button", { name: "Dismiss" })); + expect(screen.queryByRole("status")).not.toBeInTheDocument(); + + // Opening yet another conversation tab is new information — the pool is + // now fuller than when the user dismissed. + setup(6); + rerender(); + expect(screen.getByRole("status")).toHaveTextContent("6 tabs have a conversation open"); + }); + + it("stays dismissed when the count drops back", () => { + setup(6); + const { rerender } = render(); + fireEvent.click(screen.getByRole("button", { name: "Dismiss" })); + + setup(5); + rerender(); + expect(screen.queryByRole("status")).not.toBeInTheDocument(); + }); +}); diff --git a/web/src/components/StreamTabLimitBanner.tsx b/web/src/components/StreamTabLimitBanner.tsx new file mode 100644 index 0000000000..7149c99f40 --- /dev/null +++ b/web/src/components/StreamTabLimitBanner.tsx @@ -0,0 +1,63 @@ +import { useState } from "react"; +import { Button } from "@/components/ui/button"; +import { useStreamTabCount } from "@/hooks/useStreamTabCount"; +import { connectionHasLowStreamLimit } from "@/lib/streamTabRegistry"; +import { cn } from "@/lib/utils"; + +/** + * Tabs-with-an-open-conversation at which we warn. + * + * Browsers allow ~6 concurrent HTTP/1.1 connections per origin, shared across + * every tab in the profile, and each open conversation holds one for its live + * event stream. At 6 the pool is full and unrelated requests — navigation, API + * calls — queue behind the streams, which presents as the whole app hanging. + * Warn at 5 so the message arrives while the app still works, rather than + * appearing (or failing to load) once things are already wedged. + */ +const WARN_AT_TABS = 5; + +/** + * Warns when enough tabs hold a conversation stream to exhaust the browser's + * per-origin connection pool. + * + * Advisory only — it explains a stall the app cannot otherwise account for and + * names the one remedy available to the user (close a tab). It does not prevent + * the exhaustion; removing the limit needs a transport that doesn't consume an + * HTTP connection per conversation. + * + * Renders nothing when the page was served over HTTP/2/3 (multiplexed, so the + * cap doesn't bind) or where Web Locks is unavailable (count reads 0). + */ +export function StreamTabLimitBanner() { + const tabCount = useStreamTabCount(); + const [dismissedAt, setDismissedAt] = useState(null); + + // Re-arm after dismissal only if the situation gets WORSE. Dismissing at 5 + // shouldn't re-nag at 5, but crossing to 6 is new information. + const suppressed = dismissedAt !== null && tabCount <= dismissedAt; + + if (tabCount < WARN_AT_TABS || suppressed || !connectionHasLowStreamLimit()) { + return null; + } + + return ( +
+ + {tabCount} tabs have a conversation open. Browsers limit how many live connections one site + may hold, so opening more can make Omnigent slow to respond — closing a few tabs restores + it. + + +
+ ); +} diff --git a/web/src/hooks/useStreamTabCount.ts b/web/src/hooks/useStreamTabCount.ts new file mode 100644 index 0000000000..cc134545f8 --- /dev/null +++ b/web/src/hooks/useStreamTabCount.ts @@ -0,0 +1,15 @@ +// React binding for the count of same-origin tabs holding a session event +// stream. Backs the too-many-tabs warning banner. + +import { useSyncExternalStore } from "react"; +import { getStreamTabCount, subscribeStreamTabCount } from "@/lib/streamTabRegistry"; + +/** + * Subscribe to how many same-origin tabs currently hold a session event stream. + * + * @returns The observed count (including this tab), or 0 during SSR / where the + * Web Locks API is unavailable. + */ +export function useStreamTabCount(): number { + return useSyncExternalStore(subscribeStreamTabCount, getStreamTabCount, () => 0); +} diff --git a/web/src/lib/streamTabRegistry.test.ts b/web/src/lib/streamTabRegistry.test.ts new file mode 100644 index 0000000000..dcce09dbfb --- /dev/null +++ b/web/src/lib/streamTabRegistry.test.ts @@ -0,0 +1,174 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + acquireStreamSlot, + connectionHasLowStreamLimit, + getStreamTabCount, + resetStreamTabRegistryForTests, + subscribeStreamTabCount, +} from "./streamTabRegistry"; + +// Stand-in for navigator.locks. The real API can't be driven deterministically +// in jsdom (which doesn't implement it at all), and we need to control exactly +// which locks are "held" to simulate other tabs. +class FakeLockManager { + held: { name: string }[] = []; + /** Names whose request promise is still pending (i.e. lock is held). */ + private pending = new Map>(); + + request(name: string, callback: () => Promise): Promise { + this.held.push({ name }); + const promise = callback(); + this.pending.set(name, promise); + void promise.then(() => { + // Releasing resolves the callback promise → drop it from `held`, exactly + // like the browser does when the lock is released. + this.held = this.held.filter((l) => l.name !== name); + this.pending.delete(name); + }); + return promise; + } + + query(): Promise<{ held: { name: string }[] }> { + return Promise.resolve({ held: [...this.held] }); + } + + /** Test helper: simulate another tab holding a stream lock. */ + addForeignStreamLock(id: string): void { + this.held.push({ name: `omnigent.stream.${id}` }); + } +} + +let locks: FakeLockManager; + +/** Let queued microtasks (the async query + notify) settle. */ +async function settle(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +/** + * Let a macrotask turn elapse too. The release path defers its re-query past + * the microtask queue, because the lock is only dropped once the held promise + * settles. + */ +async function settleWithTimers(): Promise { + await new Promise((resolve) => { + setTimeout(resolve, 0); + }); + await settle(); +} + +beforeEach(() => { + locks = new FakeLockManager(); + vi.stubGlobal("navigator", { + locks, + // crypto.randomUUID lives on globalThis in jsdom; only locks is missing. + }); +}); + +afterEach(() => { + resetStreamTabRegistryForTests(); + vi.unstubAllGlobals(); +}); + +describe("streamTabRegistry", () => { + it("counts this tab's held stream, and releases it when the stream ends", async () => { + const notify = vi.fn(); + const unsubscribe = subscribeStreamTabCount(notify); + + const release = acquireStreamSlot(); + await settle(); + expect(getStreamTabCount()).toBe(1); + + release(); + await settleWithTimers(); + // Released promptly rather than lingering until the next poll — a closed + // conversation must stop counting against the pool immediately. + expect(getStreamTabCount()).toBe(0); + unsubscribe(); + }); + + it("counts stream locks held by other tabs", async () => { + const unsubscribe = subscribeStreamTabCount(vi.fn()); + locks.addForeignStreamLock("tab-b"); + locks.addForeignStreamLock("tab-c"); + + acquireStreamSlot(); + await settle(); + // Two peers plus this tab: the number of HTTP slots actually consumed. + expect(getStreamTabCount()).toBe(3); + unsubscribe(); + }); + + it("ignores locks that aren't session event streams", async () => { + const unsubscribe = subscribeStreamTabCount(vi.fn()); + locks.held.push({ name: "some.other.feature.lock" }); + await settle(); + expect(getStreamTabCount()).toBe(0); + unsubscribe(); + }); + + it("notifies subscribers when the count changes", async () => { + const notify = vi.fn(); + const unsubscribe = subscribeStreamTabCount(notify); + await settle(); + notify.mockClear(); + + acquireStreamSlot(); + await settle(); + expect(notify).toHaveBeenCalled(); + unsubscribe(); + }); + + it("uses a unique lock name per stream so tabs never contend", async () => { + const unsubscribe = subscribeStreamTabCount(vi.fn()); + acquireStreamSlot(); + acquireStreamSlot(); + await settle(); + // A shared name would serialize the second request behind the first and the + // count would stick at 1, silently under-reporting the real pressure. + expect(new Set(locks.held.map((l) => l.name)).size).toBe(2); + expect(getStreamTabCount()).toBe(2); + unsubscribe(); + }); + + it("degrades to a no-op where the Web Locks API is unavailable", async () => { + vi.stubGlobal("navigator", {}); + const release = acquireStreamSlot(); + await settle(); + // Count stays 0, so the banner never shows — the right failure mode for an + // advisory warning on a browser we can't measure. + expect(getStreamTabCount()).toBe(0); + expect(() => release()).not.toThrow(); + }); +}); + +describe("connectionHasLowStreamLimit", () => { + function stubNavProtocol(nextHopProtocol: string | undefined): void { + vi.stubGlobal("performance", { + getEntriesByType: () => (nextHopProtocol === undefined ? [] : [{ nextHopProtocol }]), + }); + } + + it("is true on HTTP/1.1, where the ~6-connection cap binds", () => { + stubNavProtocol("http/1.1"); + expect(connectionHasLowStreamLimit()).toBe(true); + }); + + it("is false on HTTP/2 and HTTP/3, which multiplex over one connection", () => { + stubNavProtocol("h2"); + expect(connectionHasLowStreamLimit()).toBe(false); + stubNavProtocol("h3"); + expect(connectionHasLowStreamLimit()).toBe(false); + }); + + it("assumes the limit applies when the protocol is unknown", () => { + // Some proxies omit ALPN. Fail toward showing the warning rather than + // suppressing it on exactly the setups most likely to stall. + stubNavProtocol(undefined); + expect(connectionHasLowStreamLimit()).toBe(true); + stubNavProtocol(""); + expect(connectionHasLowStreamLimit()).toBe(true); + }); +}); diff --git a/web/src/lib/streamTabRegistry.ts b/web/src/lib/streamTabRegistry.ts new file mode 100644 index 0000000000..7b8e85dd55 --- /dev/null +++ b/web/src/lib/streamTabRegistry.ts @@ -0,0 +1,231 @@ +// Counts how many same-origin tabs currently hold a session event stream. +// +// Each open conversation holds one long-lived `GET /v1/sessions/{id}/stream` +// SSE request for as long as it's bound. Browsers cap HTTP/1.1 connections at +// ~6 per origin and that budget is shared across every tab in the profile, so +// once ~6 conversations are open in parallel the held streams occupy every +// slot and unrelated requests (navigation, API calls) queue behind them — the +// UI appears hung. This module measures that pressure so the app can warn +// instead of silently stalling. +// +// Web Locks, not heartbeats: a lock is released by the browser automatically +// when its tab goes away, including a crash or force-quit. A +// localStorage/BroadcastChannel heartbeat would need an expiry window and +// would still report phantom tabs for seconds after a crash. +// +// Scope: `navigator.locks` spans same-origin contexts within ONE browser +// profile. A second browser, or a separate profile / incognito window, is +// invisible here — which is correct, because the connection pool it competes +// for is also per-profile. +// +// Counting is per BOUND STREAM, not per tab: a tab sitting on the sidebar, +// settings, or inbox holds no stream and consumes no slot, so it must not +// inflate the count. `startStreamPump` owns acquire/release for exactly the +// window in which its stream exists. + +/** Lock-name prefix identifying a held session event stream. */ +const LOCK_PREFIX = "omnigent.stream."; + +/** + * How often to re-read the lock table while something is observing the count. + * + * Tabs open and close on human timescales and the count only drives an + * advisory banner, so a slow poll is ample. `navigator.locks.query()` is + * async, so a synchronous `getSnapshot` (what `useSyncExternalStore` needs) + * has to read a cached value that this poll refreshes. + */ +const POLL_INTERVAL_MS = 3_000; + +type Listener = () => void; + +const listeners = new Set(); +let cachedCount = 0; +let pollTimer: ReturnType | null = null; +/** Guards against overlapping queries if one poll outlives its interval. */ +let querying = false; +/** A refresh arrived mid-query; re-run once the in-flight one finishes. */ +let refreshPending = false; + +/** Whether this browser exposes the Web Locks API (absent on older Safari). */ +function locksAvailable(): boolean { + return typeof navigator !== "undefined" && navigator.locks !== undefined; +} + +/** + * Whether the origin was reached over HTTP/1.1, where the ~6-connection cap + * that makes held SSE streams dangerous actually applies. + * + * HTTP/2 and HTTP/3 multiplex every request over one connection, so N held + * streams cost N cheap streams rather than N of 6 sockets and there is nothing + * to warn about. That is the shape of the Databricks Apps deployment (its + * ingress is what imposes the ~5-min HTTP/2 stream cap the pump reconnects + * around), while local `uvicorn` and plain reverse proxies serve HTTP/1.1. + * + * Read from the navigation timing entry's ALPN identifier. An unknown or empty + * value (some proxies omit it; older browsers lack the field) is treated as + * HTTP/1.1 so the warning fails toward being shown rather than silently + * suppressed on the very setups most likely to stall. + * + * @returns `true` when the banner's premise holds for this page load. + */ +export function connectionHasLowStreamLimit(): boolean { + if (typeof performance === "undefined") return true; + const [nav] = performance.getEntriesByType("navigation") as PerformanceNavigationTiming[]; + const protocol = nav?.nextHopProtocol; + if (!protocol) return true; + // ALPN ids: "http/1.0", "http/1.1", "h2", "h2c", "h3", "h3-29", … + return protocol.startsWith("http/1"); +} + +/** + * Hold a lock naming this tab's session event stream until released. + * + * The lock name is unique per call, so locks from different tabs never + * contend — every one is granted immediately and shows up in + * {@link navigator.LockManager.query}. A shared name would serialize them and + * the count would always read 1. + * + * Safe to call when Web Locks is unavailable: the returned release function is + * a no-op and the count simply stays 0 (the banner never appears, which is the + * right failure mode for an advisory warning). + * + * @returns A release function. Idempotent; call it when the stream ends. + */ +export function acquireStreamSlot(): () => void { + if (!locksAvailable()) return () => {}; + + const name = `${LOCK_PREFIX}${crypto.randomUUID()}`; + let release: (() => void) | null = null; + // The lock is held for as long as the callback's promise is pending, so + // resolving `held` is what releases it. An AbortSignal would only cancel a + // still-PENDING request, which is not what we need here. + const held = new Promise((resolve) => { + release = resolve; + }); + + void navigator.locks + .request(name, () => held) + .catch(() => { + // A rejected request means the lock was never held, so there is nothing + // to release and nothing to count. Stay silent: this is advisory only. + }); + + // Refresh promptly so a newly-bound stream is reflected without waiting a + // whole poll interval. + void refreshCount(); + + return () => { + release?.(); + release = null; + // The browser drops the lock only after the held promise settles, so an + // immediate re-query would still observe it. Defer past the microtask queue + // so the refresh sees the post-release table. + setTimeout(() => void refreshCount(), 0); + }; +} + +/** + * Re-read the lock table and notify listeners when the count changed. + * + * Failures are swallowed: a browser that rejects `query()` leaves the last + * known count in place rather than flapping the banner. + */ +async function refreshCount(): Promise { + if (!locksAvailable()) return; + if (querying) { + // Coalesce rather than drop: a slot acquired/released while a query is in + // flight must still be observed, or the count would sit stale until the + // next poll tick. + refreshPending = true; + return; + } + querying = true; + try { + const { held = [] } = await navigator.locks.query(); + const next = held.filter((lock) => lock.name?.startsWith(LOCK_PREFIX)).length; + if (next !== cachedCount) { + cachedCount = next; + for (const listener of listeners) listener(); + } + } catch { + // Keep the previous value. + } finally { + querying = false; + } + if (refreshPending) { + refreshPending = false; + await refreshCount(); + } +} + +function startPolling(): void { + if (pollTimer !== null || !locksAvailable()) return; + void refreshCount(); + pollTimer = setInterval(() => { + // Only a visible tab can show the banner, so don't spend queries while + // hidden. Returning to the tab re-reads immediately via the listener below. + if (typeof document !== "undefined" && document.hidden) return; + void refreshCount(); + }, POLL_INTERVAL_MS); +} + +function stopPolling(): void { + if (pollTimer === null) return; + clearInterval(pollTimer); + pollTimer = null; +} + +/** Re-read on focus so a tab returning to the foreground isn't stale. */ +function onVisibilityChange(): void { + if (!document.hidden) void refreshCount(); +} + +/** + * Subscribe to the number of same-origin tabs holding a session event stream. + * + * Polling runs only while at least one subscriber is listening, so an app that + * never renders the banner pays nothing. + * + * @param listener - Called whenever the count changes. + * @returns An unsubscribe function. + */ +export function subscribeStreamTabCount(listener: Listener): () => void { + listeners.add(listener); + if (listeners.size === 1) { + startPolling(); + if (typeof document !== "undefined") { + document.addEventListener("visibilitychange", onVisibilityChange); + } + } + return () => { + listeners.delete(listener); + if (listeners.size === 0) { + stopPolling(); + if (typeof document !== "undefined") { + document.removeEventListener("visibilitychange", onVisibilityChange); + } + } + }; +} + +/** + * The last observed number of stream-holding tabs. + * + * Synchronous by design (`useSyncExternalStore` requires it) and therefore + * eventually consistent — up to {@link POLL_INTERVAL_MS} stale. Reads 0 where + * Web Locks is unavailable. + * + * @returns The cached count, including this tab's own stream. + */ +export function getStreamTabCount(): number { + return cachedCount; +} + +/** Reset module state. Test-only seam so cases don't leak into each other. */ +export function resetStreamTabRegistryForTests(): void { + listeners.clear(); + stopPolling(); + cachedCount = 0; + querying = false; + refreshPending = false; +} diff --git a/web/src/main.tsx b/web/src/main.tsx index 847f5533e7..d324524ab1 100644 --- a/web/src/main.tsx +++ b/web/src/main.tsx @@ -4,6 +4,7 @@ import { createRoot } from "react-dom/client"; import { BrowserRouter } from "react-router-dom"; import App from "./App.tsx"; import { PWAUpdateBanner } from "./components/pwa/PWAUpdateBanner"; +import { StreamTabLimitBanner } from "./components/StreamTabLimitBanner"; import { ThemeProvider } from "./components/theme/ThemeProvider"; import { TooltipProvider } from "./components/ui/tooltip"; import { ImageLightboxProvider } from "./components/ImageLightbox"; @@ -118,6 +119,7 @@ void bootProbe.then((info) => { + diff --git a/web/src/store/chatStore.ts b/web/src/store/chatStore.ts index 2977174012..3e79a1e9f6 100644 --- a/web/src/store/chatStore.ts +++ b/web/src/store/chatStore.ts @@ -78,6 +78,7 @@ import type { StreamEvent, } from "@/lib/events"; import { createPresenceIdleTracker } from "@/lib/presenceIdle"; +import { acquireStreamSlot } from "@/lib/streamTabRegistry"; import { parseEvent, parseSseStream, type SseStreamResult } from "@/lib/sse"; import { clearSseLog, pushSseEvent } from "@/lib/sseEventLog"; import { childSessionsQueryKey, type ChildSessionInfo } from "@/hooks/useChildSessions"; @@ -3166,6 +3167,11 @@ export async function startStreamPump( // established stream — failed opens leave it false so a recovered first // connect is still treated as initial, not a reconnect. let hasConnected = false; + // Advertise this tab's held stream for the duration of the binding, so other + // tabs can count how many of the browser's ~6 per-origin HTTP connections are + // occupied by SSE streams. Scoped to the whole loop (not per attempt) because + // a reconnect gap still belongs to a bound stream. Released in the `finally`. + const releaseStreamSlot = acquireStreamSlot(); // A reconnect loop is inherently sequential — open → pump → reconnect — // so its awaits cannot be parallelized; no-await-in-loop doesn't apply. /* eslint-disable no-await-in-loop */ @@ -3279,6 +3285,7 @@ export async function startStreamPump( } } } finally { + releaseStreamSlot(); if (get().abortController === controller) { set({ abortController: null }); }