feat(sdk): resilient SSE reconnection in chat transport

Replace the legacy 5-attempt retry cap on SSEStreamSubscription with indefinite retry on a bounded jittered backoff. Adds a force-reconnect path so the chat transport can recover from silent-dead-socket cases on mobile (background-kill, bfcache restore) without waiting for the next backoff slot. SSEStreamSubscription: - maxRetries default Infinity (was 5), retryDelayMs 100ms (was 1s), new maxRetryDelayMs cap (5s), retryJitter 50% - retryNow(): wake an in-flight backoff - forceReconnect(): drop current connection AND wake backoff - fetchTimeoutMs (30s default): aborts stuck connect attempts that block forever on dead sockets - stallTimeoutMs (opt-in): force reconnect on silent reader - nonRetryableStatuses (default [404, 410]): short-circuit retry for stream-gone / session-closed - Fixed listener leak where each retry accumulated an abort listener on the user signal because finally only ran once the recursion unwound. Cleanup now runs per-attempt via cleanupAttempt() in both the catch (before recursion) and finally paths. TriggerChatTransport (browser): - online -> forceReconnect (existing socket may be stale) - pageshow.persisted -> forceReconnect (Safari bfcache restore) - visibilitychange -> visible only: * hidden >= 30s -> forceReconnect * hidden < 30s -> retryNow (cheap wake) - stallTimeoutMs: 60s (sized over typical agent thinking pauses) Tests: 13 vitest cases covering retry-past-legacy-cap, backoff cap, jitter variance, retryNow short-circuit, abort-during-backoff, forceReconnect during fetch and during read (verifies Last-Event-ID resume on the resumed request), fetchTimeout, stallTimeout, 404/410 short-circuit, custom nonRetryableStatuses, 503 still retries. Refs TRI-8903.
2026-05-01 18:17:56 +01:00
parent 9b8807987c
commit ea54931f4a
3 changed files with 661 additions and 30 deletions
@@ -0,0 +1,444 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { SSEStreamSubscription } from "./runStream.js";
+
+vi.setConfig({ testTimeout: 10_000 });
+
+describe("SSEStreamSubscription retry behavior", () => {
+  const originalFetch = globalThis.fetch;
+
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+    vi.restoreAllMocks();
+  });
+
+  // A response.body that emits one SSE event then closes, so each
+  // successful subscribe() exits cleanly via reader.read() done=true
+  // and the test doesn't hang reading from a long-lived stream.
+  function makeSSEResponse() {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(`id: 1\ndata: {"hello":1}\n\n`));
+        controller.close();
+      },
+    });
+    return new Response(body, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream", "X-Stream-Version": "v1" },
+    });
+  }
+
+  // Drain a ReadableStream<SSEStreamPart> until it closes or errors.
+  // Returns received chunks plus terminal state.
+  async function drain(stream: ReadableStream<{ id: string; chunk: unknown }>) {
+    const reader = stream.getReader();
+    const chunks: Array<{ id: string; chunk: unknown }> = [];
+    try {
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) return { chunks, error: undefined as Error | undefined };
+        chunks.push(value);
+      }
+    } catch (e) {
+      return { chunks, error: e as Error };
+    } finally {
+      try {
+        reader.releaseLock();
+      } catch {
+        /* already released */
+      }
+    }
+  }
+
+  it("retries past the legacy 5-attempt cap", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      if (attempts < 8) {
+        throw new TypeError("fetch failed (simulated network drop)");
+      }
+      return makeSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      // Compress the timing for the test — defaults are 100ms initial,
+      // 5s cap, retry forever; here we want fast iteration.
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+    });
+
+    const stream = await sub.subscribe();
+    const result = await drain(stream);
+
+    expect(attempts).toBe(8);
+    expect(result.error).toBeUndefined();
+    expect(result.chunks).toHaveLength(1);
+  });
+
+  it("caps the exponential backoff at maxRetryDelayMs", async () => {
+    let attempts = 0;
+    const callTimes: number[] = [];
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      callTimes.push(Date.now());
+      attempts++;
+      if (attempts < 6) {
+        throw new TypeError("fetch failed");
+      }
+      return makeSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 10,
+      maxRetryDelayMs: 30,
+    });
+
+    const stream = await sub.subscribe();
+    await drain(stream);
+
+    expect(attempts).toBe(6);
+
+    // Without the cap, backoff would be 10, 20, 40, 80, 160 (= 310ms total).
+    // With cap=30, it's 10, 20, 30, 30, 30 (= 120ms total). Allow generous
+    // slack for setTimeout jitter; the assertion is "well under uncapped".
+    const totalElapsed = callTimes.at(-1)! - callTimes[0]!;
+    expect(totalElapsed).toBeLessThan(250);
+  });
+
+  it("retryNow() wakes an in-flight backoff and reconnects immediately", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      if (attempts === 1) {
+        throw new TypeError("fetch failed");
+      }
+      return makeSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      // Backoff is intentionally long. retryNow() should short-circuit it.
+      retryDelayMs: 5_000,
+      maxRetryDelayMs: 5_000,
+    });
+
+    const subscribePromise = sub.subscribe().then(drain);
+
+    // Wait for the first attempt to fail and the backoff to start.
+    await new Promise((r) => setTimeout(r, 50));
+    sub.retryNow();
+
+    const start = Date.now();
+    const result = await subscribePromise;
+    const elapsed = Date.now() - start;
+
+    expect(attempts).toBe(2);
+    expect(result.error).toBeUndefined();
+    // Without retryNow this would have waited ~5000ms; with it, the
+    // second attempt fires nearly immediately after the first failure.
+    expect(elapsed).toBeLessThan(500);
+  });
+
+  it("respects abort signal during retry backoff", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      throw new TypeError("fetch failed");
+    });
+
+    const ac = new AbortController();
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      signal: ac.signal,
+      retryDelayMs: 1_000,
+      maxRetryDelayMs: 1_000,
+    });
+
+    const subscribePromise = sub.subscribe().then(drain);
+
+    // Let the first attempt fail and enter backoff, then abort.
+    await new Promise((r) => setTimeout(r, 50));
+    ac.abort();
+
+    const result = await subscribePromise;
+    expect(result.error).toBeUndefined();
+    // Abort should stop retries; we should have made at most a couple
+    // of attempts before the abort took effect.
+    expect(attempts).toBeLessThanOrEqual(2);
+  });
+
+  it("forceReconnect mid-read drops the stream and resumes with Last-Event-ID", async () => {
+    let attempts = 0;
+    const seenLastEventIds: Array<string | null> = [];
+    globalThis.fetch = vi.fn().mockImplementation(async (_url: string, init?: RequestInit) => {
+      attempts++;
+      const lastEventIdHeader = (init?.headers as Record<string, string> | undefined)?.[
+        "Last-Event-ID"
+      ];
+      seenLastEventIds.push(lastEventIdHeader ?? null);
+
+      if (attempts === 1) {
+        // Headers arrive immediately, body emits one chunk then hangs
+        // until aborted. The test calls forceReconnect after seeing
+        // the chunk, which should drop this stream and trigger a
+        // resume request with Last-Event-ID set.
+        const body = new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode(`id: 7\ndata: {"first":true}\n\n`));
+            init?.signal?.addEventListener("abort", () => controller.error(new Error("aborted")));
+          },
+        });
+        return new Response(body, {
+          status: 200,
+          headers: { "Content-Type": "text/event-stream", "X-Stream-Version": "v1" },
+        });
+      }
+      // Second attempt: emit a second chunk and close cleanly.
+      const body = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(new TextEncoder().encode(`id: 8\ndata: {"second":true}\n\n`));
+          controller.close();
+        },
+      });
+      return new Response(body, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream", "X-Stream-Version": "v1" },
+      });
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      fetchTimeoutMs: 60_000,
+    });
+
+    const stream = await sub.subscribe();
+    const reader = stream.getReader();
+
+    // Read the first chunk, then force-reconnect mid-stream.
+    const first = await reader.read();
+    expect(first.done).toBe(false);
+    expect((first.value!.chunk as { first?: boolean }).first).toBe(true);
+
+    sub.forceReconnect();
+
+    // Second chunk arrives from the resumed connection.
+    const second = await reader.read();
+    expect(second.done).toBe(false);
+    expect((second.value!.chunk as { second?: boolean }).second).toBe(true);
+
+    const tail = await reader.read();
+    expect(tail.done).toBe(true);
+
+    expect(attempts).toBe(2);
+    expect(seenLastEventIds[0]).toBeNull();
+    // Resumed request includes the Last-Event-ID from the first chunk.
+    expect(seenLastEventIds[1]).toBe("7");
+  });
+
+  it("forceReconnect aborts the in-flight fetch and retries", async () => {
+    let attempts = 0;
+    let firstResolve: (() => void) | undefined;
+    globalThis.fetch = vi.fn().mockImplementation(async (_url: string, init?: RequestInit) => {
+      attempts++;
+      if (attempts === 1) {
+        // Hang the first attempt forever (or until signal aborts).
+        // forceReconnect should make this attempt's signal abort and
+        // throw, taking us into the retry path.
+        return new Promise((resolve, reject) => {
+          firstResolve = () => resolve(makeSSEResponse());
+          init?.signal?.addEventListener("abort", () => {
+            reject(new DOMException("aborted", "AbortError"));
+          });
+        });
+      }
+      return makeSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      // Long fetch timeout so it doesn't fire instead of forceReconnect.
+      fetchTimeoutMs: 60_000,
+    });
+
+    const subscribePromise = sub.subscribe().then(drain);
+
+    // Let the first fetch hang, then force reconnect.
+    await new Promise((r) => setTimeout(r, 50));
+    sub.forceReconnect();
+
+    const result = await subscribePromise;
+    expect(attempts).toBe(2);
+    expect(result.error).toBeUndefined();
+    expect(result.chunks).toHaveLength(1);
+    // Sanity: the hung first fetch was abandoned, never resolved.
+    expect(firstResolve).toBeDefined();
+  });
+
+  it("aborts a slow fetch via fetchTimeoutMs and retries", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async (_url: string, init?: RequestInit) => {
+      attempts++;
+      if (attempts === 1) {
+        // Hang until aborted.
+        return new Promise((_resolve, reject) => {
+          init?.signal?.addEventListener("abort", () => {
+            reject(new DOMException("aborted", "AbortError"));
+          });
+        });
+      }
+      return makeSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      fetchTimeoutMs: 100,
+    });
+
+    const result = await sub.subscribe().then(drain);
+    expect(attempts).toBe(2);
+    expect(result.error).toBeUndefined();
+    expect(result.chunks).toHaveLength(1);
+  });
+
+  it("aborts a silent reader via stallTimeoutMs and retries", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async (_url: string, init?: RequestInit) => {
+      attempts++;
+      if (attempts === 1) {
+        // Headers arrive immediately, but the body stream emits no
+        // chunks until aborted. The stall timer should fire and
+        // force a reconnect.
+        const body = new ReadableStream<Uint8Array>({
+          start(controller) {
+            init?.signal?.addEventListener("abort", () => controller.error(new Error("aborted")));
+          },
+        });
+        return new Response(body, {
+          status: 200,
+          headers: { "Content-Type": "text/event-stream", "X-Stream-Version": "v1" },
+        });
+      }
+      return makeSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      stallTimeoutMs: 100,
+    });
+
+    const result = await sub.subscribe().then(drain);
+    expect(attempts).toBe(2);
+    expect(result.error).toBeUndefined();
+    expect(result.chunks).toHaveLength(1);
+  });
+
+  it("does not retry on 404 (stream gone)", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      return new Response("not found", { status: 404 });
+    });
+
+    const errors: Error[] = [];
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      onError: (e) => errors.push(e),
+    });
+
+    const result = await sub.subscribe().then(drain);
+    expect(attempts).toBe(1);
+    expect(result.error).toBeDefined();
+    expect(errors).toHaveLength(1);
+  });
+
+  it("does not retry on 410 (session closed)", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      return new Response("gone", { status: 410 });
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+    });
+
+    const result = await sub.subscribe().then(drain);
+    expect(attempts).toBe(1);
+    expect(result.error).toBeDefined();
+  });
+
+  it("respects custom nonRetryableStatuses", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      return new Response("forbidden", { status: 403 });
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      nonRetryableStatuses: [403],
+    });
+
+    const result = await sub.subscribe().then(drain);
+    expect(attempts).toBe(1);
+    expect(result.error).toBeDefined();
+  });
+
+  it("retries on 503 (caller-tunable nonRetryableStatuses)", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      if (attempts < 3) return new Response("unavailable", { status: 503 });
+      return makeSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      // 503 is NOT in the default non-retryable set; it should retry.
+    });
+
+    const result = await sub.subscribe().then(drain);
+    expect(attempts).toBe(3);
+    expect(result.error).toBeUndefined();
+    expect(result.chunks).toHaveLength(1);
+  });
+
+  it("applies jitter to backoff (delays vary across attempts)", async () => {
+    const callTimes: number[] = [];
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      callTimes.push(performance.now());
+      throw new TypeError("fetch failed");
+    });
+
+    const ac = new AbortController();
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      signal: ac.signal,
+      retryDelayMs: 50,
+      maxRetryDelayMs: 50,
+      retryJitter: 0.5, // 50% — final delay in [25ms, 50ms]
+    });
+
+    const promise = sub.subscribe().then(drain);
+    await new Promise((r) => setTimeout(r, 600)); // allow ~10 attempts
+    ac.abort();
+    await promise;
+
+    expect(callTimes.length).toBeGreaterThanOrEqual(5);
+
+    // Compute inter-attempt gaps (skip the first since it has no prior).
+    const gaps = callTimes.slice(1).map((t, i) => t - callTimes[i]!);
+    // Without jitter all gaps would be ~50ms. With 50% jitter they
+    // should land in [~25ms, ~50ms] and not all be identical.
+    const min = Math.min(...gaps);
+    const max = Math.max(...gaps);
+    expect(min).toBeGreaterThanOrEqual(20); // a little slack for timer scheduling
+    expect(max).toBeLessThanOrEqual(80);
+    // Variance check — at least one gap should differ from another by
+    // a measurable amount (rules out a deterministic-delay regression).
+    expect(max - min).toBeGreaterThan(2);
+  });
+});
@@ -182,8 +182,15 @@ export type SSEStreamPart<TChunk = unknown> = {
 export class SSEStreamSubscription implements StreamSubscription {
  private lastEventId: string | undefined;
  private retryCount = 0;
-  private maxRetries = 5;
-  private retryDelayMs = 1000;
+  private maxRetries: number;
+  private retryDelayMs: number;
+  private maxRetryDelayMs: number;
+  private retryJitter: number;
+  private fetchTimeoutMs: number;
+  private stallTimeoutMs: number;
+  private nonRetryableStatuses: ReadonlySet<number>;
+  private retryNowController: AbortController | null = null;
+  private internalAbort: AbortController | null = null;

  constructor(
    private url: string,
@@ -194,9 +201,69 @@ export class SSEStreamSubscription implements StreamSubscription {
      onError?: (error: Error) => void;
      timeoutInSeconds?: number;
      lastEventId?: string;
+      // Retry knobs. Defaults: retry forever, 100ms initial backoff,
+      // capped at 5s with 50% jitter. Keeps mobile clients reconnecting
+      // through transient drops without giving up after a fixed window
+      // and prevents thundering-herd when many clients reconnect after
+      // a brief server blip.
+      maxRetries?: number;
+      retryDelayMs?: number;
+      maxRetryDelayMs?: number;
+      retryJitter?: number;
+      // Per-attempt fetch timeout — aborts the connect attempt if
+      // response headers don't arrive in time. Catches stuck TCP
+      // sockets where `fetch()` blocks forever waiting on a dead
+      // server. Cleared once headers arrive; long-lived chunk reads
+      // are governed by `stallTimeoutMs` instead.
+      fetchTimeoutMs?: number;
+      // Stall detector — if no chunks arrive within this window after
+      // the connection is established, force a reconnect. Catches
+      // silent-dead-socket cases (mobile OS killed the TCP socket but
+      // the read just blocks). Disabled (`0`) by default; opt in
+      // explicitly. Servers that emit periodic keepalive comments
+      // reset the timer naturally.
+      stallTimeoutMs?: number;
+      // HTTP statuses that should NOT be retried — fail the stream
+      // permanently. `404` (stream gone) and `410` (session closed)
+      // are sensible defaults; tune per-caller for other 4xx.
+      nonRetryableStatuses?: readonly number[];
    }
  ) {
    this.lastEventId = options.lastEventId;
+    this.maxRetries = options.maxRetries ?? Infinity;
+    this.retryDelayMs = options.retryDelayMs ?? 100;
+    this.maxRetryDelayMs = options.maxRetryDelayMs ?? 5000;
+    this.retryJitter = options.retryJitter ?? 0.5;
+    this.fetchTimeoutMs = options.fetchTimeoutMs ?? 30_000;
+    this.stallTimeoutMs = options.stallTimeoutMs ?? 0;
+    this.nonRetryableStatuses = new Set(options.nonRetryableStatuses ?? [404, 410]);
+  }
+
+  /**
+   * Wake an in-flight retry backoff and reconnect immediately.
+   *
+   * No-op if no retry is currently waiting (i.e. we're already
+   * connected and reading). Use this for cheap "hint" wakeups like
+   * the `online` event or a short-hidden visibility return —
+   * `forceReconnect()` is the heavier hammer.
+   */
+  retryNow(): void {
+    this.retryNowController?.abort();
+  }
+
+  /**
+   * Drop the current connection (or wake a pending backoff) and
+   * reconnect.
+   *
+   * Use when the existing TCP socket is suspected dead but the reader
+   * hasn't noticed yet — common after a mobile tab background-kill or
+   * a Safari bfcache restore. Aborts the in-flight fetch / read so
+   * the catch path takes us through `retryConnection` and re-fetches
+   * with `Last-Event-ID`.
+   */
+  forceReconnect(): void {
+    this.internalAbort?.abort();
+    this.retryNowController?.abort();
  }

  async subscribe(): Promise<ReadableStream<SSEStreamPart>> {
@@ -206,7 +273,7 @@ export class SSEStreamSubscription implements StreamSubscription {
      async start(controller) {
        await self.connectStream(controller);
      },
-      cancel(reason) {
+      cancel() {
        self.options.onComplete?.();
      },
    });
@@ -215,25 +282,51 @@ export class SSEStreamSubscription implements StreamSubscription {
  private async connectStream(
    controller: ReadableStreamDefaultController<SSEStreamPart>
  ): Promise<void> {
+    // Two abort sources flow through `internalAbort.signal`:
+    //   - this.options.signal: caller cancel — bypass retry, exit cleanly.
+    //   - this.internalAbort: per-attempt force-reconnect / fetch-timeout
+    //     / stall-timeout — treated as a transient error, retry path runs.
+    // Use `this.options.signal?.aborted` in the catch to distinguish.
+    this.internalAbort = new AbortController();
+    const unlinkUserAbort = linkAbort(this.options.signal, this.internalAbort);
+
+    // Per-attempt fetch timeout. Cleared once response headers arrive;
+    // chunk-read latency is governed by `stallTimeoutMs` instead.
+    const fetchTimer = setTimeout(() => this.internalAbort?.abort(), this.fetchTimeoutMs);
+
+    let stallTimer: ReturnType<typeof setTimeout> | undefined;
+    const armStall = () => {
+      if (this.stallTimeoutMs <= 0) return;
+      clearTimeout(stallTimer);
+      stallTimer = setTimeout(() => this.internalAbort?.abort(), this.stallTimeoutMs);
+    };
+
+    // Idempotent — both the catch (before recursion) and the finally
+    // call this. Without the catch-side call, every retry leaks an
+    // abort listener on `this.options.signal` because the finally
+    // doesn't run until the entire recursion unwinds.
+    const cleanupAttempt = () => {
+      clearTimeout(fetchTimer);
+      clearTimeout(stallTimer);
+      unlinkUserAbort();
+      this.internalAbort = null;
+    };
+
    try {
      const headers: Record<string, string> = {
        Accept: "text/event-stream",
        ...this.options.headers,
      };
-
-      // Include Last-Event-ID header if we're resuming
-      if (this.lastEventId) {
-        headers["Last-Event-ID"] = this.lastEventId;
-      }
-
+      if (this.lastEventId) headers["Last-Event-ID"] = this.lastEventId;
      if (this.options.timeoutInSeconds) {
        headers["Timeout-Seconds"] = this.options.timeoutInSeconds.toString();
      }

      const response = await fetch(this.url, {
        headers,
-        signal: this.options.signal,
+        signal: this.internalAbort.signal,
      });
+      clearTimeout(fetchTimer);

      if (!response.ok) {
        const error = ApiError.generate(
@@ -242,22 +335,23 @@ export class SSEStreamSubscription implements StreamSubscription {
          "Could not subscribe to stream",
          Object.fromEntries(response.headers)
        );
-
        this.options.onError?.(error);
+        if (this.nonRetryableStatuses.has(response.status)) {
+          controller.error(error);
+          return;
+        }
        throw error;
      }

      if (!response.body) {
        const error = new Error("No response body");
-
        this.options.onError?.(error);
        throw error;
      }

      const streamVersion = response.headers.get("X-Stream-Version") ?? "v1";
-
-      // Reset retry count on successful connection
-      this.retryCount = 0;
+      this.retryCount = 0; // reset on success
+      armStall();

      const seenIds = new Set<string>();

@@ -268,13 +362,10 @@ export class SSEStreamSubscription implements StreamSubscription {
          new TransformStream<EventSourceMessage, SSEStreamPart>({
            transform: (chunk, chunkController) => {
              if (streamVersion === "v1") {
-                // Track the last event ID for resume support
                if (chunk.id) {
                  this.lastEventId = chunk.id;
                }
-
                const timestamp = parseRedisStreamIdTimestamp(chunk.id);
-
                chunkController.enqueue({
                  id: chunk.id ?? "unknown",
                  chunk: safeParseJSON(chunk.data),
@@ -288,13 +379,9 @@ export class SSEStreamSubscription implements StreamSubscription {

                  for (const record of data.records) {
                    this.lastEventId = record.seq_num.toString();
-
                    const parsedBody = safeParseJSON(record.body) as { data: unknown; id: string };
-                    if (seenIds.has(parsedBody.id)) {
-                      continue;
-                    }
+                    if (seenIds.has(parsedBody.id)) continue;
                    seenIds.add(parsedBody.id);
-
                    chunkController.enqueue({
                      id: record.seq_num.toString(),
                      chunk: parsedBody.data,
@@ -310,7 +397,6 @@ export class SSEStreamSubscription implements StreamSubscription {
      const reader = stream.getReader();

      try {
-        let chunkCount = 0;
        while (true) {
          const { done, value } = await reader.read();

@@ -329,7 +415,7 @@ export class SSEStreamSubscription implements StreamSubscription {
            return;
          }

-          chunkCount++;
+          armStall(); // any chunk (including server keepalives) resets the silence timer
          controller.enqueue(value);
        }
      } catch (error) {
@@ -338,7 +424,7 @@ export class SSEStreamSubscription implements StreamSubscription {
      }
    } catch (error) {
      if (this.options.signal?.aborted) {
-        // Don't retry if aborted
+        // User cancel — exit cleanly, don't retry.
        controller.close();
        this.options.onComplete?.();
        return;
@@ -350,8 +436,10 @@ export class SSEStreamSubscription implements StreamSubscription {
        return;
      }

-      // Retry on error
+      cleanupAttempt();
      await this.retryConnection(controller, error as Error);
+    } finally {
+      cleanupAttempt();
    }
  }

@@ -373,10 +461,33 @@ export class SSEStreamSubscription implements StreamSubscription {
    }

    this.retryCount++;
-    const delay = this.retryDelayMs * Math.pow(2, this.retryCount - 1);
+    const baseDelay = Math.min(
+      this.retryDelayMs * Math.pow(2, this.retryCount - 1),
+      this.maxRetryDelayMs
+    );
+    // Jitter scales the delay into [(1 - retryJitter) * base, base].
+    // E.g. retryJitter=0.5 → final delay is in [50%, 100%] of base.
+    // Spreads simultaneous reconnect attempts so many clients don't
+    // dogpile on the server right after a brief outage.
+    const delay = baseDelay * (1 - this.retryJitter * Math.random());

-    // Wait before retrying
-    await new Promise((resolve) => setTimeout(resolve, delay));
+    // Wait before retrying. The wait is wakeable: `retryNow()` aborts
+    // `retryNowController` so the timer resolves immediately and the
+    // next connect attempt starts now (e.g. on tab focus / `online`
+    // event from the browser layer).
+    this.retryNowController = new AbortController();
+    await new Promise<void>((resolve) => {
+      const timer = setTimeout(() => {
+        this.retryNowController?.signal.removeEventListener("abort", onAbort);
+        resolve();
+      }, delay);
+      const onAbort = () => {
+        clearTimeout(timer);
+        resolve();
+      };
+      this.retryNowController!.signal.addEventListener("abort", onAbort, { once: true });
+    });
+    this.retryNowController = null;

    if (this.options.signal?.aborted) {
      controller.close();
@@ -389,6 +500,22 @@ export class SSEStreamSubscription implements StreamSubscription {
  }
 }

+/**
+ * One-way abort link: when `parent` aborts, abort `child` too. Returns
+ * a cleanup that removes the listener so `parent` doesn't accumulate
+ * subscriptions across many connect attempts.
+ */
+function linkAbort(parent: AbortSignal | undefined, child: AbortController): () => void {
+  if (!parent) return () => {};
+  if (parent.aborted) {
+    child.abort();
+    return () => {};
+  }
+  const onAbort = () => child.abort();
+  parent.addEventListener("abort", onAbort, { once: true });
+  return () => parent.removeEventListener("abort", onAbort);
+}
+
 export class SSEStreamSubscriptionFactory implements StreamSubscriptionFactory {
  constructor(
    private baseUrl: string,
@@ -779,6 +779,60 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {

    return new ReadableStream<UIMessageChunk>({
      start: async (controller) => {
+        // Track the live subscription so browser wake events can act
+        // on it. Three classes of wake:
+        //   - `online`: network came back. Existing connection might
+        //     be silently dead; force a fresh one.
+        //   - `visibilitychange` → visible after long hidden: tab
+        //     was backgrounded long enough that the OS likely killed
+        //     the TCP socket. Force reconnect.
+        //   - `visibilitychange` → visible after short hidden: cheap
+        //     wake of any in-flight backoff.
+        //   - `pageshow` with `event.persisted`: bfcache restore
+        //     (mobile Safari back/forward, app-switcher resume). The
+        //     socket is definitely dead. Force reconnect.
+        let currentSubscription: SSEStreamSubscription | null = null;
+        let hiddenSince: number | null = null;
+        const FORCE_RECONNECT_AFTER_HIDDEN_MS = 30_000;
+
+        const onVisibilityChange = () => {
+          if (typeof document === "undefined") return;
+          if (document.visibilityState === "hidden") {
+            hiddenSince = Date.now();
+            return;
+          }
+          const wasHiddenForMs = hiddenSince ? Date.now() - hiddenSince : 0;
+          hiddenSince = null;
+          if (wasHiddenForMs >= FORCE_RECONNECT_AFTER_HIDDEN_MS) {
+            currentSubscription?.forceReconnect();
+          } else {
+            currentSubscription?.retryNow();
+          }
+        };
+
+        const onPageShow = (event: Event) => {
+          // PageTransitionEvent in browsers; type guard via `persisted`.
+          if ((event as PageTransitionEvent).persisted) {
+            currentSubscription?.forceReconnect();
+          }
+        };
+
+        const onOnline = () => currentSubscription?.forceReconnect();
+
+        const teardownWakeListeners =
+          typeof document !== "undefined" && typeof window !== "undefined"
+            ? (() => {
+                document.addEventListener("visibilitychange", onVisibilityChange);
+                window.addEventListener("online", onOnline);
+                window.addEventListener("pageshow", onPageShow);
+                return () => {
+                  document.removeEventListener("visibilitychange", onVisibilityChange);
+                  window.removeEventListener("online", onOnline);
+                  window.removeEventListener("pageshow", onPageShow);
+                };
+              })()
+            : () => {};
+
        const connectSseOnce = async (token: string) => {
          const subscription = new SSEStreamSubscription(streamUrl, {
            headers: {
@@ -789,7 +843,12 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
            signal: combinedSignal,
            timeoutInSeconds: this.streamTimeoutSeconds,
            lastEventId: state.lastEventId,
+            // Catch silent-dead-socket: if no chunk (or server
+            // keepalive) arrives in 60s, force reconnect. Sized
+            // generously over typical agent thinking pauses.
+            stallTimeoutMs: 60_000,
          });
+          currentSubscription = subscription;
          const sseStream = await subscription.subscribe();
          const reader = sseStream.getReader();
          try {
@@ -929,6 +988,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
          }
          controller.error(error);
        } finally {
+          teardownWakeListeners();
          this.activeStreams.delete(chatId);
          this.coordinator?.release(chatId);
        }