Add run-scoped PAT renewal for chat transport

Author: ericallam

.changeset/chat-run-pat-renewal.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Add run-scoped PAT renewal for chat transport (`renewRunAccessToken`), fail fast on 401/403 for SSE without retry backoff, and export `isTriggerRealtimeAuthError` for auth-error detection.

packages/core/src/v3/apiClient/errors.ts

@@ -128,6 +128,18 @@ export class PermissionDeniedError extends ApiError {
   override readonly status: 403 = 403;
 }
 
+/**
+ * True when `error` is a 401/403 from the Trigger API (e.g. expired run-scoped PAT on realtime streams).
+ * Uses structural checks so it works even if multiple copies of `@trigger.dev/core` are bundled (subclass `instanceof` can fail).
+ */
+export function isTriggerRealtimeAuthError(error: unknown): boolean {
+  if (error === null || typeof error !== "object") {
+    return false;
+  }
+  const e = error as ApiError;
+  return e.name === "TriggerApiError" && (e.status === 401 || e.status === 403);
+}
+
 export class NotFoundError extends ApiError {
   override readonly status: 404 = 404;
 }

packages/core/src/v3/apiClient/runStream.ts

@@ -14,7 +14,7 @@ import {
   IOPacket,
   parsePacket,
 } from "../utils/ioSerialization.js";
-import { ApiError } from "./errors.js";
+import { ApiError, isTriggerRealtimeAuthError } from "./errors.js";
 import { ApiClient } from "./index.js";
 import { zodShapeStream } from "./stream.js";
 
@@ -344,6 +344,12 @@ export class SSEStreamSubscription implements StreamSubscription {
         return;
       }
 
+      if (isTriggerRealtimeAuthError(error)) {
+        this.options.onError?.(error as Error);
+        controller.error(error as Error);
+        return;
+      }
+
       // Retry on error
       await this.retryConnection(controller, error as Error);
     }

packages/core/test/runStream.test.ts

@@ -1,7 +1,8 @@
-import { describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import {
   RunSubscription,
   SSEStreamPart,
+  SSEStreamSubscription,
   StreamSubscription,
   StreamSubscriptionFactory,
 } from "../src/v3/apiClient/runStream.js";
@@ -470,6 +471,47 @@ describe("RunSubscription", () => {
   });
 });
 
+describe("SSEStreamSubscription", () => {
+  let originalFetch: typeof global.fetch;
+
+  beforeEach(() => {
+    originalFetch = global.fetch;
+  });
+
+  afterEach(() => {
+    global.fetch = originalFetch;
+    vi.restoreAllMocks();
+  });
+
+  it("does not retry the initial fetch on 401", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 401 }));
+    global.fetch = fetchMock;
+
+    const sub = new SSEStreamSubscription("https://api.test/realtime/v1/streams/run_x/chat", {
+      headers: { Authorization: "Bearer expired" },
+    });
+
+    const stream = await sub.subscribe();
+    const reader = stream.getReader();
+    await expect(reader.read()).rejects.toMatchObject({ status: 401 });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not retry the initial fetch on 403", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 403 }));
+    global.fetch = fetchMock;
+
+    const sub = new SSEStreamSubscription("https://api.test/realtime/v1/streams/run_x/chat", {
+      headers: { Authorization: "Bearer denied" },
+    });
+
+    const stream = await sub.subscribe();
+    const reader = stream.getReader();
+    await expect(reader.read()).rejects.toMatchObject({ status: 403 });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+});
+
 export async function convertAsyncIterableToArray<T>(iterable: AsyncIterable<T>): Promise<T[]> {
   const result: T[] = [];
   for await (const item of iterable) {

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

@@ -15,7 +15,7 @@
  * function Chat() {
  *   const transport = useTriggerChatTransport<typeof chat>({
  *     task: "ai-chat",
- *     accessToken: () => fetchToken(),
+ *     accessToken: ({ chatId }) => fetchToken(chatId),
  *   });
  *
  *   const { messages, sendMessage } = useChat({ transport });
@@ -74,7 +74,7 @@ export type { InferChatUIMessage };
  * function Chat() {
  *   const transport = useTriggerChatTransport<typeof chat>({
  *     task: "ai-chat",
- *     accessToken: () => fetchToken(),
+ *     accessToken: ({ chatId }) => fetchToken(chatId),
  *   });
  *
  *   const { messages, sendMessage } = useChat({ transport });
@@ -90,11 +90,15 @@ export function useTriggerChatTransport<TTask extends AnyTask = AnyTask>(
   }
 
   // Keep onSessionChange up to date without recreating the transport
-  const { onSessionChange } = options;
+  const { onSessionChange, renewRunAccessToken } = options;
   useEffect(() => {
     ref.current?.setOnSessionChange(onSessionChange);
   }, [onSessionChange]);
 
+  useEffect(() => {
+    ref.current?.setRenewRunAccessToken(renewRunAccessToken);
+  }, [renewRunAccessToken]);
+
   return ref.current;
 }