fix: stop retrying silent chat streams indefinitely

.changeset/quiet-chat-stream-retries.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Chat streams now report `Stream stalled: no records received` after five retries of a connected stream that sends no records. Network failures and browser wakeups retain automatic recovery. Healthy tool calls with no records for about six minutes also reach this silence limit. Watch subscriptions remain unlimited, and caller cancellation still closes cleanly.

packages/core/src/v3/apiClient/runStream-retries.test.ts

@@ -0,0 +1,188 @@
+import { createServer, type Server, type ServerResponse } from "node:http";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { SSEStreamSubscription } from "./runStream.js";
+
+describe("SSE retry exhaustion", () => {
+  let server: Server;
+  let url: string;
+  let abort: AbortController;
+  let attempts: number;
+  let respond: (response: ServerResponse) => void;
+  let subscription: SSEStreamSubscription;
+
+  beforeEach(async () => {
+    attempts = 0;
+    abort = new AbortController();
+    server = createServer((_request, response) => {
+      attempts++;
+      respond(response);
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("Expected a TCP address");
+    url = `http://127.0.0.1:${address.port}`;
+  });
+
+  afterEach(async () => {
+    abort.abort();
+    server.closeAllConnections();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+  });
+
+  async function open(
+    options: {
+      fetchTimeoutMs?: number;
+      stallTimeoutMs?: number;
+      maxRetries?: number;
+      maxStallRetries?: number;
+    } = {}
+  ) {
+    subscription = new SSEStreamSubscription(url, {
+      signal: abort.signal,
+      maxRetries: 2,
+      retryDelayMs: 1,
+      retryJitter: 0,
+      ...options,
+    });
+    return (await subscription.subscribe()).getReader();
+  }
+
+  it.each(["fetch", "stall"] as const)(
+    "reports exhausted %s timeouts as failures",
+    async (failure) => {
+      respond = (response) => {
+        if (failure === "stall") {
+          response.writeHead(200, { "Content-Type": "text/event-stream" });
+          response.flushHeaders();
+        }
+      };
+      const reader = await open({
+        fetchTimeoutMs: failure === "fetch" ? 100 : 1_000,
+        stallTimeoutMs: 100,
+      });
+
+      await expect(reader.read()).rejects.toMatchObject({
+        name: "Error",
+        message: "Stream connection retries exhausted",
+      });
+      expect(attempts).toBe(3);
+    }
+  );
+
+  it.each([
+    ["comment", ": keepalive\n\n"],
+    ["keepalive event", "event: keepalive\ndata: {}\n\n"],
+    ["empty batch", 'event: batch\ndata: {"records":[]}\n\n'],
+  ])("does not reset the retry budget after a %s", async (_name, payload) => {
+    respond = (response) => {
+      response.writeHead(200, {
+        "Content-Type": "text/event-stream",
+        "X-Stream-Version": "v2",
+      });
+      response.write(payload);
+    };
+    const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
+
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
+    expect(attempts).toBe(3);
+  });
+
+  it("restores the retry budget after a decoded record", async () => {
+    respond = (response) => {
+      if (attempts !== 3) {
+        response.writeHead(503).end();
+        return;
+      }
+      response.writeHead(200, { "Content-Type": "text/event-stream" });
+      response.write('id: 1\ndata: {"hello":1}\n\n');
+    };
+    const reader = await open({ stallTimeoutMs: 100 });
+
+    expect(await reader.read()).toMatchObject({ done: false, value: { chunk: { hello: 1 } } });
+    await expect(reader.read()).rejects.toMatchObject({ status: 503 });
+    expect(attempts).toBe(5);
+  });
+
+  it("closes without retries when the caller cancels", async () => {
+    respond = () => abort.abort();
+    const reader = await open();
+
+    expect(await reader.read()).toEqual({ done: true, value: undefined });
+    expect(attempts).toBe(1);
+  });
+
+  it("limits silent stalls without a general retry limit", async () => {
+    respond = (response) => {
+      response.writeHead(200, { "Content-Type": "text/event-stream" });
+      response.flushHeaders();
+    };
+    const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
+
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
+    expect(attempts).toBe(3);
+  });
+
+  it("restores the stall budget only after a decoded record", async () => {
+    respond = (response) => {
+      response.writeHead(200, { "Content-Type": "text/event-stream" });
+      response.flushHeaders();
+      if (attempts === 3) response.write('id: 1\ndata: {"hello":1}\n\n');
+    };
+    const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
+
+    expect(await reader.read()).toMatchObject({ done: false, value: { chunk: { hello: 1 } } });
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
+    expect(attempts).toBe(5);
+  });
+
+  it.each(["http", "fetch", "body", "wake"] as const)(
+    "does not charge %s failures to the stall budget",
+    async (failure) => {
+      respond = (response) => {
+        if (attempts === 5) {
+          response.writeHead(200, { "Content-Type": "text/event-stream" });
+          response.end('id: 1\ndata: {"hello":1}\n\n');
+        } else if (failure === "http") {
+          response.writeHead(503).end();
+        } else if (failure !== "fetch") {
+          response.writeHead(200, { "Content-Type": "text/event-stream" });
+          response.write(": keepalive\n\n");
+          setTimeout(() => {
+            if (failure === "wake") subscription.forceReconnect();
+            else response.destroy();
+          }, 10);
+        }
+      };
+      const reader = await open({
+        maxRetries: Infinity,
+        maxStallRetries: 0,
+        fetchTimeoutMs: 100,
+        stallTimeoutMs: 1_000,
+      });
+
+      expect(await reader.read()).toMatchObject({ done: false, value: { chunk: { hello: 1 } } });
+      expect(attempts).toBe(5);
+    }
+  );
+
+  it("retains the stall budget across connection failures and wakeups", async () => {
+    respond = (response) => {
+      if (attempts === 2) {
+        response.writeHead(503).end();
+      } else if (attempts !== 4) {
+        response.writeHead(200, { "Content-Type": "text/event-stream" });
+        response.flushHeaders();
+        if (attempts === 3) setTimeout(() => subscription.forceReconnect(), 10);
+      }
+    };
+    const reader = await open({
+      maxRetries: Infinity,
+      maxStallRetries: 1,
+      fetchTimeoutMs: 100,
+      stallTimeoutMs: 100,
+    });
+
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
+    expect(attempts).toBe(5);
+  });
+});

packages/core/src/v3/apiClient/runStream.ts

@@ -219,7 +219,9 @@ export class SSEStreamSubscription implements StreamSubscription {
   private lastEventId: string | undefined;
   private from: "beginning" | "latest";
   private retryCount = 0;
+  private stallCount = 0;
   private maxRetries: number;
+  private maxStallRetries: number;
   private retryDelayMs: number;
   private maxRetryDelayMs: number;
   private retryJitter: number;
@@ -273,9 +275,11 @@ export class SSEStreamSubscription implements StreamSubscription {
       // 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.
+      // explicitly. Only decoded records reset the timer.
       stallTimeoutMs?: number;
+      // Reconnects after stall timeouts before the stream errors.
+      // Only decoded records restore this budget. Defaults to Infinity.
+      maxStallRetries?: number;
       // HTTP statuses that should NOT be retried — fail the stream
       // permanently. Defaults cover the permanent client-error set:
       // `400` (bad request), `404` (stream gone), `409` (conflict),
@@ -293,6 +297,7 @@ export class SSEStreamSubscription implements StreamSubscription {
     this.lastEventId = options.lastEventId;
     this.from = options.from ?? "beginning";
     this.maxRetries = options.maxRetries ?? Infinity;
+    this.maxStallRetries = options.maxStallRetries ?? Infinity;
     this.retryDelayMs = options.retryDelayMs ?? 100;
     this.maxRetryDelayMs = options.maxRetryDelayMs ?? 5000;
     this.retryJitter = options.retryJitter ?? 0.5;
@@ -403,7 +408,11 @@ export class SSEStreamSubscription implements StreamSubscription {
     const armStall = () => {
       if (this.stallTimeoutMs <= 0) return;
       clearTimeout(stallTimer);
-      stallTimer = setTimeout(() => this.internalAbort?.abort(), this.stallTimeoutMs);
+      stallTimer = setTimeout(() => {
+        if (!this.internalAbort || this.internalAbort.signal.aborted) return;
+        this.stallCount++;
+        this.internalAbort.abort();
+      }, this.stallTimeoutMs);
     };
 
     // Idempotent — both the catch (before recursion) and the finally
@@ -461,7 +470,6 @@ export class SSEStreamSubscription implements StreamSubscription {
 
       const streamVersion = response.headers.get("X-Stream-Version") ?? "v1";
       this.sessionSettled = response.headers.get("X-Session-Settled") === "true";
-      this.retryCount = 0; // reset on success
       armStall();
 
       // Dedup window for record ids. Bounded with FIFO eviction so a
@@ -576,8 +584,11 @@ export class SSEStreamSubscription implements StreamSubscription {
             return;
           }
 
-          armStall(); // any chunk (including server keepalives) resets the silence timer
+          armStall(); // Each decoded record resets the silence timer.
           this.authRefreshed = false;
+          // Headers alone do not establish stream recovery.
+          this.retryCount = 0;
+          this.stallCount = 0;
           controller.enqueue(value);
         }
       } catch (error) {
@@ -644,8 +655,14 @@ export class SSEStreamSubscription implements StreamSubscription {
       return;
     }
 
-    if (this.retryCount >= this.maxRetries) {
-      const finalError = error || new Error("Max retries reached");
+    const stallsExhausted = this.stallCount > this.maxStallRetries;
+    if (this.retryCount >= this.maxRetries || stallsExhausted) {
+      // Internal timeouts are failures, not caller cancellation.
+      const finalError = stallsExhausted
+        ? new Error("Stream stalled: no records received")
+        : error?.name === "AbortError"
+          ? new Error("Stream connection retries exhausted")
+          : error || new Error("Max retries reached");
       controller.error(finalError);
       this.options.onError?.(finalError);
       return;