feat(react-hooks,core): useSessionStream hook and stream access-token refresh

Folds two related changes into the realtime stream work.

useSessionStream reads one channel of a session's realtime stream (out by
default, or in), accumulating records with automatic resume from the last
record seen. It is read-only; useSession is reserved for two-way (read and
write) communication.

Realtime stream subscriptions can refresh an expired access token and
reconnect once, via an optional refreshAccessToken on the client configuration
and the React hooks. Behavior is unchanged when no refresher is supplied:
auth errors stay terminal.

Author: ericallam

.changeset/eighty-donkeys-shake.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/react-hooks": patch
+---
+
+Realtime stream subscriptions can now refresh an expired access token and reconnect, via a new optional `refreshAccessToken` option on the client configuration and the React hooks.

.changeset/spotty-pillows-visit.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/react-hooks": patch
+---
+
+Added a `useSessionStream` React hook for reading a session's output or input channel in realtime, with automatic resume from the last record you received.

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

@@ -110,6 +110,7 @@ import {
   zodfetchOffsetLimitPage,
 } from "./core.js";
 import { ApiConnectionError, ApiError, BatchNotSealedError } from "./errors.js";
+import { refreshAccessTokenOnce, type RefreshAccessTokenFn } from "./refreshAccessToken.js";
 import {
   type AnyRealtimeRun,
   type AnyRunShape,
@@ -196,6 +197,7 @@ export type {
   AnyRealtimeRun,
   AnyRunShape,
   ApiRequestOptions,
+  ControlEvent,
   RealtimeRun,
   RunShape,
   RunStreamCallback,
@@ -225,6 +227,7 @@ export class ApiClient {
   public readonly futureFlags: ApiClientFutureFlags;
   private readonly additionalHeaders?: Record<string, string>;
   private readonly defaultRequestOptions: ZodFetchOptions;
+  private readonly refreshAccessToken?: RefreshAccessTokenFn;
 
   constructor(
     baseUrl: string,
@@ -233,9 +236,11 @@ export class ApiClient {
     // x-trigger-branch header, and the server disambiguates by the token's env.
     previewBranch?: string,
     requestOptions: ApiRequestOptions = {},
-    futureFlags: ApiClientFutureFlags = {}
+    futureFlags: ApiClientFutureFlags = {},
+    refreshAccessToken?: RefreshAccessTokenFn
   ) {
     this.accessToken = accessToken;
+    this.refreshAccessToken = refreshAccessToken;
     this.baseUrl = baseUrl.replace(/\/$/, "");
     this.previewBranch = previewBranch;
     const { additionalHeaders, ...restRequestOptions } = requestOptions;
@@ -281,6 +286,32 @@ export class ApiClient {
     return this.#getHeaders(false);
   }
 
+  /**
+   * Header resolver handed to stream subscriptions so a connection rejected with
+   * a 401/403 can reconnect with a freshly minted token. `undefined` when no
+   * `refreshAccessToken` was configured, which keeps auth errors terminal.
+   */
+  #resolveStreamHeaders(): (() => Promise<Record<string, string>>) | undefined {
+    const refreshAccessToken = this.refreshAccessToken;
+    if (!refreshAccessToken) return undefined;
+
+    return async () => {
+      const accessToken = await refreshAccessTokenOnce(refreshAccessToken);
+      return this.#getHeaders(false, { Authorization: `Bearer ${accessToken}` });
+    };
+  }
+
+  /** As {@link ApiClient.#resolveStreamHeaders}, for the leaner realtime header set. */
+  #resolveRealtimeHeaders(): (() => Promise<Record<string, string>>) | undefined {
+    const refreshAccessToken = this.refreshAccessToken;
+    if (!refreshAccessToken) return undefined;
+
+    return async () => {
+      const accessToken = await refreshAccessTokenOnce(refreshAccessToken);
+      return { ...this.#getRealtimeHeaders(), Authorization: `Bearer ${accessToken}` };
+    };
+  }
+
   async getRunResult(
     runId: string,
     requestOptions?: ZodFetchOptions
@@ -1463,6 +1494,7 @@ export class ApiClient {
 
     const subscription = new SSEStreamSubscription(url, {
       headers: this.getHeaders(),
+      resolveHeaders: this.#resolveStreamHeaders(),
       signal: options?.signal,
       onComplete: options?.onComplete,
       onError: options?.onError,
@@ -1666,6 +1698,7 @@ export class ApiClient {
         closeOnComplete:
           typeof options?.closeOnComplete === "boolean" ? options.closeOnComplete : true,
         headers: this.#getRealtimeHeaders(),
+        resolveHeaders: this.#resolveRealtimeHeaders(),
         client: this,
         signal: options?.signal,
         onFetchError: options?.onFetchError,
@@ -1689,6 +1722,7 @@ export class ApiClient {
       {
         closeOnComplete: false,
         headers: this.#getRealtimeHeaders(),
+        resolveHeaders: this.#resolveRealtimeHeaders(),
         client: this,
         signal: options?.signal,
         onFetchError: options?.onFetchError,
@@ -1715,6 +1749,7 @@ export class ApiClient {
       {
         closeOnComplete: false,
         headers: this.#getRealtimeHeaders(),
+        resolveHeaders: this.#resolveRealtimeHeaders(),
         client: this,
         signal: options?.signal,
         onFetchError: options?.onFetchError,
@@ -1792,6 +1827,7 @@ export class ApiClient {
     const streamFactory = new SSEStreamSubscriptionFactory(options?.baseUrl ?? this.baseUrl, {
       headers: this.getHeaders(),
       signal: options?.signal,
+      resolveHeaders: this.#resolveStreamHeaders(),
     });
 
     const subscription = streamFactory.createSubscription(runId, streamKey, {

packages/core/src/v3/apiClient/refreshAccessToken.test.ts

@@ -0,0 +1,62 @@
+import { describe, expect, it } from "vitest";
+import { refreshAccessTokenOnce } from "./refreshAccessToken.js";
+
+describe("refreshAccessTokenOnce", () => {
+  it("shares one in-flight mint between concurrent callers", async () => {
+    let calls = 0;
+    let release: (token: string) => void = () => {};
+    const refresh = () => {
+      calls++;
+      return new Promise<string>((resolve) => {
+        release = resolve;
+      });
+    };
+
+    const results = Promise.all([
+      refreshAccessTokenOnce(refresh),
+      refreshAccessTokenOnce(refresh),
+      refreshAccessTokenOnce(refresh),
+    ]);
+    release("fresh");
+
+    expect(await results).toEqual(["fresh", "fresh", "fresh"]);
+    expect(calls).toBe(1);
+  });
+
+  it("does not share a mint between different refreshers", async () => {
+    const a = async () => "a";
+    const b = async () => "b";
+
+    expect(await Promise.all([refreshAccessTokenOnce(a), refreshAccessTokenOnce(b)])).toEqual([
+      "a",
+      "b",
+    ]);
+  });
+
+  it("mints again once the previous call has settled", async () => {
+    let calls = 0;
+    const refresh = async () => `token-${++calls}`;
+
+    expect(await refreshAccessTokenOnce(refresh)).toBe("token-1");
+    expect(await refreshAccessTokenOnce(refresh)).toBe("token-2");
+  });
+
+  it("rejects every concurrent caller and does not poison later calls", async () => {
+    let calls = 0;
+    const refresh = async () => {
+      calls++;
+      if (calls === 1) throw new Error("mint failed");
+      return "recovered";
+    };
+
+    const first = refreshAccessTokenOnce(refresh);
+    const second = refreshAccessTokenOnce(refresh);
+
+    await expect(first).rejects.toThrow("mint failed");
+    await expect(second).rejects.toThrow("mint failed");
+    expect(calls).toBe(1);
+
+    expect(await refreshAccessTokenOnce(refresh)).toBe("recovered");
+    expect(calls).toBe(2);
+  });
+});

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

@@ -0,0 +1,22 @@
+export type RefreshAccessTokenFn = () => Promise<string>;
+
+const pendingRefreshes = new WeakMap<RefreshAccessTokenFn, Promise<string>>();
+
+/**
+ * Call `refreshAccessToken`, deduping concurrent calls to the same function.
+ * Several subscriptions (or several React hooks sharing one refresher) can hit
+ * an expired token at once; they reuse the in-flight mint instead of firing one
+ * per caller. Keyed on the refresher itself so callers only share a mint when
+ * they share a token owner.
+ */
+export function refreshAccessTokenOnce(refreshAccessToken: RefreshAccessTokenFn): Promise<string> {
+  const pending = pendingRefreshes.get(refreshAccessToken);
+  if (pending) return pending;
+
+  const promise = refreshAccessToken().finally(() => {
+    pendingRefreshes.delete(refreshAccessToken);
+  });
+  pendingRefreshes.set(refreshAccessToken, promise);
+
+  return promise;
+}