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

.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

@@ -213,6 +213,7 @@ export class SpanPresenter extends BasePresenter {
     const span = await this.#getSpan({
       eventStore,
       spanId,
+      runFriendlyId,
       traceId,
       environmentId: parentRun.runtimeEnvironmentId,
       projectId: parentRun.projectId,
@@ -640,6 +641,7 @@ export class SpanPresenter extends BasePresenter {
     eventRepository,
     traceId,
     spanId,
+    runFriendlyId,
     environmentId,
     projectId,
     createdAt,
@@ -648,6 +650,7 @@ export class SpanPresenter extends BasePresenter {
     eventRepository: IEventRepository;
     traceId: string;
     spanId: string;
+    runFriendlyId: string;
     environmentId: string;
     projectId: string;
     eventStore: TaskEventStoreTable;
@@ -840,6 +843,46 @@ 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: {
+              runId: runFriendlyId,
+              sessionId,
+              channel: channel.length > 0 ? channel : undefined,
+              io,
+              metadata,
+            },
+          },
+        };
+      }
       case "prompt": {
         const promptData = extractPromptSpanData(span.properties as Record<string, unknown>);
 

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

@@ -0,0 +1,145 @@
+import { json } from "@remix-run/server-runtime";
+import { tryCatch } from "@trigger.dev/core/utils";
+import { nanoid } from "nanoid";
+import { z } from "zod";
+import { logger } from "~/services/logger.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  SESSION_CHANNEL_NAME_REGEX,
+  sessionChannelResources,
+} from "~/services/realtime/sessionChannels.server";
+import {
+  canonicalSessionAddressingKey,
+  resolveSessionWithWriterFallback,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import {
+  claimSessionStreamPart,
+  releaseSessionStreamPart,
+} from "~/services/sessionStreamWaitpointCache.server";
+import { anyResource, createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { ServiceValidationError } from "~/v3/services/common.server";
+
+const ParamsSchema = z.object({
+  session: z.string(),
+  channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
+  io: z.enum(["out", "in"]),
+});
+
+const MAX_APPEND_BODY_BYTES = 1024 * 1024;
+
+const { action, loader } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+    method: "POST",
+    maxContentLength: MAX_APPEND_BODY_BYTES,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) =>
+      resolveSessionWithWriterFallback(auth.environment.id, params.session),
+    authorization: {
+      action: "write",
+      resource: (params, _s, _h, _b, session) => {
+        const ids = new Set<string>([params.session]);
+        if (session) {
+          ids.add(session.friendlyId);
+          if (session.externalId) ids.add(session.externalId);
+        }
+        return anyResource(sessionChannelResources(params.channel, ids));
+      },
+    },
+  },
+  async ({ request, params, authentication, resource: session }) => {
+    if (!session) {
+      return new Response("Session not found", { status: 404 });
+    }
+
+    if (session.closedAt) {
+      return json({ ok: false, error: "Cannot append to a closed session" }, { status: 400 });
+    }
+
+    if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
+      return json({ ok: false, error: "Cannot append to an expired session" }, { status: 400 });
+    }
+
+    if (params.io === "out" && authentication.type !== "PRIVATE") {
+      return json(
+        { ok: false, error: "Appending to the out channel requires secret key authentication" },
+        { status: 403 }
+      );
+    }
+
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session,
+    });
+
+    if (!(realtimeStream instanceof S2RealtimeStreams)) {
+      return json(
+        { ok: false, error: "Session channels require the S2 realtime backend" },
+        { status: 501 }
+      );
+    }
+
+    const addressingKey = canonicalSessionAddressingKey(session, params.session);
+    const claimKey = `${addressingKey}:channels:${params.channel}`;
+
+    const part = await request.text();
+
+    const clientPartId = request.headers.get("X-Part-Id");
+    const partId = clientPartId ?? nanoid(7);
+
+    const wonClaim = clientPartId
+      ? await claimSessionStreamPart(
+          authentication.environment.id,
+          claimKey,
+          params.io,
+          clientPartId
+        )
+      : true;
+
+    let appendSeq: number | undefined;
+    if (wonClaim) {
+      const [appendError, seq] = await tryCatch(
+        realtimeStream.appendPartToSessionStream(
+          part,
+          partId,
+          addressingKey,
+          params.io,
+          params.channel
+        )
+      );
+      appendSeq = seq ?? undefined;
+
+      if (appendError) {
+        if (clientPartId) {
+          await releaseSessionStreamPart(
+            authentication.environment.id,
+            claimKey,
+            params.io,
+            clientPartId
+          );
+        }
+        if (appendError instanceof ServiceValidationError) {
+          return json(
+            { ok: false, error: appendError.message },
+            { status: appendError.status ?? 422 }
+          );
+        }
+        logger.error("Failed to append to session channel stream", {
+          sessionId: session.id,
+          io: params.io,
+          channel: params.channel,
+          error: appendError,
+        });
+        return json(
+          { ok: false, error: "Something went wrong, please try again." },
+          { status: 500 }
+        );
+      }
+    }
+
+    return json({ ok: true, seq: appendSeq }, { status: 200 });
+  }
+);
+
+export { action, loader };

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

@@ -0,0 +1,81 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  SESSION_CHANNEL_NAME_REGEX,
+  sessionChannelResources,
+} from "~/services/realtime/sessionChannels.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(),
+  channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
+  io: z.enum(["out", "in"]),
+});
+
+const SearchSchema = z.object({
+  afterEventId: z.string().regex(/^\d+$/).optional(),
+});
+
+export const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    searchParams: SearchSchema,
+    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;
+      }
+      return {
+        row,
+        addressingKey: canonicalSessionAddressingKey(row, params.session),
+      };
+    },
+    authorization: {
+      action: "read",
+      resource: ({ row, addressingKey }, params) => {
+        const ids = new Set<string>([addressingKey]);
+        if (row) {
+          ids.add(row.friendlyId);
+          if (row.externalId) ids.add(row.externalId);
+        }
+        return anyResource(sessionChannelResources(params.channel, ids));
+      },
+    },
+  },
+  async ({ params, authentication, resource, searchParams }) => {
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: resource.row,
+      organization: resource.row ? null : authentication.environment.organization,
+    });
+
+    if (!(realtimeStream instanceof S2RealtimeStreams)) {
+      return new Response("Session channels require the S2 realtime backend", { status: 501 });
+    }
+
+    const afterSeqNum =
+      searchParams.afterEventId !== undefined ? Number(searchParams.afterEventId) : undefined;
+
+    const records = await realtimeStream.readSessionStreamRecords(
+      resource.addressingKey,
+      params.io,
+      afterSeqNum,
+      params.channel
+    );
+
+    return json({ records });
+  }
+);