feat(sdk,core,webapp,react-hooks): named side channels on a Session (#4815)

## Summary

Adds **named side channels** to a Session: durable, two-way realtime
streams that outlive a single run and are shared across every run of the
session. Today a Session has exactly one reserved `.in`/`.out` pair (the
chat transcript). This lets a session hold any number of *named*
channels alongside it, each its own `.in`/`.out` pair, so an agent can
stream out-of-band data (a feed of frames, telemetry, a control channel)
on a stream separate from the transcript while many clients read it
live.

The two properties a named channel adds over the reserved pair:

1. It is addressed by a name that outlives a run and is shared across
runs, not welded to the chat turn loop.
2. Writing its `.in` does **not** wake or trigger a run. A run observes
it by subscribing; an external client writes it without spawning
anything.

This is the generalization half of the Momentic ask (stream browser
screenshots from a `chat.agent` to the frontend on a channel separate
from the chat). It builds directly on the start-from-latest /
`useSessionStream` subscribe seam from #4811.

## Usage

Declare the channel's record types once and infer them on both sides:

```ts
// channels.ts (shared, client imports it type-only)
import { sessions } from "@trigger.dev/sdk";

export const screenshots = sessions.defineChannel<{ out: ScreenshotFrame; in: ViewportControl }>(
  "screenshots"
);
```

Open a channel from a session handle (`sessions.open(id)` returns one
for a known session id). Writing its `.out` is durable, cross-run, and
wakes nothing; a run observes its `.in` by tailing, without suspending:

```ts
import { sessions } from "@trigger.dev/sdk";
import { screenshots } from "./channels";

const channel = sessions.open(sessionId).channel(screenshots);
await channel.out.append(frame);             // frame: ScreenshotFrame (typed from the definition)
channel.in.on((control) => { /* ... */ });    // control: ViewportControl, tail, no suspend
```

Passing the definition types `.out.append` / `.in.on` on the producer
side; a bare name string also works, with records typed `unknown`.

An external client writes the `.in` without waking a run, and reads the
`.out` from React:

```ts
sessions.open(sessionId).channel("screenshots").in.send({ paused: true });

const { records } = useSessionStreamChannel<typeof screenshots>("screenshots", {
  sessionId,
  accessToken,
  io: "out",
  from: "latest",
  maxRecords: 1,
});
```

`session.channel(name)` returns the same `{ in, out }` handle shape as
the reserved pair, so `append` / `pipe` / `writer` / `read` /
`writeControl` / `trimTo` on `.out` and `send` / `on` / `once` / `peek`
on `.in` all carry over. Passing a name other than the declared one is a
type error; a bare-string call without the generic stays valid with
`records` typed `unknown`.

### With `chat.agent`

This is the motivating case: a `chat.agent` answers on the reserved
transcript as usual, and streams screenshot frames on a side channel in
parallel. `chat.channel(name)` opens a channel on the current run's own
Session, so there's no id to thread:

```ts
import { chat } from "@trigger.dev/sdk/ai";
import { streamText } from "ai";
import { screenshots } from "./channels";

export const browserAgent = chat.agent({
  id: "browser-agent",
  run: async ({ messages, signal }) => {
    const frames = chat.channel(screenshots);

    // client pause/resume arrives here without waking a turn
    frames.in.on((control: ViewportControl) => applyViewport(control));

    // frames stream on their own channel, not the chat transcript
    driveBrowser({ signal, onFrame: (frame) => frames.out.append(frame) });

    // the assistant reply still goes to the reserved transcript
    return streamText({ model: openai("gpt-4o"), messages, abortSignal: signal });
  },
});
```

`chat.channel(name)` is a shortcut for `chat.session().channel(name)`;
`chat.session()` returns the current run's full `SessionHandle` if you
need it.

The frontend renders the transcript with `useChat` as before, and the
screenshots with `useSessionStreamChannel<typeof
screenshots>("screenshots", { sessionId: chatId, io: "out", from:
"latest", maxRecords: 1 })`: a live view of the newest frame that
survives across turns (each turn is a new run), because the channel is
keyed on the session, not the run.

### From MCP

An MCP client can observe and write a session's channels with two tools,
built on the same apiClient surface as the hook and the dashboard
viewer:

- `read_session_channel` reads records from a channel (or the reserved
pair). It is a point-in-time drain with cursor pagination
(`afterEventId` / `nextCursor`, `maxRecords`); pass `timeoutInSeconds`
to wait for the next record when none exist yet.
- `write_session_channel` appends one record to a channel's `.in` (an
object or a raw string), so an agent can send control input without
waking a run. `.out` is producer-only, so it is not writable here.

### On the session page

The session detail page lists a session's channels (via an S2 prefix
list in the loader) and shows each as a tab beside `Rendered` and `Raw`.
Selecting a channel renders its records in the same table as the Raw
transcript view, sourced from that channel's `out` and `in` streams.

## How it works

**Addressing.** A channel is a stream name segment:
`sessions/{id}/channels/{name}/{io}`. The reserved pair keeps its
two-part `sessions/{id}/{io}` name for back-compat, and the `channels/`
segment means a user channel named `in`/`out` can never collide with it.
The channel dimension is threaded through the session stream manager
(keyed on `(session, channel, io)`, reserved = absent),
`subscribeToSessionStream`, the session apiClient methods, and the
`realtime.v1.sessions.$session.channels.$channel.$io.{ts,append,records}`
routes. The reserved-pair routes are untouched. The start-from-latest
tail path from #4811 is channel-agnostic, so `from: "latest"` and
`maxRecords` compose unchanged.

**No-wake.** The reserved `.in` append route ensures a run and drains
waitpoints so a chat turn advances. The channel `.in` append route
deliberately does neither: the record lands durably and a run picks it
up when it next subscribes, so writing a side channel can't spawn or
resume a run. A named channel's `.in` is therefore subscribe-only from
the run side (`.on` / `.once` / `.peek`); `.wait()` /
`waitWithIdleTimeout()` throw with a message pointing at the observe
methods.

**Auth.** Channel scope folds into the existing resource id
(`sessions:<key>:channels:<channel>`), so no RBAC grammar change. A
channel route authorizes both the channel-folded id and the bare session
id, which means a session-wide token grants every channel while a
channel-scoped token grants only its own. The per-io rule is preserved
per channel: writing `.out` requires secret-key auth so a browser can't
forge frames; `.in` is writable with the session token.

**Retention.** Channel streams are created on demand on first write and
inherit the org's stream retention (bounded age plus delete-on-empty
from the store's default config), the same as the reserved chat streams.
There is no per-channel control-plane call on the write path. Custom
per-channel retention is deferred until the stream store can set config
inline on the on-demand create, which avoids a control-plane round trip.

**Spans.** Channel writes carry `channel` and `io` attributes, an
accessory chip, and the session icon. Clicking a channel span in the
run's span inspector renders the channel's actual records with the same
viewer the run realtime streams use, rather than the raw properties
JSON.

## Verification

- **Unit (core):** the stream manager isolates channels: two channels on
the same `(session, io)` never cross buffers, and a named channel is
isolated from the reserved pair.
- **Full-stack e2e** against a real stack (webapp, stream store,
Postgres, real runs):
- a named `.out` record is readable back **after the triggering run has
gone terminal** (durable, cross-run);
- a channel `.in` append creates **no** run, while a reserved `.in`
append **does** wake one (the differential is the red/green);
- `from: "latest"` on a named channel delivers the live record and does
**not** replay the backlog from the start;
- the span inspector renders a channel span's records, and the MCP
read/write tools round-trip records on a real session;
  - an invalid channel name is rejected.

## Notes

- **Channel listing works on the self-hosted store too.** The stream
store's list operation is available on s2-lite, so the session page's
channel list is an OSS feature. It is a control-plane call made once per
session-page load (best-effort; a failure just hides the tabs), not on
the write path.
- **The ~1 MiB per-record cap is unchanged.** Large payloads (e.g. raw
screenshots) still need object-store pointers on the channel rather than
inline bytes; that's independent of this change.
- Docs ride this branch: the side channels guide, the
`useSessionStreamChannel` reference, and the MCP tools list are all
updated here.

## Screenshots

<img width="3444" height="1870" alt="CleanShot 2026-08-28 at 21 46
27@2x"
src="https://github.com/user-attachments/assets/c192aaee-b946-4824-87b7-ca057514d25e"
/>

Author: ericallam

.changeset/session-side-channels.md

@@ -0,0 +1,16 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Named side channels on a Session: durable, two-way realtime streams that outlive a single run and are shared across runs. Open a channel with `sessions.open(id).channel(name)` (or `chat.channel(name)` inside a `chat.agent`) to get an `.in`/`.out` pair addressed by name rather than the reserved default pair. Writing a side channel's `.in` does not wake or trigger a run, so a channel can carry out-of-band data (a stream of frames, a control signal) that many clients read while the agent produces it.
+
+```ts
+// Inside a chat.agent: stream frames on a named channel, wakes nothing
+const frames = chat.channel("screenshots");
+await frames.out.append(frame);
+frames.in.on((control) => { /* client control, no suspend */ });
+```
+
+Declare channel record types once with `sessions.defineChannel(...)` and infer them on both the producer and the consumer, including `useSessionStreamChannel` in React. Channels get a default retention that keeps them bounded, overridable per channel.

apps/webapp/app/components/runs/v3/RunIcon.tsx

@@ -36,6 +36,7 @@ import { PythonLogoIcon } from "~/assets/icons/PythonLogoIcon";
 import { TraceIcon } from "~/assets/icons/TraceIcon";
 import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";
 import { StreamsIcon } from "~/assets/icons/StreamsIcon";
+import { AIChatIcon } from "~/assets/icons/AIChatIcon";
 
 type TaskIconProps = {
   name: string | undefined;
@@ -169,6 +170,8 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
           className={cn(className, "text-text-dimmed group-hover/spannode:text-text-bright")}
         />
       );
+    case "sessions":
+      return <AIChatIcon className={cn(className, "text-sessions")} />;
     case "hero-sparkles":
       return (
         <SparklesIcon

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -840,6 +840,45 @@ export class SpanPresenter extends BasePresenter {
           },
         };
       }
+      case "session-stream": {
+        if (!span.entity.id) {
+          logger.error(`SpanPresenter: No session stream id`, {
+            spanId,
+            sessionStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const parts = span.entity.id.split(":");
+        const io = parts.at(-1);
+        const channel = parts.at(-2) ?? "";
+        const sessionId = parts.at(-3);
+
+        if (!sessionId || (io !== "out" && io !== "in")) {
+          logger.error(`SpanPresenter: Invalid session stream id`, {
+            spanId,
+            sessionStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const metadata = span.entity.metadata
+          ? (safeJsonParse(span.entity.metadata) as Record<string, unknown> | undefined)
+          : undefined;
+
+        return {
+          ...data,
+          entity: {
+            type: "session-stream" as const,
+            object: {
+              sessionId,
+              channel: channel.length > 0 ? channel : undefined,
+              io,
+              metadata,
+            },
+          },
+        };
+      }
       case "prompt": {
         const promptData = extractPromptSpanData(span.properties as Record<string, unknown>);
 

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.sessions.$sessionParam/route.tsx

@@ -57,6 +57,14 @@ import { redirectWithErrorMessage } from "~/models/message.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { SessionPresenter } from "~/presenters/v3/SessionPresenter.server";
+import { tryCatch } from "@trigger.dev/core/utils";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  canonicalSessionAddressingKey,
+  resolveSessionByIdOrExternalId,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { logger } from "~/services/logger.server";
 import {
   type StreamChunk,
   useRealtimeStream,
@@ -115,11 +123,34 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     throw new Response("Session not found", { status: 404 });
   }
 
-  return typedjson({ session, loadedAt: Date.now() });
+  let channels: string[] = [];
+  const streamSessionId = session.agentView?.sessionId;
+  if (streamSessionId) {
+    const [channelsError, listed] = await tryCatch(
+      (async () => {
+        const row = await resolveSessionByIdOrExternalId($replica, environment.id, streamSessionId);
+        if (!row) return [] as string[];
+        const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session: row });
+        if (!(realtimeStream instanceof S2RealtimeStreams)) return [] as string[];
+        const addressingKey = canonicalSessionAddressingKey(row, streamSessionId);
+        return realtimeStream.listSessionChannels(addressingKey);
+      })()
+    );
+    if (channelsError) {
+      logger.warn("Failed to list session channels", {
+        sessionId: streamSessionId,
+        error: channelsError,
+      });
+    } else {
+      channels = listed ?? [];
+    }
+  }
+
+  return typedjson({ session, channels, loadedAt: Date.now() });
 };
 
 export default function Page() {
-  const { session, loadedAt } = useTypedLoaderData<typeof loader>();
+  const { session, channels, loadedAt } = useTypedLoaderData<typeof loader>();
   const organization = useOrganization();
   const project = useProject();
   const environment = useEnvironment();
@@ -158,7 +189,7 @@ export default function Page() {
       <PageBody scrollable={false}>
         <ResizablePanelGroup orientation="horizontal" className="max-h-full">
           <ResizablePanel id="session-conversation" min={"300px"}>
-            <ConversationPane session={session} />
+            <ConversationPane session={session} channels={channels} />
           </ResizablePanel>
           <ResizableHandle id="session-handle" />
           <ResizablePanel
@@ -177,18 +208,49 @@ export default function Page() {
 
 type LoadedSession = ReturnType<typeof useTypedLoaderData<typeof loader>>["session"];
 
-function ConversationPane({ session }: { session: LoadedSession }) {
+function ConversationPane({ session, channels }: { session: LoadedSession; channels: string[] }) {
   const organization = useOrganization();
   const project = useProject();
   const environment = useEnvironment();
   const { value, replace } = useSearchParams();
   const isRaw = value("raw") === "1";
+  const channelParam = value("channel");
+  const activeChannel = channelParam && channels.includes(channelParam) ? channelParam : undefined;
 
   const sessionId = session.agentView.sessionId;
   const encodedSession = encodeURIComponent(sessionId);
   const sessionResourceBase = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/sessions/${encodedSession}/realtime/v1`;
 
-  const setView = useCallback((raw: boolean) => replace({ raw: raw ? "1" : undefined }), [replace]);
+  const setView = useCallback(
+    (raw: boolean) => replace({ raw: raw ? "1" : undefined, channel: undefined }),
+    [replace]
+  );
+  const selectChannel = useCallback(
+    (channel: string) => replace({ channel, raw: undefined }),
+    [replace]
+  );
+
+  const utilityBarProps = {
+    channels,
+    activeChannel,
+    onSelectChannel: selectChannel,
+  };
+
+  if (activeChannel) {
+    const channelBase = `${sessionResourceBase}/channels/${encodeURIComponent(activeChannel)}`;
+    return (
+      <div className="flex h-full max-h-full flex-col overflow-hidden bg-background-bright">
+        <RawConversationView
+          key={activeChannel}
+          inResourcePath={`${channelBase}/in`}
+          outResourcePath={`${channelBase}/out`}
+          isRaw={isRaw}
+          onChangeView={setView}
+          {...utilityBarProps}
+        />
+      </div>
+    );
+  }
 
   return (
     <div className="flex h-full max-h-full flex-col overflow-hidden bg-background-bright">
@@ -198,10 +260,11 @@ function ConversationPane({ session }: { session: LoadedSession }) {
           outResourcePath={`${sessionResourceBase}/out`}
           isRaw={isRaw}
           onChangeView={setView}
+          {...utilityBarProps}
         />
       ) : (
         <>
-          <ConversationUtilityBar isRaw={isRaw} onChangeView={setView} />
+          <ConversationUtilityBar isRaw={isRaw} onChangeView={setView} {...utilityBarProps} />
           <div className="min-w-0 flex-1 overflow-y-auto overflow-x-hidden px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-surface-control">
             <AgentView agentView={session.agentView} />
           </div>
@@ -214,29 +277,45 @@ function ConversationPane({ session }: { session: LoadedSession }) {
 function ConversationUtilityBar({
   isRaw,
   onChangeView,
+  channels = [],
+  activeChannel,
+  onSelectChannel,
   right,
 }: {
   isRaw: boolean;
   onChangeView: (raw: boolean) => void;
+  channels?: string[];
+  activeChannel?: string;
+  onSelectChannel?: (channel: string) => void;
   right?: React.ReactNode;
 }) {
   return (
     <div className="flex h-9 items-center justify-between gap-3 border-b border-grid-bright px-3">
       <TabContainer className="-mb-2">
         <TabButton
-          isActive={!isRaw}
+          isActive={!isRaw && !activeChannel}
           layoutId="conversation-view-mode"
           onClick={() => onChangeView(false)}
         >
           Rendered
         </TabButton>
         <TabButton
-          isActive={isRaw}
+          isActive={isRaw && !activeChannel}
           layoutId="conversation-view-mode"
           onClick={() => onChangeView(true)}
         >
           Raw
         </TabButton>
+        {channels.map((channel) => (
+          <TabButton
+            key={channel}
+            isActive={activeChannel === channel}
+            layoutId="conversation-view-mode"
+            onClick={() => onSelectChannel?.(channel)}
+          >
+            {channel}
+          </TabButton>
+        ))}
       </TabContainer>
       {right}
     </div>
@@ -266,11 +345,17 @@ function RawConversationView({
   outResourcePath,
   isRaw,
   onChangeView,
+  channels,
+  activeChannel,
+  onSelectChannel,
 }: {
   inResourcePath: string;
   outResourcePath: string;
   isRaw: boolean;
   onChangeView: (raw: boolean) => void;
+  channels?: string[];
+  activeChannel?: string;
+  onSelectChannel?: (channel: string) => void;
 }) {
   const {
     chunks: inChunks,
@@ -496,7 +581,14 @@ function RawConversationView({
 
   return (
     <>
-      <ConversationUtilityBar isRaw={isRaw} onChangeView={onChangeView} right={controls} />
+      <ConversationUtilityBar
+        isRaw={isRaw}
+        onChangeView={onChangeView}
+        channels={channels}
+        activeChannel={activeChannel}
+        onSelectChannel={onSelectChannel}
+        right={controls}
+      />
       <div className="flex min-h-0 flex-1 flex-col bg-background-deep">
         <div
           ref={scrollRef}

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

@@ -12,6 +12,10 @@ import { $replica, prisma, type PrismaClient } from "~/db.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { logger } from "~/services/logger.server";
 import { mintSessionToken } from "~/services/realtime/mintSessionToken.server";
+import {
+  isSafeSessionExternalId,
+  SESSION_CHANNEL_SCOPE_INFIX,
+} from "~/services/realtime/sessionChannels.server";
 import {
   ensureRunForSession,
   type SessionTriggerConfig,
@@ -168,6 +172,13 @@ const { action } = createActionApiRoute(
   },
   async ({ authentication, body }) => {
     try {
+      if (body.externalId && !isSafeSessionExternalId(body.externalId)) {
+        return json(
+          { error: `externalId cannot contain "${SESSION_CHANNEL_SCOPE_INFIX}"` },
+          { status: 422 }
+        );
+      }
+
       // Idempotent on (env, externalId): two concurrent POSTs converge to the same row, and
       // `triggerConfig` is refreshed on the cached path so a redeployed config reaches the next run.
       const { session, isCached } = await findOrCreateSession({