Merge remote-tracking branch 'origin/feat/dashboard-agent-flows' into feat/dashboard-agent-ui

kathiekiwi · kathiekiwi · commit 4378c691658d · 2026-08-09T23:01:29.000Z
diff --git a/.changeset/chat-stream-mid-turn-reconnect.md b/.changeset/chat-stream-mid-turn-reconnect.md
@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Chat in the browser now reconnects when the connection drops mid-turn, instead of leaving the reply stuck as if it were still generating.
diff --git a/packages/core/src/v3/apiClient/runStream.ts b/packages/core/src/v3/apiClient/runStream.ts
@@ -206,6 +206,13 @@ export class SSEStreamSubscription implements StreamSubscription {
   private cancelledByConsumer = false;
   private completeNotified = false;
 
+  /**
+   * True when the most recent response carried `X-Session-Settled: true` —
+   * the server has no more records coming, so a clean end of the body is
+   * terminal rather than the end of a long-poll window.
+   */
+  sessionSettled = false;
+
   constructor(
     private url: string,
     private options: {
@@ -414,6 +421,7 @@ 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();
 
diff --git a/packages/trigger-sdk/src/v3/chat.test.ts b/packages/trigger-sdk/src/v3/chat.test.ts
@@ -1054,6 +1054,170 @@ describe("TriggerChatTransport", () => {
     });
   });
 
+  describe("stream body ends mid-turn", () => {
+    it("resubscribes from the last event id when the close was not settled", async () => {
+      const subscribeHeaders: Headers[] = [];
+      global.fetch = vi.fn().mockImplementation(async (url: string | URL, init?: RequestInit) => {
+        const urlStr = typeof url === "string" ? url : url.toString();
+        if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse();
+        if (isSessionOutSubscribeUrl(urlStr)) {
+          subscribeHeaders.push(new Headers(init?.headers));
+          // First connection ends mid-turn: one chunk, no turn-complete,
+          // no `X-Session-Settled`.
+          return subscribeHeaders.length === 1
+            ? defaultSseResponse([{ type: "text-start", id: "part-1" }])
+            : defaultSseResponse([
+                { type: "text-delta", id: "part-1", delta: "resumed" },
+                { type: "trigger:turn-complete" },
+              ]);
+        }
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        sessions: { "chat-eof": { publicAccessToken: "p" } },
+      });
+
+      const stream = await transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "chat-eof",
+        messageId: undefined,
+        messages: [createUserMessage("hi")],
+        abortSignal: undefined,
+      });
+      const chunks = await drainChunks(stream);
+
+      expect(subscribeHeaders).toHaveLength(2);
+      expect(subscribeHeaders[1]?.get("Last-Event-ID")).toBe("1");
+      expect(chunks).toEqual([
+        { type: "text-start", id: "part-1" },
+        { type: "text-delta", id: "part-1", delta: "resumed" },
+      ]);
+      expect(transport.getSession("chat-eof")?.isStreaming).toBe(false);
+    });
+
+    it("stops and clears isStreaming when the close was settled", async () => {
+      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++;
+          const response = defaultSseResponse([{ type: "text-start", id: "part-1" }]);
+          const headers = new Headers(response.headers);
+          headers.set("X-Session-Settled", "true");
+          return new Response(response.body, { status: 200, headers });
+        }
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+
+      const onSessionChange = vi.fn();
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        onSessionChange,
+        sessions: { "chat-settled": { publicAccessToken: "p" } },
+      });
+
+      const stream = await transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "chat-settled",
+        messageId: undefined,
+        messages: [createUserMessage("hi")],
+        abortSignal: undefined,
+      });
+      await drainChunks(stream);
+
+      expect(subscribeCount).toBe(1);
+      expect(transport.getSession("chat-settled")?.isStreaming).toBe(false);
+      expect(
+        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", () => {
     it("isReadOnly defaults to false when multiTab is disabled", () => {
       const transport = new TriggerChatTransport({
diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts
@@ -1759,6 +1759,60 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
           }
         };
 
+        const openWithAuthRetry = async () => {
+          try {
+            return await connectSseOnce(state.publicAccessToken);
+          } catch (e) {
+            if (!isAuthError(e)) throw e;
+            const fresh = await this.resolveAccessToken({ chatId });
+            state.publicAccessToken = fresh;
+            this.notifySessionChange(chatId, state);
+            return await connectSseOnce(fresh);
+          }
+        };
+
+        // A body that ends without a turn-complete is only terminal when the
+        // server says the session settled — otherwise the turn is still
+        // running and we lost the connection (long-poll window closed, proxy
+        // restarted). Resubscribe from `state.lastEventId`, bounded so a
+        // permanently empty stream can't spin.
+        const MAX_EOF_RESUBSCRIBES = 5;
+        let eofResubscribes = 0;
+
+        const resumeAfterEof = async () => {
+          while (
+            state.isStreaming &&
+            !currentSubscription?.sessionSettled &&
+            !combinedSignal.aborted &&
+            eofResubscribes < MAX_EOF_RESUBSCRIBES
+          ) {
+            eofResubscribes++;
+            // 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;
+          }
+
+          // Settled close, or the turn is gone — tell the UI instead of
+          // leaving it spinning on a stream nobody will finish.
+          if (state.isStreaming) {
+            state.isStreaming = false;
+            this.notifySessionChange(chatId, state);
+          }
+          return null;
+        };
+
         try {
           let reader: ReadableStreamDefaultReader<{
             id: string;
@@ -1767,30 +1821,13 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
           }>;
           let primed: { id: string; chunk: unknown; timestamp: number } | undefined;
 
-          try {
-            const opened = await connectSseOnce(state.publicAccessToken);
-            if (opened === null) {
-              controller.close();
-              return;
-            }
-            reader = opened.reader;
-            primed = opened.primed;
-          } catch (e) {
-            if (isAuthError(e)) {
-              const fresh = await this.resolveAccessToken({ chatId });
-              state.publicAccessToken = fresh;
-              this.notifySessionChange(chatId, state);
-              const opened = await connectSseOnce(fresh);
-              if (opened === null) {
-                controller.close();
-                return;
-              }
-              reader = opened.reader;
-              primed = opened.primed;
-            } else {
-              throw e;
-            }
+          const opened = (await openWithAuthRetry()) ?? (await resumeAfterEof());
+          if (opened === null) {
+            controller.close();
+            return;
           }
+          reader = opened.reader;
+          primed = opened.primed;
 
           this.emitEvent({
             type: "stream-connected",
@@ -1814,8 +1851,14 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             } else {
               const next = await reader.read();
               if (next.done) {
-                controller.close();
-                return;
+                const resumed = await resumeAfterEof();
+                if (resumed === null) {
+                  controller.close();
+                  return;
+                }
+                reader = resumed.reader;
+                primed = resumed.primed;
+                continue;
               }
               value = next.value;
             }
@@ -1828,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
