Report stream subscribe failures through onError lifecycle

Co-authored-by: Eric Allam <eric@trigger.dev>

Author: cursoragent

docs/tasks/streams.mdx

@@ -618,8 +618,9 @@ The default payload sent to your task is a rich, typed object that includes:
 - headers passed through transport can be object, `Headers`, or tuple arrays
 
 `onError` receives phase-aware details (`payloadMapper`, `triggerOptions`, `triggerTask`,
-`onTriggeredRun`, `consumeTrackingStream`, `reconnect`) plus `chatId`, optional `runId`,
-and the underlying `error` (non-Error throws are normalized to `Error` instances).
+`streamSubscribe`, `onTriggeredRun`, `consumeTrackingStream`, `reconnect`) plus `chatId`,
+optional `runId`, and the underlying `error` (non-Error throws are normalized to `Error`
+instances).
 
 ```ts
 import type { TriggerChatRunState, TriggerChatRunStore } from "@trigger.dev/ai";

packages/ai/CHANGELOG.md

@@ -13,5 +13,5 @@
 - Added reconnect lifecycle handling that cleans run state after completion/error and gracefully returns `null` when reconnect cannot be resumed.
 - Added explicit helper option types for chat send/reconnect request inputs.
 - Added optional `onError` callback support for observing non-fatal transport issues.
-- Added phase-aware `onError` reporting across send, reconnect, and stream-consumption paths.
+- Added phase-aware `onError` reporting across send, stream-subscribe, reconnect, and stream-consumption paths.
 - Added normalization of non-Error throw values into Error instances before `onError` reporting.

packages/ai/README.md

@@ -135,7 +135,7 @@ You can optionally provide `onError` to observe non-fatal transport errors
 
 The callback receives:
 
-- `phase`: `"payloadMapper" | "triggerOptions" | "triggerTask" | "onTriggeredRun" | "consumeTrackingStream" | "reconnect"`
+- `phase`: `"payloadMapper" | "triggerOptions" | "triggerTask" | "streamSubscribe" | "onTriggeredRun" | "consumeTrackingStream" | "reconnect"`
 - `chatId`
 - `runId` (may be `undefined` before a run is created)
 - `error`

packages/ai/src/chatTransport.test.ts

@@ -1074,6 +1074,103 @@ describe("TriggerChatTransport", function () {
     ).rejects.toThrowError("trigger task failed root");
   });
 
+  it("reports stream subscription failures through onError", async function () {
+    const errors: TriggerChatTransportError[] = [];
+    const runStore = new InMemoryTriggerChatRunStore();
+
+    const server = await startServer(function (req, res) {
+      if (req.method === "POST" && req.url === "/api/v1/tasks/chat-task/trigger") {
+        res.writeHead(200, {
+          "content-type": "application/json",
+          "x-trigger-jwt": "pk_stream_subscribe_error",
+        });
+        res.end(JSON.stringify({ id: "run_stream_subscribe_error" }));
+        return;
+      }
+
+      res.writeHead(404);
+      res.end();
+    });
+
+    const transport = new TriggerChatTransport({
+      task: "chat-task",
+      stream: "chat-stream",
+      accessToken: "pk_trigger",
+      baseURL: server.url,
+      runStore,
+      onError: function onError(error) {
+        errors.push(error);
+      },
+    });
+
+    (transport as any).fetchRunStream = async function fetchRunStream() {
+      throw new Error("stream subscribe failed root");
+    };
+
+    await expect(
+      transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "chat-stream-subscribe-error",
+        messageId: undefined,
+        messages: [],
+        abortSignal: undefined,
+      })
+    ).rejects.toThrowError("stream subscribe failed root");
+
+    expect(errors).toHaveLength(1);
+    expect(errors[0]).toMatchObject({
+      phase: "streamSubscribe",
+      chatId: "chat-stream-subscribe-error",
+      runId: "run_stream_subscribe_error",
+    });
+    expect(runStore.get("chat-stream-subscribe-error")).toBeUndefined();
+  });
+
+  it("keeps original stream subscription failure when onError callback also fails", async function () {
+    const runStore = new InMemoryTriggerChatRunStore();
+
+    const server = await startServer(function (req, res) {
+      if (req.method === "POST" && req.url === "/api/v1/tasks/chat-task/trigger") {
+        res.writeHead(200, {
+          "content-type": "application/json",
+          "x-trigger-jwt": "pk_stream_subscribe_onerror_failure",
+        });
+        res.end(JSON.stringify({ id: "run_stream_subscribe_onerror_failure" }));
+        return;
+      }
+
+      res.writeHead(404);
+      res.end();
+    });
+
+    const transport = new TriggerChatTransport({
+      task: "chat-task",
+      stream: "chat-stream",
+      accessToken: "pk_trigger",
+      baseURL: server.url,
+      runStore,
+      onError: async function onError() {
+        throw new Error("onError failed");
+      },
+    });
+
+    (transport as any).fetchRunStream = async function fetchRunStream() {
+      throw new Error("stream subscribe failed root");
+    };
+
+    await expect(
+      transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "chat-stream-subscribe-onerror-failure",
+        messageId: undefined,
+        messages: [],
+        abortSignal: undefined,
+      })
+    ).rejects.toThrowError("stream subscribe failed root");
+
+    expect(runStore.get("chat-stream-subscribe-onerror-failure")).toBeUndefined();
+  });
+
   it("supports creating transport with factory function", async function () {
     let observedRunId: string | undefined;
     let callbackCompleted = false;

packages/ai/src/chatTransport.ts

@@ -231,7 +231,21 @@ export class TriggerChatTransport<
       }
     }
 
-    const stream = await this.fetchRunStream(runState, options.abortSignal);
+    let stream: ReadableStream<SSEStreamPart<InferUIMessageChunk<UI_MESSAGE>>>;
+    try {
+      stream = await this.fetchRunStream(runState, options.abortSignal);
+    } catch (error) {
+      runState.isActive = false;
+      await this.runStore.set(runState);
+      await this.runStore.delete(runState.chatId);
+      await this.reportError({
+        phase: "streamSubscribe",
+        chatId: runState.chatId,
+        runId: runState.runId,
+        error: normalizeError(error),
+      });
+      throw error;
+    }
 
     return this.createTrackedStream(runState.chatId, stream);
   }

packages/ai/src/chatTransport.types.test.ts

@@ -112,6 +112,7 @@ function createTypedOnErrorCallback(): TriggerChatOnError {
       | "payloadMapper"
       | "triggerOptions"
       | "triggerTask"
+      | "streamSubscribe"
       | "onTriggeredRun"
       | "consumeTrackingStream"
       | "reconnect"

packages/ai/src/types.ts

@@ -74,6 +74,7 @@ export type TriggerChatTransportErrorPhase =
   | "payloadMapper"
   | "triggerOptions"
   | "triggerTask"
+  | "streamSubscribe"
   | "onTriggeredRun"
   | "consumeTrackingStream"
   | "reconnect";