feat(realtime): start-from-latest streams and a useSessionStream hook

.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/realtime-streams-from-latest.md

@@ -0,0 +1,19 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Subscribe to a realtime stream from its latest record instead of replaying the whole history. Pass `from: "latest"` to `useRealtimeStream`, `streams.read()`, or `fetchStream` to start at the current tail (the latest record, then live updates) instead of replaying (a live "last value" view), and `maxParts` to keep the accumulated `parts` array bounded. A reconnect or remount resumes from the last record it saw, so no records are missed and none are replayed. `from: "latest"` needs a server that supports it; older servers safely fall back to a full replay.
+
+`useRealtimeStream` also gains a `lastEventId` option and returns the `lastEventId` of the last part seen, so you can persist the cursor (for example across a page reload) and resume exactly where you left off. An `onParts` callback delivers each throttled batch of parts with their event ids.
+
+```tsx
+const { parts, lastEventId } = useRealtimeStream<Frame>(runId, "frames", {
+  from: "latest", // skip history, start at the current tail
+  maxParts: 1, // keep only the most recent frame
+  lastEventId: savedCursor, // resume from a persisted cursor
+  onParts: (batch) => save(batch.at(-1)?.id), // track the cursor
+  accessToken,
+});
+```

.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. It accumulates records with automatic resume from the last record you received, and supports `from: "latest"` (start at the current tail, only new records after you connect), `maxRecords` (keep a bounded number of records in memory), a `lastEventId` resume cursor, and an `onRecords` callback that delivers each throttled batch of records with their event ids.

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.ts

@@ -1,4 +1,5 @@
 import { json } from "@remix-run/server-runtime";
+import { STREAM_START_HEADER } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
@@ -185,12 +186,15 @@ const loader = createLoaderApiRoute(
     // turn's first chunk and the SSE closes before records land.
     const peekSettled = request.headers.get("X-Peek-Settled") === "1";
 
+    const startFrom =
+      request.headers.get(STREAM_START_HEADER)?.toLowerCase() === "latest" ? "latest" : undefined;
+
     return realtimeStream.streamResponseFromSessionStream(
       request,
       resource.addressingKey,
       params.io,
       getRequestAbortSignal(),
-      { lastEventId, timeoutInSeconds, peekSettled }
+      { lastEventId, timeoutInSeconds, peekSettled, startFrom }
     );
   }
 );

apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts

@@ -1,3 +1,4 @@
+import { STREAM_START_HEADER } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
@@ -59,6 +60,9 @@ export const loader = createLoaderApiRoute(
     // Get Last-Event-ID header for resuming from a specific position
     const lastEventId = request.headers.get("Last-Event-ID") || undefined;
 
+    const startFrom =
+      request.headers.get(STREAM_START_HEADER)?.toLowerCase() === "latest" ? "latest" : undefined;
+
     const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
     const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
 
@@ -88,6 +92,7 @@ export const loader = createLoaderApiRoute(
       {
         lastEventId,
         timeoutInSeconds,
+        startFrom,
       }
     );
   }

apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts

@@ -70,8 +70,7 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
 
     const stream = new ReadableStream<StreamChunk>({
       start: async (controller) => {
-        // Start from lastEventId if provided, otherwise from beginning
-        let lastId = options?.lastEventId ?? "0";
+        let lastId = options?.lastEventId ?? (options?.startFrom === "latest" ? "$" : "0");
         let retryCount = 0;
         const maxRetries = 3;
         let lastDataTime = Date.now();

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -527,12 +527,19 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
   ): Promise<Response> {
     const startSeq = this.parseLastEventId(options?.lastEventId);
 
-    this.logger.info(`S2 streaming records from stream`, { stream: s2Stream, startSeq });
+    const tailFromLatest = startSeq == null && options?.startFrom === "latest";
+
+    this.logger.info(`S2 streaming records from stream`, {
+      stream: s2Stream,
+      startSeq,
+      tailFromLatest,
+    });
 
     // Request SSE stream from S2 and return it directly
     const s2Response = await this.s2StreamRecords(s2Stream, {
-      seq_num: startSeq ?? 0,
-      clamp: true,
+      ...(tailFromLatest
+        ? { tail_offset: 1, clamp: true }
+        : { seq_num: startSeq ?? 0, clamp: true }),
       wait: options?.timeoutInSeconds ?? this.s2WaitSeconds, // S2 will keep the connection open and stream new records
       signal, // Pass abort signal so S2 connection is cleaned up when client disconnects
     });
@@ -672,6 +679,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     stream: string,
     opts: {
       seq_num?: number;
+      tail_offset?: number;
       clamp?: boolean;
       wait?: number;
       signal?: AbortSignal;
@@ -680,6 +688,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     // GET /v1/streams/{stream}/records with Accept: text/event-stream for SSE streaming
     const qs = new URLSearchParams();
     if (opts.seq_num != null) qs.set("seq_num", String(opts.seq_num));
+    if (opts.tail_offset != null) qs.set("tail_offset", String(opts.tail_offset));
     if (opts.clamp != null) qs.set("clamp", String(opts.clamp));
     if (opts.wait != null) qs.set("wait", String(opts.wait));
 

apps/webapp/app/services/realtime/types.ts

@@ -36,6 +36,13 @@ export interface StreamIngestor {
 export type StreamResponseOptions = {
   timeoutInSeconds?: number;
   lastEventId?: string;
+  /**
+   * Where a fresh subscription (no `lastEventId`) starts reading. `"latest"`
+   * starts at the current tail so the subscriber sees only records appended
+   * after it connects; `"beginning"` (the default when unset) replays history.
+   * Ignored when `lastEventId` is set.
+   */
+  startFrom?: "beginning" | "latest";
   /**
    * Session-stream-only. When `true`, the responder MAY peek the tail
    * of `.out` and short-circuit to `wait=0` + `X-Session-Settled: true`

apps/webapp/test/redisRealtimeStreams.test.ts

@@ -1521,4 +1521,146 @@ describe("RedisRealtimeStreams", () => {
       await redis.quit();
     }
   );
+
+  redisTest(
+    "startFrom 'latest' skips the backlog and delivers only new records",
+    { timeout: 30_000 },
+    async ({ redisOptions }) => {
+      const redis = new Redis(redisOptions);
+      const redisRealtimeStreams = new RedisRealtimeStreams({ redis: redisOptions });
+
+      const runId = "run_latest_test";
+      const streamId = "latest-stream";
+      const encoder = new TextEncoder();
+
+      const ingest = async (line: string) => {
+        const stream = new ReadableStream({
+          start(controller) {
+            controller.enqueue(encoder.encode(line + "\n"));
+            controller.close();
+          },
+        });
+        await redisRealtimeStreams.ingestData(stream, runId, streamId, "default");
+      };
+
+      await ingest("old-0");
+      await ingest("old-1");
+
+      const abortController = new AbortController();
+      const response = await redisRealtimeStreams.streamResponse(
+        new Request("http://localhost/test"),
+        runId,
+        streamId,
+        abortController.signal,
+        { startFrom: "latest", timeoutInSeconds: 10 }
+      );
+
+      const reader = response.body!.getReader();
+      const decoder = new TextDecoder();
+      const receivedData: string[] = [];
+
+      const readLoop = (async () => {
+        let done = false;
+        while (!done && receivedData.length < 1) {
+          const { value, done: streamDone } = await reader.read().catch(() => ({
+            value: undefined,
+            done: true,
+          }));
+          done = streamDone;
+          if (value) {
+            const events = decoder
+              .decode(value)
+              .split("\n\n")
+              .filter((event) => event.trim());
+            for (const event of events) {
+              for (const l of event.split("\n")) {
+                if (l.startsWith("data: ")) {
+                  const data = l.substring(6).trim();
+                  if (data) receivedData.push(data);
+                }
+              }
+            }
+          }
+        }
+      })();
+
+      await new Promise((resolve) => setTimeout(resolve, 500));
+      await ingest("new-0");
+
+      await readLoop;
+      abortController.abort();
+      reader.releaseLock();
+
+      expect(receivedData).toContain("new-0");
+      expect(receivedData).not.toContain("old-0");
+      expect(receivedData).not.toContain("old-1");
+
+      await redis.del(`stream:${runId}:${streamId}`);
+      await redis.quit();
+    }
+  );
+
+  redisTest(
+    "default start replays the backlog from the beginning",
+    { timeout: 30_000 },
+    async ({ redisOptions }) => {
+      const redis = new Redis(redisOptions);
+      const redisRealtimeStreams = new RedisRealtimeStreams({ redis: redisOptions });
+
+      const runId = "run_beginning_test";
+      const streamId = "beginning-stream";
+      const encoder = new TextEncoder();
+
+      const ingestStream = new ReadableStream({
+        start(controller) {
+          controller.enqueue(encoder.encode("old-0\n"));
+          controller.enqueue(encoder.encode("old-1\n"));
+          controller.close();
+        },
+      });
+      await redisRealtimeStreams.ingestData(ingestStream, runId, streamId, "default");
+
+      const abortController = new AbortController();
+      const response = await redisRealtimeStreams.streamResponse(
+        new Request("http://localhost/test"),
+        runId,
+        streamId,
+        abortController.signal,
+        { timeoutInSeconds: 10 }
+      );
+
+      const reader = response.body!.getReader();
+      const decoder = new TextDecoder();
+      const receivedData: string[] = [];
+
+      let done = false;
+      while (!done && receivedData.length < 2) {
+        const { value, done: streamDone } = await reader.read();
+        done = streamDone;
+        if (value) {
+          const events = decoder
+            .decode(value)
+            .split("\n\n")
+            .filter((event) => event.trim());
+          for (const event of events) {
+            for (const l of event.split("\n")) {
+              if (l.startsWith("data: ")) {
+                const data = l.substring(6).trim();
+                if (data) receivedData.push(data);
+              }
+            }
+          }
+        }
+      }
+
+      abortController.abort();
+      reader.releaseLock();
+
+      expect(receivedData).toContain("old-0");
+      expect(receivedData).toContain("old-1");
+
+      await redis.del(`stream:${runId}:${streamId}`);
+      await redis.quit();
+    }
+  );
 });

docs/docs.json

@@ -217,6 +217,7 @@
                   "realtime/react-hooks/triggering",
                   "realtime/react-hooks/subscribe",
                   "realtime/react-hooks/streams",
+                  "realtime/react-hooks/session-stream",
                   "realtime/react-hooks/swr",
                   "realtime/react-hooks/use-wait-token"
                 ]

docs/realtime/auth.mdx

@@ -135,6 +135,24 @@ When using non-root API keys (recommended), the expiration cannot be more than 3
 
 The format used for a time span is the same as the [jose package](https://github.com/panva/jose), which is a number followed by a unit. Valid units are: "sec", "secs", "second", "seconds", "s", "minute", "minutes", "min", "mins", "m", "hour", "hours", "hr", "hrs", "h", "day", "days", "d", "week", "weeks", "w", "year", "years", "yr", "yrs", and "y". It is not possible to specify months. 365.25 days is used as an alias for a year. If the string is suffixed with "ago", or prefixed with a "-", the resulting time span gets subtracted from the current unix timestamp. A "from now" suffix can also be used for readability when adding to the current unix timestamp.
 
+### Refreshing an expired token
+
+A realtime stream subscription can outlive its token. Pass a `refreshAccessToken` callback and a subscription rejected with a 401/403 re-mints once and reconnects, instead of failing. With no refresher, auth errors stay terminal. Mint the fresh token from your backend, where your secret key lives:
+
+```tsx
+import { useRealtimeStream } from "@trigger.dev/react-hooks";
+
+const { parts } = useRealtimeStream<{ url: string }>(runId, "frames", {
+  accessToken,
+  refreshAccessToken: async () => {
+    const res = await fetch("/api/realtime-token"); // your backend calls auth.createPublicToken
+    return (await res.json()).token;
+  },
+});
+```
+
+`refreshAccessToken` is available on every realtime hook, and on `useApiClient` and `TriggerAuthContext` so hooks under a provider share one refresher.
+
 ### Auto-generated tokens
 
 When you [trigger tasks](/triggering) from your backend, the `handle` received includes a `publicAccessToken` field. This token can be used to authenticate real-time requests in your frontend application.