fix(sdk): re-earn the reconnect budget on any record and honor abort during backoff

Author: kathiekiwi

.changeset/chat-stream-mid-turn-reconnect.md

@@ -3,4 +3,4 @@
 "@trigger.dev/sdk": patch
 ---
 
-Chat streams now reconnect when the connection drops mid-turn, instead of leaving the reply stuck as if it were still generating.
+Chat in the browser now reconnects when the connection drops mid-turn, instead of leaving the reply stuck as if it were still generating.

packages/trigger-sdk/src/v3/chat.test.ts

@@ -1136,6 +1136,86 @@ describe("TriggerChatTransport", () => {
         onSessionChange.mock.calls.some(([, session]) => session && session.isStreaming === false)
       ).toBe(true);
     });
+
+    it("keeps streaming when every window delivers a single record", async () => {
+      // One record per window arrives via `primed` on the resumed connection —
+      // it must still re-earn the budget, otherwise a slow turn is truncated.
+      const WINDOWS = 8;
+      let subscribeCount = 0;
+      global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+        const urlStr = typeof url === "string" ? url : url.toString();
+        if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse();
+        if (isSessionOutSubscribeUrl(urlStr)) {
+          subscribeCount++;
+          return subscribeCount > WINDOWS
+            ? defaultSseResponse([{ type: "trigger:turn-complete" }])
+            : defaultSseResponse([
+                { type: "text-delta", id: "part-1", delta: `d${subscribeCount}` },
+              ]);
+        }
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        sessions: { "chat-slow": { publicAccessToken: "p" } },
+      });
+
+      const stream = await transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "chat-slow",
+        messageId: undefined,
+        messages: [createUserMessage("hi")],
+        abortSignal: undefined,
+      });
+      const chunks = await drainChunks(stream);
+
+      expect(subscribeCount).toBe(WINDOWS + 1);
+      expect(chunks).toHaveLength(WINDOWS);
+      expect(transport.getSession("chat-slow")?.isStreaming).toBe(false);
+    });
+
+    it("gives up after a bounded number of resubscribes", async () => {
+      // Fake timers so the 100ms..1.6s backoffs don't cost real seconds.
+      vi.useFakeTimers();
+      try {
+        let subscribeCount = 0;
+        global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+          const urlStr = typeof url === "string" ? url : url.toString();
+          if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse();
+          if (isSessionOutSubscribeUrl(urlStr)) {
+            subscribeCount++;
+            // Never any records, never settled — the pathological case.
+            return defaultSseResponse([]);
+          }
+          throw new Error(`Unexpected URL: ${urlStr}`);
+        });
+
+        const transport = new TriggerChatTransport({
+          task: "my-chat-task",
+          accessToken: () => "pat",
+          sessions: { "chat-empty": { publicAccessToken: "p" } },
+        });
+
+        const stream = await transport.sendMessages({
+          trigger: "submit-message",
+          chatId: "chat-empty",
+          messageId: undefined,
+          messages: [createUserMessage("hi")],
+          abortSignal: undefined,
+        });
+        const drained = drainChunks(stream);
+        await vi.advanceTimersByTimeAsync(10_000);
+        await drained;
+
+        // One initial connect plus the five-attempt resubscribe budget.
+        expect(subscribeCount).toBe(6);
+        expect(transport.getSession("chat-empty")?.isStreaming).toBe(false);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
   });
 
   describe("multi-tab coordination", () => {

packages/trigger-sdk/src/v3/chat.ts

@@ -1787,9 +1787,19 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             eofResubscribes < MAX_EOF_RESUBSCRIBES
           ) {
             eofResubscribes++;
-            await new Promise((resolve) =>
-              setTimeout(resolve, Math.min(100 * 2 ** (eofResubscribes - 1), 5_000))
-            );
+            // Sleep, but wake immediately on abort — otherwise a stop lands
+            // mid-backoff and the stream stays open for the rest of it.
+            await new Promise<void>((resolve) => {
+              let timer: ReturnType<typeof setTimeout>;
+              const done = () => {
+                clearTimeout(timer);
+                combinedSignal.removeEventListener("abort", done);
+                resolve();
+              };
+              timer = setTimeout(done, Math.min(100 * 2 ** (eofResubscribes - 1), 5_000));
+              combinedSignal.addEventListener("abort", done);
+            });
+            if (combinedSignal.aborted) break;
             const opened = await openWithAuthRetry();
             if (opened) return opened;
           }
@@ -1851,9 +1861,6 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
                 continue;
               }
               value = next.value;
-              // A productive connection re-earns the resubscribe budget, so a
-              // long turn spanning many long-poll windows keeps streaming.
-              eofResubscribes = 0;
             }
 
             if (combinedSignal.aborted) {
@@ -1864,6 +1871,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             }
 
             if (value.id) state.lastEventId = value.id;
+            // Any record — including the first of a resumed connection, which
+            // arrives via `primed` — re-earns the resubscribe budget, so a long
+            // turn spanning many long-poll windows keeps streaming.
+            eofResubscribes = 0;
 
             // Trigger control record (turn-complete, upgrade-required) —
             // routed by header, body is empty. Detect via the