feat(sdk,core,webapp): direct-from-store reads for chat session .out streams

Chat session response streams (.out) read directly from the realtime stream
store instead of relaying every chunk through Trigger.dev, removing a network
hop from the streamed response. useTriggerChatTransport and AgentChat take the
direct path automatically for the active turn, obtain the read grant
themselves, and fall back to the relayed path whenever a grant cannot be
obtained or used (so failing to get one never breaks the chat).

Off by default when a custom baseURL routes .out through your own edge, and
gated server-side, so existing setups are unaffected. Set directStreamReads
to override.

Author: ericallam

.changeset/chat-session-direct-reads.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Chat session response streams (`.out`) can now be read directly from the realtime stream store instead of relaying every chunk through Trigger.dev's servers, removing a network hop from the streamed response. `useTriggerChatTransport` and `AgentChat` use the direct path automatically for the active streaming turn. The SDK obtains the read grant on its own (from the session-start response, or from a lightweight grant endpoint if your `startSession` doesn't forward it), refreshes it on turn-complete as it nears expiry, and transparently falls back to the relayed path whenever a grant can't be obtained or used (reconnects, hydrated sessions after a reload, or watch mode). Failing to obtain a grant never breaks the chat. No code changes are required to benefit.
+
+The direct read turns on by default only when you haven't customized `.out` routing. If you set a custom `baseURL`/`streamBaseURL` (e.g. fronting chat traffic with your own edge proxy), it stays off and `.out` follows your routing, so updating changes nothing for those setups. Use `directStreamReads: true`/`false` to override.

apps/webapp/app/env.server.ts

@@ -1740,6 +1740,20 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
     REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
     REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
+    // When "true", `POST /api/v1/sessions` hands the client a read-scoped S2
+    // token + endpoint so the browser reads the session's `.out` stream
+    // straight from S2 instead of proxying through the realtime host. Off
+    // keeps every read on the proxy. Requires real S2 (not s2-lite / a custom
+    // endpoint, which don't issue scoped tokens).
+    REALTIME_STREAMS_SESSIONS_DIRECT_READ_ENABLED: z.enum(["true", "false"]).default("false"),
+    // TTL for the browser-held read token (read-only, scoped to one stream).
+    // Long by default so a token rarely expires mid-session; the client also
+    // proactively refreshes it on turn-complete as it nears expiry, and falls
+    // back to the proxy if a refresh ever can't be obtained in time.
+    REALTIME_STREAMS_S2_READ_TOKEN_EXPIRATION_IN_MS: z.coerce
+      .number()
+      .int()
+      .default(60_000 * 60 * 24), // 24 hours
     // When "true", provision a dedicated S2 basin per org and stamp
     // `streamBasinName` on new rows. Off keeps everything on the single
     // basin defined by `REALTIME_STREAMS_S2_BASIN`.

apps/webapp/app/routes/api.v1.sessions.ts

@@ -4,15 +4,19 @@ import {
   type CreatedSessionResponseBody,
   ListSessionsQueryParams,
   type ListSessionsResponseBody,
+  type SessionDirectReadGrant,
   type SessionItem,
   type SessionStatus,
 } from "@trigger.dev/core/v3";
 import { SessionId } from "@trigger.dev/core/v3/isomorphic";
 import type { Prisma, Session } from "@trigger.dev/database";
 import { $replica, prisma, type PrismaClient } from "~/db.server";
+import { env } from "~/env.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { logger } from "~/services/logger.server";
 import { mintSessionToken } from "~/services/realtime/mintSessionToken.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import {
   ensureRunForSession,
   type SessionTriggerConfig,
@@ -257,6 +261,35 @@ const { action } = createActionApiRoute(
         addressingKey
       );
 
+      // When direct session reads are enabled, hand the client a read-scoped
+      // S2 token + endpoint so it reads `.out` straight from S2 instead of
+      // proxying through the realtime host. Best-effort: a failure here only
+      // means the client falls back to the proxy, so it must never fail the
+      // create.
+      let directReadOut: SessionDirectReadGrant | undefined;
+      if (env.REALTIME_STREAMS_SESSIONS_DIRECT_READ_ENABLED === "true") {
+        try {
+          const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+            session,
+          });
+          if (realtimeStream instanceof S2RealtimeStreams) {
+            // The `.out` stream is keyed by the canonical addressing key
+            // (externalId if set, else friendlyId) — the same key the worker
+            // writes with and the `.out` SSE route reads with. Mint the grant
+            // against that, NOT the friendlyId, or it points at an empty stream.
+            const grant = await realtimeStream.issueSessionOutReadGrant(addressingKey);
+            if (grant) {
+              directReadOut = { provider: "s2", ...grant };
+            }
+          }
+        } catch (error) {
+          logger.warn("Failed to mint session direct-read grant; client will use the proxy", {
+            sessionId: session.id,
+            error,
+          });
+        }
+      }
+
       const sessionItem: SessionItem = {
         ...serializeSession(session),
         triggerConfig: session.triggerConfig as unknown as SessionTriggerConfig,
@@ -268,6 +301,7 @@ const { action } = createActionApiRoute(
         runId: run.friendlyId,
         publicAccessToken,
         isCached,
+        ...(directReadOut ? { directReadOut } : {}),
       };
 
       return json<CreatedSessionResponseBody>(responseBody, {

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

@@ -0,0 +1,107 @@
+import { json } from "@remix-run/server-runtime";
+import { type SessionDirectReadGrant } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { env } from "~/env.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  canonicalSessionAddressingKey,
+  isSessionFriendlyIdForm,
+  resolveSessionByIdOrExternalId,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import {
+  anyResource,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+
+const ParamsSchema = z.object({
+  session: z.string(),
+});
+
+const SearchParamsSchema = z.object({
+  // `peek=1` is sent on the reconnect handshake: do the settled-peek
+  // server-side (next to S2) and return the verdict alongside the grant so the
+  // client doesn't make its own client→S2 peek round-trip. Absent on the
+  // active send path (no verdict needed there).
+  peek: z.string().optional(),
+});
+
+/**
+ * GET /realtime/v1/sessions/:session/out/grant
+ *
+ * Mint a fresh direct-read grant for a session's `.out` stream so the client
+ * can keep reading directly from S2 without the token expiring mid-session.
+ * Lightweight: authed by the session PAT the client already holds, no run
+ * trigger, no DB write. Returns `{ directReadOut: null }` when direct reads
+ * aren't available (flag off, or a backend that can't mint scoped tokens) so
+ * the client transparently stays on the proxied read.
+ */
+const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    searchParams: SearchParamsSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) => {
+      const row = await resolveSessionByIdOrExternalId(
+        $replica,
+        auth.environment.id,
+        params.session
+      );
+      if (!row && isSessionFriendlyIdForm(params.session)) {
+        return undefined; // 404 — opaque friendlyId must reference a real row
+      }
+      return {
+        row,
+        addressingKey: canonicalSessionAddressingKey(row, params.session),
+      };
+    },
+    authorization: {
+      action: "read",
+      resource: ({ row, addressingKey }) => {
+        const ids = new Set<string>([addressingKey]);
+        if (row) {
+          ids.add(row.friendlyId);
+          if (row.externalId) ids.add(row.externalId);
+        }
+        return anyResource([...ids].map((id) => ({ type: "sessions", id })));
+      },
+    },
+  },
+  async ({ authentication, resource, searchParams }) => {
+    let directReadOut: SessionDirectReadGrant | null = null;
+    // Reconnect settled-peek verdict, returned only when the client asks for
+    // it (`?peek=1`). `settled` undefined means "not peeked / unavailable" —
+    // the client then does its own peek.
+    let settled: boolean | undefined;
+    let tailSeq: number | undefined;
+
+    if (env.REALTIME_STREAMS_SESSIONS_DIRECT_READ_ENABLED === "true") {
+      const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+        session: resource.row,
+        organization: resource.row ? null : authentication.environment.organization,
+      });
+
+      if (realtimeStream instanceof S2RealtimeStreams) {
+        const grant = await realtimeStream.issueSessionOutReadGrant(resource.addressingKey);
+        if (grant) {
+          directReadOut = { provider: "s2", ...grant };
+
+          // Piggyback the settled-peek on the grant fetch the reconnecting
+          // client is already making — done here next to S2 (sub-ms) so the
+          // client can pick direct-drain vs direct-stream without its own peek.
+          if (searchParams.peek === "1") {
+            const verdict = await realtimeStream.peekSessionOutSettled(resource.addressingKey);
+            settled = verdict.settled;
+            tailSeq = verdict.tailSeq;
+          }
+        }
+      }
+    }
+
+    return json({ directReadOut, settled, tailSeq });
+  }
+);
+
+export { loader };