fix(trigger-sdk): resubscribe to session.out across empty long-poll windows (#5056)

.changeset/chat-head-start-empty-poll-resubscribe.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fix `chat.headStart` relay ending the first turn after an empty `session.out` long-poll window. Resubscribes across empty windows until `turn-complete` with exponential jittered backoff.

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

@@ -40,6 +40,44 @@ import { apiClientManager } from "@trigger.dev/core/v3";
 
 // ── Helpers ────────────────────────────────────────────────────────────
 
+function sseBatch(
+  records: Array<{ seq: number; chunk?: unknown; turnComplete?: boolean; control?: string }>
+): string {
+  const encoded = records.map((record) =>
+    record.turnComplete || record.control
+      ? {
+          body: "",
+          seq_num: record.seq,
+          timestamp: 1700000000000 + record.seq,
+          headers: [["trigger-control", record.control ?? "turn-complete"]],
+        }
+      : {
+          body: JSON.stringify({ data: record.chunk, id: `p-${record.seq}` }),
+          seq_num: record.seq,
+          timestamp: 1700000000000 + record.seq,
+          headers: [],
+        }
+  );
+  return `event: batch\ndata: ${JSON.stringify({ records: encoded })}\n\n`;
+}
+
+function sseResponse(sseText: string, extraHeaders: Record<string, string> = {}): Response {
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      if (sseText) controller.enqueue(new TextEncoder().encode(sseText));
+      controller.close();
+    },
+  });
+  return new Response(body, {
+    status: 200,
+    headers: {
+      "content-type": "text/event-stream",
+      "X-Stream-Version": "v2",
+      ...extraHeaders,
+    },
+  });
+}
+
 function textStream(text: string): ReadableStream<LanguageModelV3StreamPart> {
   return simulateReadableStream({
     chunks: [
@@ -271,14 +309,7 @@ describe("chat.headStart (route handler)", () => {
       }
       // Stitched response subscribes to `.out` after handover.
       if (/\/realtime\/v1\/sessions\/[^/]+\/out$/.test(urlStr)) {
-        return new Response(
-          new ReadableStream({
-            start(c) {
-              c.close();
-            },
-          }),
-          { status: 200, headers: { "content-type": "text/event-stream" } }
-        );
+        return sseResponse(sseBatch([{ seq: 1, turnComplete: true }]));
       }
       throw new Error(`Unexpected URL: ${urlStr}`);
     });
@@ -381,17 +412,7 @@ describe("chat.headStart (route handler)", () => {
         return appendOkResponse();
       }
       if (/\/realtime\/v1\/sessions\/[^/]+\/out$/.test(urlStr)) {
-        return new Response(
-          new ReadableStream({
-            start(c) {
-              c.close();
-            },
-          }),
-          {
-            status: 200,
-            headers: { "content-type": "text/event-stream" },
-          }
-        );
+        return sseResponse(sseBatch([{ seq: 1, turnComplete: true }]));
       }
       throw new Error(`Unexpected URL: ${urlStr}`);
     });
@@ -446,17 +467,7 @@ describe("chat.headStart (route handler)", () => {
       }
       // Stitched response subscribes to `.out` after handover.
       if (/\/realtime\/v1\/sessions\/[^/]+\/out$/.test(urlStr)) {
-        return new Response(
-          new ReadableStream({
-            start(c) {
-              c.close();
-            },
-          }),
-          {
-            status: 200,
-            headers: { "content-type": "text/event-stream" },
-          }
-        );
+        return sseResponse(sseBatch([{ seq: 1, turnComplete: true }]));
       }
       throw new Error(`Unexpected URL: ${urlStr}`);
     });
@@ -528,17 +539,7 @@ describe("chat.headStart (route handler)", () => {
       // closes immediately — this test validates dispatch only, not
       // the agent-side resume.
       if (/\/realtime\/v1\/sessions\/[^/]+\/out$/.test(urlStr)) {
-        return new Response(
-          new ReadableStream({
-            start(c) {
-              c.close();
-            },
-          }),
-          {
-            status: 200,
-            headers: { "content-type": "text/event-stream" },
-          }
-        );
+        return sseResponse(sseBatch([{ seq: 1, turnComplete: true }]));
       }
       throw new Error(`Unexpected URL: ${urlStr}`);
     });
@@ -616,6 +617,127 @@ describe("chat.headStart (route handler)", () => {
     expect(approvalResponsePart.approved).toBe(true);
   });
 
+  it("resubscribes to session.out when an empty long-poll window closes mid-turn (#5056)", async () => {
+    let outCalls = 0;
+    const requestedLastEventIds: Array<string | undefined> = [];
+
+    global.fetch = vi.fn().mockImplementation(async (url: string | URL, init?: RequestInit) => {
+      const urlStr = typeof url === "string" ? url : url.toString();
+      if (urlStr.endsWith("/api/v1/sessions") || urlStr.endsWith("/api/v1/sessions/")) {
+        return createSessionResponse("chat-empty-poll");
+      }
+      if (urlStr.includes("/realtime/v1/sessions/") && urlStr.endsWith("/in/append")) {
+        return appendOkResponse();
+      }
+      if (/\/realtime\/v1\/sessions\/[^/]+\/out$/.test(urlStr)) {
+        outCalls++;
+        const headers = new Headers(init?.headers);
+        requestedLastEventIds.push(headers.get("last-event-id") ?? undefined);
+
+        if (outCalls === 1) {
+          // Window 1: long-poll closes empty (~60s elapsed without chunks)
+          return sseResponse("");
+        }
+        // Window 2: agent completes tool execution and finishes turn
+        return sseResponse(
+          sseBatch([
+            { seq: 10, chunk: { type: "text-delta", id: "t1", delta: "from agent" } },
+            { seq: 11, turnComplete: true },
+          ])
+        );
+      }
+      throw new Error(`Unexpected URL: ${urlStr}`);
+    });
+
+    const handler = chat.headStart({
+      agentId: "test-agent",
+      run: async ({ chat: chatHelper }) => {
+        return streamText({
+          ...chatHelper.toStreamTextOptions(),
+          model: new MockLanguageModelV3({
+            doStream: async () => ({ stream: textStream("step 1") }),
+          }),
+        });
+      },
+    });
+
+    const res = await withApiContext(() =>
+      handler(
+        makeRequest({
+          chatId: "chat-empty-poll",
+          trigger: "submit-message",
+          headStartMessages: [{ id: "m1", role: "user", parts: [{ type: "text", text: "hi" }] }],
+        })
+      )
+    );
+
+    const chunks = await readSSEBodyToChunks(res);
+
+    expect(outCalls).toBe(2);
+    expect(requestedLastEventIds[1]).toBeUndefined();
+
+    // Verify turn-complete and session-state chunks were enqueued
+    const turnCompleteChunk = chunks.find((c: any) => c.type === "trigger:turn-complete");
+    expect(turnCompleteChunk).toBeDefined();
+
+    const sessionStateChunk = chunks.find((c: any) => c.type === "trigger:session-state") as any;
+    expect(sessionStateChunk).toBeDefined();
+    expect(sessionStateChunk.lastEventId).toBe("11");
+
+    // Verify agent delta was received
+    const agentDelta = chunks.find(
+      (c: any) => c.type === "text-delta" && (c as any).delta === "from agent"
+    );
+    expect(agentDelta).toBeDefined();
+  });
+
+  it("errors the stream once reconnect budget is exhausted on consecutive empty windows", async () => {
+    let outCalls = 0;
+
+    global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+      const urlStr = typeof url === "string" ? url : url.toString();
+      if (urlStr.endsWith("/api/v1/sessions") || urlStr.endsWith("/api/v1/sessions/")) {
+        return createSessionResponse("chat-reconnect-budget");
+      }
+      if (urlStr.includes("/realtime/v1/sessions/") && urlStr.endsWith("/in/append")) {
+        return appendOkResponse();
+      }
+      if (/\/realtime\/v1\/sessions\/[^/]+\/out$/.test(urlStr)) {
+        outCalls++;
+        // Always close empty without turn-complete
+        return sseResponse("");
+      }
+      throw new Error(`Unexpected URL: ${urlStr}`);
+    });
+
+    const handler = chat.headStart({
+      agentId: "test-agent",
+      run: async ({ chat: chatHelper }) => {
+        return streamText({
+          ...chatHelper.toStreamTextOptions(),
+          model: new MockLanguageModelV3({
+            doStream: async () => ({ stream: textStream("step 1") }),
+          }),
+        });
+      },
+    });
+
+    const res = await withApiContext(() =>
+      handler(
+        makeRequest({
+          chatId: "chat-reconnect-budget",
+          trigger: "submit-message",
+          headStartMessages: [{ id: "m1", role: "user", parts: [{ type: "text", text: "hi" }] }],
+        })
+      )
+    );
+
+    await expect(readSSEBodyToChunks(res)).rejects.toThrow(
+      "Chat stream ended before the turn completed (reconnect budget exhausted)."
+    );
+    expect(outCalls).toBe(5);
+  });
+
   it("rejects requests missing chatId", async () => {
     global.fetch = vi.fn().mockResolvedValue(new Response("nope", { status: 500 }));
 

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

@@ -79,7 +79,7 @@ import type {
   UIMessageChunk,
 } from "ai";
 import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js";
-import { chatRunTags } from "./ai-shared.js";
+import { chatRunTags, MAX_EOF_RESUBSCRIBES, waitBeforeEofResubscribe } from "./ai-shared.js";
 import { triggerConcurrencyBody } from "./concurrency-shared.js";
 import { withResolvedExternalDeploymentId } from "./externalDeploymentId.js";
 import type { SessionTriggerConfigInput } from "./sessions.js";
@@ -958,55 +958,85 @@ async function openHandoverSession(opts: {
           // legacy chunk shape for the customer-server-to-browser hop).
           let latestEventId: string | undefined;
           let turnComplete = false;
-          // Dedicated abort signal for this agent subscription. Aborted
-          // from `onControl` the moment turn-complete fires so the
-          // `for await` loop below exits immediately instead of blocking
-          // until S2's long-poll closes (~60s). Combined with the outer
-          // `abortController.signal` via `AbortSignal.any` so a request-
-          // wide abort still tears the subscription down.
-          const subscriptionAbort = new AbortController();
-          const agentStream = await apiClient.subscribeToSessionStream<UIMessageChunk>(
-            chatId,
-            "out",
-            {
-              ...(customerLastEventId != null ? { lastEventId: customerLastEventId } : {}),
-              signal: AbortSignal.any([abortController.signal, subscriptionAbort.signal]),
-              onPart: (part) => {
-                if (part.id) latestEventId = part.id;
-              },
-              onControl: (event) => {
-                if (event.subtype === TRIGGER_CONTROL_SUBTYPE.TURN_COMPLETE) {
-                  turnComplete = true;
-                  // Synthesise the data-chunk shape for the browser
-                  // bridge. The customer-server-to-browser response is
-                  // not S2; it keeps the legacy chunk shape so the
-                  // browser's transport can recognise turn-complete the
-                  // same way it always has.
-                  controller.enqueue({
-                    type: "trigger:turn-complete",
-                  } as unknown as UIMessageChunk);
-                  // Stop the SSE read now. Without this the `for await`
-                  // can't see the control event (control records are
-                  // never enqueued into the data stream) and would idle
-                  // until S2's long-poll timeout closes the connection.
-                  subscriptionAbort.abort();
-                }
-              },
+          let emptyWindows = 0;
+
+          // A long-poll window on session.out may close empty (~60s) if the
+          // agent is running a long tool execution without writing records.
+          // Loop the subscription until turn-complete or abort, resuming from
+          // the latest event ID seen (or customerLastEventId), with backoff
+          // between empty windows.
+          while (!turnComplete && !abortController.signal.aborted) {
+            if (emptyWindows > 0) {
+              if (emptyWindows >= MAX_EOF_RESUBSCRIBES) {
+                throw new Error(
+                  "Chat stream ended before the turn completed (reconnect budget exhausted)."
+                );
+              }
+              await waitBeforeEofResubscribe(emptyWindows, abortController.signal);
+              if (abortController.signal.aborted) break;
             }
-          );
 
-          try {
-            for await (const chunk of agentStream) {
-              // Data records only — control records are routed via
-              // `onControl` above and trigger the subscription abort.
-              controller.enqueue(chunk);
-              if (turnComplete) break;
+            let windowHadRecords = false;
+            const resumeFrom = latestEventId ?? customerLastEventId;
+
+            // Dedicated abort signal for this agent subscription window. Aborted
+            // from `onControl` the moment turn-complete fires so the
+            // `for await` loop below exits immediately instead of blocking
+            // until S2's long-poll closes (~60s). Combined with the outer
+            // `abortController.signal` via `AbortSignal.any` so a request-
+            // wide abort still tears the subscription down.
+            const subscriptionAbort = new AbortController();
+            const subscriptionSignal = AbortSignal.any([
+              abortController.signal,
+              subscriptionAbort.signal,
+            ]);
+
+            const agentStream = await apiClient.subscribeToSessionStream<UIMessageChunk>(
+              chatId,
+              "out",
+              {
+                ...(resumeFrom != null ? { lastEventId: resumeFrom } : {}),
+                signal: subscriptionSignal,
+                onPart: (part) => {
+                  windowHadRecords = true;
+                  if (part.id) latestEventId = part.id;
+                },
+                onControl: (event) => {
+                  if (event.subtype === TRIGGER_CONTROL_SUBTYPE.TURN_COMPLETE) {
+                    turnComplete = true;
+                    // Synthesise the data-chunk shape for the browser
+                    // bridge. The customer-server-to-browser response is
+                    // not S2; it keeps the legacy chunk shape so the
+                    // browser's transport can recognise turn-complete the
+                    // same way it always has.
+                    controller.enqueue({
+                      type: "trigger:turn-complete",
+                    } as unknown as UIMessageChunk);
+                    // Stop the SSE read now. Without this the `for await`
+                    // can't see the control event (control records are
+                    // never enqueued into the data stream) and would idle
+                    // until S2's long-poll timeout closes the connection.
+                    subscriptionAbort.abort();
+                  }
+                },
+              }
+            );
+
+            try {
+              for await (const chunk of agentStream) {
+                // Data records only — control records are routed via
+                // `onControl` above and trigger the subscription abort.
+                controller.enqueue(chunk);
+                if (turnComplete) break;
+              }
+            } catch (err) {
+              // AbortError from `subscriptionAbort` is the expected exit
+              // path once turn-complete fires; surface anything else.
+              const isAbort = err instanceof Error && err.name === "AbortError";
+              if (!isAbort || !turnComplete) throw err;
             }
-          } catch (err) {
-            // AbortError from `subscriptionAbort` is the expected exit
-            // path once turn-complete fires; surface anything else.
-            const isAbort = err instanceof Error && err.name === "AbortError";
-            if (!isAbort || !turnComplete) throw err;
+
+            emptyWindows = windowHadRecords ? 0 : emptyWindows + 1;
           }
 
           // Final control chunk: hand the browser transport the