fix(sdk): keep the successor stream's abort controller when an aborted stream tears down

Author: kathiekiwi

.changeset/chat-stream-supersede-race.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fixed a race where quickly restarting a chat stream could break stop and reconnect for the new stream.

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

@@ -132,6 +132,35 @@ function defaultSseResponse(
   });
 }
 
+/**
+ * An SSE response whose body stays open until the request signal aborts.
+ * Models a live subscription sitting on a quiet server.
+ */
+function openSseResponse(signal?: AbortSignal | null): Response {
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      const onAbort = () => {
+        const err = new Error("aborted");
+        err.name = "AbortError";
+        try {
+          controller.error(err);
+        } catch {
+          /* already errored */
+        }
+      };
+      if (signal?.aborted) onAbort();
+      else signal?.addEventListener("abort", onAbort, { once: true });
+    },
+  });
+  return new Response(body, {
+    status: 200,
+    headers: {
+      "content-type": "text/event-stream",
+      "X-Stream-Version": "v2",
+    },
+  });
+}
+
 function authError(status = 401): Response {
   return new Response(JSON.stringify({ error: "Unauthorized", name: "TriggerApiError", status }), {
     status,
@@ -1524,6 +1553,112 @@ describe("TriggerChatTransport", () => {
     });
   });
 
+  describe("superseded stream teardown", () => {
+    it("keeps the successor's controller registered when the aborted stream tears down", async () => {
+      vi.useFakeTimers();
+      try {
+        let appendCount = 0;
+        global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+          const urlStr = typeof url === "string" ? url : url.toString();
+          if (isSessionStreamAppendUrl(urlStr)) {
+            appendCount++;
+            return defaultAppendResponse();
+          }
+          // Quiet stream: EOF, no records, never settled — watch keeps it open.
+          if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse([]);
+          throw new Error(`Unexpected URL: ${urlStr}`);
+        });
+
+        const transport = new TriggerChatTransport({
+          task: "my-chat-task",
+          accessToken: () => "pat",
+          watch: true,
+          sessions: { "chat-race": { publicAccessToken: "p", isStreaming: true } },
+        });
+
+        const send = () =>
+          transport.sendMessages({
+            trigger: "submit-message" as const,
+            chatId: "chat-race",
+            messageId: undefined,
+            messages: [createUserMessage("hi")],
+            abortSignal: undefined,
+          });
+
+        const first = drainChunks(await send());
+        await vi.advanceTimersByTimeAsync(1_000);
+
+        // Supersede: the new stream registers its controller synchronously,
+        // the aborted one tears down a microtask later.
+        const second = await send();
+        let secondClosed = false;
+        const secondDrain = drainChunks(second).then(() => {
+          secondClosed = true;
+        });
+        await first;
+        await vi.advanceTimersByTimeAsync(1_000);
+
+        // stopGeneration posts the stop chunk either way — only the
+        // closing assertion proves it found the successor to abort.
+        appendCount = 0;
+        expect(await transport.stopGeneration("chat-race")).toBe(true);
+        await vi.advanceTimersByTimeAsync(1_000);
+        expect(appendCount).toBe(1);
+        expect(secondClosed).toBe(true);
+
+        transport.dispose();
+        await secondDrain;
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("keeps the tab claim the successor took (multi-tab)", async () => {
+      vi.useFakeTimers();
+      try {
+        global.fetch = vi.fn().mockImplementation(async (url: string | URL, init?: RequestInit) => {
+          const urlStr = typeof url === "string" ? url : url.toString();
+          if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse();
+          // Open SSE that only ends when the subscription is aborted, so
+          // the superseded stream tears down while the successor is live.
+          if (isSessionOutSubscribeUrl(urlStr)) return openSseResponse(init?.signal);
+          throw new Error(`Unexpected URL: ${urlStr}`);
+        });
+
+        const transport = new TriggerChatTransport({
+          task: "my-chat-task",
+          accessToken: () => "pat",
+          multiTab: true,
+          sessions: { "chat-race-tab": { publicAccessToken: "p", isStreaming: true } },
+        });
+
+        const send = () =>
+          transport.sendMessages({
+            trigger: "submit-message" as const,
+            chatId: "chat-race-tab",
+            messageId: undefined,
+            messages: [createUserMessage("hi")],
+            abortSignal: undefined,
+          });
+
+        const first = drainChunks(await send());
+        await vi.advanceTimersByTimeAsync(1_000);
+        const secondDrain = drainChunks(await send());
+        await first;
+        await vi.advanceTimersByTimeAsync(1_000);
+
+        // The superseded stream must not release the claim its successor
+        // holds — otherwise this tab flips to read-only mid-turn.
+        expect(transport.hasClaim("chat-race-tab")).toBe(true);
+
+        transport.dispose();
+        await secondDrain;
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+  });
+
   describe("multi-tab coordination", () => {
     it("isReadOnly defaults to false when multiTab is disabled", () => {
       const transport = new TriggerChatTransport({

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

@@ -2054,8 +2054,13 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
           controller.error(error);
         } finally {
           teardownWakeListeners();
-          this.activeStreams.delete(chatId);
-          this.coordinator?.release(chatId);
+          // Only clear the entry (and drop the tab claim) if it is still
+          // ours — a superseding send registers its controller before this
+          // teardown runs, and owns the claim from then on.
+          if (this.activeStreams.get(chatId) === internalAbort) {
+            this.activeStreams.delete(chatId);
+            this.coordinator?.release(chatId);
+          }
         }
       },
       // A consumer that stops reading without aborting (drops the reader)