From bf3834ac2f95482a9212c3bfc112931f5796eea1 Mon Sep 17 00:00:00 2001 From: Mahmoud Ghazy Date: Sun, 25 Oct 2026 13:33:22 +0300 Subject: [PATCH] fix(trigger-sdk): resubscribe to session.out across empty long-poll windows (#5056) In chat.headStart, stitchHandoverStream reads the agent's session.out stream to relay chunks to the browser until turn-complete. When a long-running tool or agent step executes for longer than the S2 long-poll deadline (~60s) without emitting chunks, the SSE connection ends with an empty window before turn-complete. Previously, this caused the relay to close prematurely, writing only trigger:session-state and cutting off the first turn while the agent run was still in progress. This commit loops the session.out subscription until turn-complete or request abort: - Resumes each subscription window from latestEventId ?? customerLastEventId. - Applies exponential jittered backoff via waitBeforeEofResubscribe between empty windows. - Resets consecutive empty windows count when chunks are received. - Binds consecutive empty reconnects to MAX_EOF_RESUBSCRIBES to prevent infinite spins if a turn hangs permanently. - Adds comprehensive unit tests covering empty window resubscribe and reconnect budget exhaustion. Fixes #5056 --- .../chat-head-start-empty-poll-resubscribe.md | 5 + .../trigger-sdk/src/v3/chat-server.test.ts | 204 ++++++++++++++---- packages/trigger-sdk/src/v3/chat-server.ts | 124 +++++++---- 3 files changed, 245 insertions(+), 88 deletions(-) create mode 100644 .changeset/chat-head-start-empty-poll-resubscribe.md diff --git a/.changeset/chat-head-start-empty-poll-resubscribe.md b/.changeset/chat-head-start-empty-poll-resubscribe.md new file mode 100644 index 00000000000..80b7fed572a --- /dev/null +++ b/.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. diff --git a/packages/trigger-sdk/src/v3/chat-server.test.ts b/packages/trigger-sdk/src/v3/chat-server.test.ts index ae85f092d41..987f4538c87 100644 --- a/packages/trigger-sdk/src/v3/chat-server.test.ts +++ b/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 = {}): Response { + const body = new ReadableStream({ + 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 { 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 = []; + + 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 })); diff --git a/packages/trigger-sdk/src/v3/chat-server.ts b/packages/trigger-sdk/src/v3/chat-server.ts index 22fd2e284e6..3c19115dcd8 100644 --- a/packages/trigger-sdk/src/v3/chat-server.ts +++ b/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( - 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( + 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