fix: distinguish stall exhaustion errors

Author: gtremper

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

@@ -83,7 +83,7 @@ describe("SSE retry exhaustion", () => {
     };
     const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
 
-    await expect(reader.read()).rejects.toThrow("Stream connection retries exhausted");
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
     expect(attempts).toBe(3);
   });
 
@@ -118,7 +118,7 @@ describe("SSE retry exhaustion", () => {
     };
     const reader = await open({ stallTimeoutMs: 100, maxRetries: Infinity, maxStallRetries: 2 });
 
-    await expect(reader.read()).rejects.toThrow("Stream connection retries exhausted");
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
     expect(attempts).toBe(3);
   });
 
@@ -131,7 +131,7 @@ describe("SSE retry exhaustion", () => {
     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 connection retries exhausted");
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
     expect(attempts).toBe(5);
   });
 
@@ -182,7 +182,7 @@ describe("SSE retry exhaustion", () => {
       stallTimeoutMs: 100,
     });
 
-    await expect(reader.read()).rejects.toThrow("Stream connection retries exhausted");
+    await expect(reader.read()).rejects.toThrow("Stream stalled: no records received");
     expect(attempts).toBe(5);
   });
 });

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

@@ -221,6 +221,7 @@ export class SSEStreamSubscription implements StreamSubscription {
   private retryCount = 0;
   private stallCount = 0;
   private maxRetries: number;
+  private maxStallRetries: number;
   private retryDelayMs: number;
   private maxRetryDelayMs: number;
   private retryJitter: number;
@@ -296,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;
@@ -653,13 +655,12 @@ export class SSEStreamSubscription implements StreamSubscription {
       return;
     }
 
-    if (
-      this.retryCount >= this.maxRetries ||
-      this.stallCount > (this.options.maxStallRetries ?? Infinity)
-    ) {
+    const stallsExhausted = this.stallCount > this.maxStallRetries;
+    if (this.retryCount >= this.maxRetries || stallsExhausted) {
       // Internal timeouts are failures, not caller cancellation.
-      const finalError =
-        error?.name === "AbortError"
+      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);