+
+ );
+ }
return (
@@ -198,10 +260,11 @@ function ConversationPane({ session }: { session: LoadedSession }) {
outResourcePath={`${sessionResourceBase}/out`}
isRaw={isRaw}
onChangeView={setView}
+ {...utilityBarProps}
/>
) : (
<>
-
onChangeView(false)}
>
Rendered
onChangeView(true)}
>
Raw
+ {channels.map((channel) => (
+ onSelectChannel?.(channel)}
+ >
+ {channel}
+
+ ))}
{right}
@@ -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 (
<>
-
{
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({
diff --git a/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.append.ts b/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.append.ts
new file mode 100644
index 00000000000..7372c644713
--- /dev/null
+++ b/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([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 };
diff --git a/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.records.ts b/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.records.ts
new file mode 100644
index 00000000000..148de201145
--- /dev/null
+++ b/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.records.ts
@@ -0,0 +1,76 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+ SESSION_CHANNEL_NAME_REGEX,
+ sessionChannelResources,
+} from "~/services/realtime/sessionChannels.server";
+import {
+ canonicalSessionAddressingKey,
+ isSessionFriendlyIdForm,
+ resolveSessionWithWriterFallback,
+} 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 resolveSessionWithWriterFallback(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([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 });
+ }
+);
diff --git a/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.ts b/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.ts
new file mode 100644
index 00000000000..0a740020a60
--- /dev/null
+++ b/apps/webapp/app/routes/realtime.v1.sessions.$session.channels.$channel.$io.ts
@@ -0,0 +1,157 @@
+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";
+import {
+ SESSION_CHANNEL_NAME_REGEX,
+ sessionChannelResources,
+} from "~/services/realtime/sessionChannels.server";
+import {
+ canonicalSessionAddressingKey,
+ isSessionFriendlyIdForm,
+ resolveSessionWithWriterFallback,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import {
+ anyResource,
+ createActionApiRoute,
+ 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 { action } = createActionApiRoute(
+ {
+ params: ParamsSchema,
+ method: "PUT",
+ allowJWT: true,
+ corsStrategy: "all",
+ authorization: {
+ action: "write",
+ resource: (params) => anyResource(sessionChannelResources(params.channel, [params.session])),
+ },
+ },
+ async ({ params, authentication }) => {
+ if (params.io === "out" && authentication.type !== "PRIVATE") {
+ return new Response("Initializing the out channel requires secret key authentication", {
+ status: 403,
+ });
+ }
+
+ const maybeSession = await resolveSessionWithWriterFallback(
+ authentication.environment.id,
+ params.session
+ );
+
+ if (!maybeSession && isSessionFriendlyIdForm(params.session)) {
+ return new Response("Session not found", { status: 404 });
+ }
+
+ if (maybeSession?.closedAt) {
+ return new Response("Cannot initialize a channel on a closed session", { status: 400 });
+ }
+
+ if (maybeSession?.expiresAt && maybeSession.expiresAt.getTime() < Date.now()) {
+ return new Response("Cannot initialize a channel on an expired session", { status: 400 });
+ }
+
+ const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+ session: maybeSession,
+ organization: maybeSession ? null : authentication.environment.organization,
+ });
+
+ if (!(realtimeStream instanceof S2RealtimeStreams)) {
+ return new Response("Session channels require the S2 realtime backend", { status: 501 });
+ }
+
+ const addressingKey = canonicalSessionAddressingKey(maybeSession, params.session);
+
+ const { responseHeaders } = await realtimeStream.initializeSessionStream(
+ addressingKey,
+ params.io,
+ params.channel
+ );
+
+ return json({ version: "v2" }, { status: 202, headers: responseHeaders });
+ }
+);
+
+const loader = createLoaderApiRoute(
+ {
+ params: ParamsSchema,
+ allowJWT: true,
+ corsStrategy: "all",
+ findResource: async (params, auth) => {
+ const row = await resolveSessionWithWriterFallback(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([addressingKey]);
+ if (row) {
+ ids.add(row.friendlyId);
+ if (row.externalId) ids.add(row.externalId);
+ }
+ return anyResource(sessionChannelResources(params.channel, ids));
+ },
+ },
+ },
+ async ({ params, request, authentication, resource }) => {
+ 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 });
+ }
+
+ if (request.method === "HEAD") {
+ return new Response(null, { status: 200, headers: { "X-Last-Chunk-Index": "0" } });
+ }
+
+ const lastEventId = request.headers.get("Last-Event-ID") ?? undefined;
+
+ const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds");
+ let timeoutInSeconds: number | undefined;
+ if (timeoutInSecondsRaw) {
+ const parsed = Number(timeoutInSecondsRaw);
+ if (!Number.isFinite(parsed) || !Number.isInteger(parsed)) {
+ return new Response("Invalid timeout seconds", { status: 400 });
+ }
+ if (parsed < 1) {
+ return new Response("Timeout seconds must be greater than 0", { status: 400 });
+ }
+ if (parsed > 600) {
+ return new Response("Timeout seconds must be less than 600", { status: 400 });
+ }
+ timeoutInSeconds = parsed;
+ }
+
+ const startFrom =
+ request.headers.get(STREAM_START_HEADER)?.toLowerCase() === "latest" ? "latest" : undefined;
+
+ return realtimeStream.streamResponseFromSessionStream(
+ request,
+ resource.addressingKey,
+ params.io,
+ getRequestAbortSignal(),
+ { lastEventId, timeoutInSeconds, startFrom },
+ params.channel
+ );
+ }
+);
+
+export { action, loader };
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx
index 1a59963656b..02571272fec 100644
--- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx
+++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx
@@ -1803,6 +1803,21 @@ function SpanEntity({ span }: { span: Span }) {
/>
);
}
+ case "session-stream": {
+ const { sessionId, channel, io } = span.entity.object;
+ const base = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/sessions/${encodeURIComponent(sessionId)}/realtime/v1`;
+ const resourcePath = channel
+ ? `${base}/channels/${encodeURIComponent(channel)}/${io}`
+ : `${base}/${io}`;
+ const displayName = channel ? `${channel}.${io}` : `${sessionId}.${io}`;
+ return (
+ }> {
return this.#initializeStreamByName(
- this.toSessionStreamName(friendlyId, io),
- `/sessions/${friendlyId}/${io}`
+ this.toSessionStreamName(friendlyId, io, channel),
+ this.#sessionStreamRelativeName(friendlyId, io, channel)
);
}
@@ -217,9 +224,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
part: string,
partId: string,
friendlyId: string,
- io: "out" | "in"
+ io: "out" | "in",
+ channel?: string
): Promise {
- return this.#appendPartByName(part, partId, this.toSessionStreamName(friendlyId, io));
+ return this.#appendPartByName(part, partId, this.toSessionStreamName(friendlyId, io, channel));
}
async #appendPartByName(part: string, partId: string, s2Stream: string): Promise {
@@ -259,9 +267,62 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
async readSessionStreamRecords(
friendlyId: string,
io: "out" | "in",
- afterSeqNum?: number
+ afterSeqNum?: number,
+ channel?: string
): Promise {
- return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io), afterSeqNum);
+ return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io, channel), afterSeqNum);
+ }
+
+ async listSessionChannels(friendlyId: string): Promise {
+ const prefix = `${this.streamPrefix}/sessions/${friendlyId}/channels/`;
+ const names = await this.#s2ListStreamNames(prefix);
+ const channels = new Set();
+ for (const name of names) {
+ const rest = name.slice(prefix.length);
+ const channel = rest.split("/")[0];
+ if (channel) channels.add(channel);
+ }
+ return [...channels];
+ }
+
+ async #s2ListStreamNames(prefix: string): Promise {
+ const names: string[] = [];
+ let startAfter: string | undefined;
+
+ for (let page = 0; page < 100; page++) {
+ const qs = new URLSearchParams();
+ qs.set("prefix", prefix);
+ if (startAfter) qs.set("start_after", startAfter);
+
+ const res = await fetch(`${this.baseUrl}/streams?${qs}`, {
+ method: "GET",
+ headers: {
+ Authorization: `Bearer ${this.token}`,
+ Accept: "application/json",
+ "S2-Basin": this.basin,
+ },
+ });
+
+ if (!res.ok) {
+ if (res.status === 404) return names;
+ const text = await res.text().catch(() => "");
+ throw new Error(`S2 listStreams failed: ${res.status} ${res.statusText} ${text}`);
+ }
+
+ const body = (await res.json()) as {
+ has_more?: boolean;
+ streams?: Array<{ name: string; deleted_at?: string | null }>;
+ };
+ const streams = body.streams ?? [];
+ for (const stream of streams) {
+ if (stream.deleted_at) continue;
+ names.push(stream.name);
+ }
+ if (!body.has_more || streams.length === 0) break;
+ startAfter = streams[streams.length - 1]!.name;
+ }
+
+ return names;
}
async #readRecordsByName(s2Stream: string, afterSeqNum?: number): Promise {
@@ -402,9 +463,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
friendlyId: string,
io: "out" | "in",
signal: AbortSignal,
- options?: StreamResponseOptions
+ options?: StreamResponseOptions,
+ channel?: string
): Promise {
- const s2Stream = this.toSessionStreamName(friendlyId, io);
+ const s2Stream = this.toSessionStreamName(friendlyId, io, channel);
let waitSeconds = options?.timeoutInSeconds ?? this.s2WaitSeconds;
let settled = false;
diff --git a/apps/webapp/app/services/realtime/sessionChannels.server.test.ts b/apps/webapp/app/services/realtime/sessionChannels.server.test.ts
new file mode 100644
index 00000000000..f5606964616
--- /dev/null
+++ b/apps/webapp/app/services/realtime/sessionChannels.server.test.ts
@@ -0,0 +1,35 @@
+import { describe, expect, it } from "vitest";
+import {
+ isSafeSessionExternalId,
+ SESSION_CHANNEL_SCOPE_INFIX,
+ sessionChannelResources,
+} from "./sessionChannels.server";
+
+describe("isSafeSessionExternalId", () => {
+ it("rejects an externalId that collides with the channel-scope fold", () => {
+ expect(isSafeSessionExternalId(`session_abc${SESSION_CHANNEL_SCOPE_INFIX}screencast`)).toBe(
+ false
+ );
+ expect(isSafeSessionExternalId(":channels:")).toBe(false);
+ expect(isSafeSessionExternalId("a:channels:b:channels:c")).toBe(false);
+ });
+
+ it("allows normal externalIds, including single colons that are not the fold infix", () => {
+ expect(isSafeSessionExternalId("chat-3c3a1756-a49a-4c78-891a-51f78596c984")).toBe(true);
+ expect(isSafeSessionExternalId("user:123")).toBe(true);
+ expect(isSafeSessionExternalId("org:abc:chat:1")).toBe(true);
+ expect(isSafeSessionExternalId("channels")).toBe(true);
+ expect(isSafeSessionExternalId("plain")).toBe(true);
+ });
+
+ it("keeps a channel-scoped token's folded id from equaling any allowed session's bare key", () => {
+ const channel = "screencast";
+ const foldedIds = sessionChannelResources(channel, ["session_abc"])
+ .map((r) => r.id)
+ .filter((id) => id.includes(SESSION_CHANNEL_SCOPE_INFIX));
+
+ for (const foldedId of foldedIds) {
+ expect(isSafeSessionExternalId(foldedId)).toBe(false);
+ }
+ });
+});
diff --git a/apps/webapp/app/services/realtime/sessionChannels.server.ts b/apps/webapp/app/services/realtime/sessionChannels.server.ts
new file mode 100644
index 00000000000..ee98ffce0a3
--- /dev/null
+++ b/apps/webapp/app/services/realtime/sessionChannels.server.ts
@@ -0,0 +1,39 @@
+import type { RbacResource } from "@trigger.dev/rbac";
+
+/**
+ * Channel names are both a URL path segment and an S2 stream-name segment, and
+ * they fold into the RBAC resource id as `${key}:channels:${channel}`, so a
+ * `/` would break addressing and a `:` would break scope parsing. Constrain to
+ * a safe, bounded alphabet.
+ */
+export const SESSION_CHANNEL_NAME_REGEX = /^[A-Za-z0-9._-]{1,128}$/;
+
+/**
+ * The infix the channel-scope fold uses in the RBAC resource id
+ * (`${key}:channels:${channel}`). A session externalId is used verbatim as a
+ * resource key, so an externalId containing this infix could equal a
+ * channel-scoped token's folded id and collide with it. Reject it at session
+ * creation so a bare session key can never look like a folded channel key.
+ */
+export const SESSION_CHANNEL_SCOPE_INFIX = ":channels:";
+
+export function isSafeSessionExternalId(externalId: string): boolean {
+ return !externalId.includes(SESSION_CHANNEL_SCOPE_INFIX);
+}
+
+/**
+ * Build the authorization resource set for a named channel. For each candidate
+ * session key (URL form, friendlyId, externalId) we authorize BOTH the
+ * channel-folded id (`${key}:channels:${channel}`, matched by a narrow
+ * channel-scoped token) and the bare session id (`${key}`, matched by a
+ * session-wide token so it grants every channel). RBAC matches ids exactly, so
+ * a channel token cannot match the bare session and vice versa.
+ */
+export function sessionChannelResources(channel: string, keys: Iterable): RbacResource[] {
+ const resources: RbacResource[] = [];
+ for (const key of keys) {
+ resources.push({ type: "sessions", id: `${key}:channels:${channel}` });
+ resources.push({ type: "sessions", id: key });
+ }
+ return resources;
+}
diff --git a/apps/webapp/vitest.config.ts b/apps/webapp/vitest.config.ts
index dabe517bf4f..3880e304573 100644
--- a/apps/webapp/vitest.config.ts
+++ b/apps/webapp/vitest.config.ts
@@ -17,6 +17,7 @@ export default defineConfig({
"app/v3/services/bulk/**/*.test.ts",
"app/runEngine/concerns/**/*.test.ts",
"app/runEngine/services/**/*.test.ts",
+ "app/services/realtime/**/*.test.ts",
"app/utils/**/*.test.ts",
"app/components/code/**/*.test.ts",
"app/components/runs/**/*.test.ts",
diff --git a/docs/ai-chat/side-channels.mdx b/docs/ai-chat/side-channels.mdx
new file mode 100644
index 00000000000..d46a8947e2a
--- /dev/null
+++ b/docs/ai-chat/side-channels.mdx
@@ -0,0 +1,172 @@
+---
+title: "Side channels"
+sidebarTitle: "Side channels"
+description: "Named, durable stream pairs on a Session, separate from the chat transcript. A side channel outlives a single run, is shared across runs, and its input does not wake a run."
+---
+
+**A side channel is a named `.in`/`.out` stream pair on a [Session](/ai-chat/sessions), separate from the reserved chat transcript.** Like the transcript it is durable and cross-run, but it is addressed by a name, and writing its `.in` does not wake or trigger a run.
+
+Side channels are a Session primitive, not a chat feature. Any Session can carry them: a `chat.agent`, a task-bound Session, or an external process holding your secret key. Use one to stream out-of-band data alongside (or instead of) a transcript: a feed of browser screenshots, progress telemetry, or a control channel the client writes to. Many clients can read the channel live while a run, or your backend, produces it.
+
+```mermaid
+flowchart LR
+ A["chat.agent run"] -- "frames" --> OUT([channel .out])
+ OUT --> C[Browser clients]
+ C -- "control (pause, viewport)" --> IN([channel .in])
+ IN -. "observed, no run wake" .-> A
+```
+
+## Define the channel once
+
+Declare the channel's record types in one shared module with `sessions.defineChannel`, then import it on both the producer and the consumer so the types line up.
+
+```ts /trigger/channels.ts
+import { sessions } from "@trigger.dev/sdk";
+
+export type ScreenshotFrame = { url: string; step: number };
+export type ViewportControl = { paused: boolean };
+
+export const screenshots = sessions.defineChannel<{
+ out: ScreenshotFrame;
+ in: ViewportControl;
+}>("screenshots");
+```
+
+## Produce on `.out` from a chat.agent
+
+Inside a `chat.agent` run, `chat.channel(...)` opens a channel on the current run's Session. Writing `.out` is durable and cross-run, and wakes nothing. The client control arrives on `.in.on(...)` without waking a run:
+
+```ts /trigger/browser-agent.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);
+
+ frames.in.on((control) => setPaused(control.paused)); // control: ViewportControl
+
+ driveBrowser({
+ signal,
+ onFrame: (frame) => frames.out.append(frame), // frame: ScreenshotFrame
+ });
+
+ return streamText({ model, messages, abortSignal: signal }); // transcript, as usual
+ },
+});
+```
+
+
+ A side channel's `.in` is subscribe-only from the run's side (`.on` / `.once` / `.peek`). `.wait()`
+ is not supported on a named channel, because a side channel never suspends or wakes a run.
+
+
+## From a task or your backend
+
+Nothing here needs a `chat.agent`. Open a channel on any Session by id with `sessions.open(sessionId).channel(...)`; the handle exposes the same `.out` (`append` / `pipe` / `writer`) and `.in` (`send` / `on` / `once` / `peek`) surface as the reserved pair. Create the Session with [`sessions.start`](/ai-chat/sessions) bound to any task, then produce from that task's run:
+
+```ts /trigger/render-frames.ts
+import { sessions, task } from "@trigger.dev/sdk";
+import { screenshots } from "./channels";
+
+export const renderFrames = task({
+ id: "render-frames",
+ run: async (payload: { sessionId: string; steps: number }) => {
+ const frames = sessions.open(payload.sessionId).channel(screenshots);
+ for (let step = 1; step <= payload.steps; step++) {
+ frames.in.on((control) => setPaused(control.paused));
+ await frames.out.append({ url: await renderStep(step), step });
+ }
+ },
+});
+```
+
+Or produce from your own backend, which holds the secret key that `.out` writes require:
+
+```ts Your backend code
+import { sessions } from "@trigger.dev/sdk";
+import { screenshots } from "./trigger/channels";
+
+await sessions.open(sessionId).channel(screenshots).out.append({ url, step });
+```
+
+Either way the client reads the channel the same way, below.
+
+## Read `.out` in React
+
+`useSessionStreamChannel` reads one side of a channel and updates a `records` array. Pass the channel definition as the type argument so `records` is typed from it. `from: "latest"` with `maxRecords: 1` gives a live "latest frame" view with bounded memory:
+
+```tsx app/components/Screencast.tsx
+"use client";
+
+import { useSessionStreamChannel } from "@trigger.dev/react-hooks";
+import type { screenshots } from "../trigger/channels";
+
+export function Screencast({ sessionId, accessToken }: { sessionId: string; accessToken: string }) {
+ const { records } = useSessionStreamChannel("screenshots", {
+ sessionId,
+ accessToken,
+ io: "out",
+ from: "latest",
+ maxRecords: 1,
+ });
+
+ const latest = records[0]; // ScreenshotFrame | undefined
+ return latest ? 
:
+ Records are capped at ~1 MiB each. Stream a pointer, not bytes: write large payloads (a screenshot
+ PNG) to object storage and put the URL on the channel. A base64 image inflates ~33% and will exceed
+ the cap. Pointers also keep the channel small.
+
+
+## Auth
+
+A side channel is covered by the session's public access token: a token scoped to `read:sessions:{id}` / `write:sessions:{id}` grants every channel of that session. Mint a narrower token scoped to a single channel with `read:sessions:{id}:channels:{name}`. Writing a channel's `.out` requires secret-key auth (only the agent run), so a browser cannot forge frames; `.in` is writable with the session token. See [Realtime auth](/realtime/auth).
+
+### Scope tokens to the channel, not the whole session
+
+Two properties of the session token are worth designing around when a browser only needs one channel:
+
+- **A session-wide token grants every channel, including ones added later.** `read:sessions:{id}` reads the reserved chat transcript and all named channels. If a client should see only the screencast frames and not the chat, give it `read:sessions:{id}:channels:screencast` instead. The channel-scoped token reads only that channel: it cannot read another channel or the reserved transcript.
+- **A session write token can write the reserved `.in` too, not just a channel's.** `write:sessions:{id}` can send a chat message on the reserved `.in`, so a client meant only to send control input on one channel should hold `write:sessions:{id}:channels:{name}`, which confines it to that channel's `.in`.
+
+```ts Mint a channel-scoped token (your backend)
+import { auth } from "@trigger.dev/sdk";
+
+const token = await auth.createPublicToken({
+ scopes: { read: { sessions: `${sessionId}:channels:screencast` } },
+});
+```
+
+
+ A session's `externalId` cannot contain `:channels:`, since that is the delimiter the channel scope
+ uses. `sessions.start` rejects it. Any other string, including single colons, is fine.
+
+
+## Next steps
+
+
+
+ The durable, cross-run primitive side channels are built on.
+
+
+ The `useSessionStream` hook `useSessionStreamChannel` mirrors.
+
+
diff --git a/docs/docs.json b/docs/docs.json
index 669bc1ebc9d..d7524a9664b 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -97,6 +97,7 @@
"ai-chat/frontend",
"ai-chat/server-chat",
"ai-chat/sessions",
+ "ai-chat/side-channels",
"ai-chat/chat-local",
"ai-chat/types",
"ai-chat/custom-agents",
diff --git a/docs/mcp-tools.mdx b/docs/mcp-tools.mdx
index d35a733b091..6f95b3c0965 100644
--- a/docs/mcp-tools.mdx
+++ b/docs/mcp-tools.mdx
@@ -271,3 +271,40 @@ Close an agent chat conversation. The agent exits its loop gracefully. Without t
The `start_agent_chat`, `send_agent_message`, and `close_agent_chat` tools are write operations and are not available in readonly mode.
+
+## Session Channel Tools
+
+Read and write a session's realtime streams: a named [side channel](/ai-chat/side-channels) or the reserved chat transcript pair. Use these to observe an agent's out-of-band output (a screencast, telemetry) or to send it control input.
+
+### read_session_channel
+
+Read records from a session's realtime stream. By default it returns the records that exist right now after an optional cursor and closes, so it is a point-in-time drain, not a live subscription. Set `timeoutInSeconds` to wait for the next record when none exist yet.
+
+**Parameters:**
+- `sessionId` (required): the session id (`session_*`) or the externalId it was created with
+- `channel` (optional): the named side channel to read. Omit to read the reserved chat transcript pair
+- `io` (optional, default: `out`): which side to read, `out` (producer feed) or `in` (client input)
+- `afterEventId` (optional): cursor. Only return records after this event id. Use the `nextCursor` from a prior read to page forward
+- `maxRecords` (optional, default: `100`): maximum records to return
+- `timeoutInSeconds` (optional): wait up to this many seconds for at least one record when none exist yet
+
+**Example usage:**
+- `"Read the latest frames on the screencast channel for this session"`
+- `"Wait for the next control message on the session's status channel"`
+
+### write_session_channel
+
+Append one record to a named side channel's `in` stream. Sends control input to a running agent (e.g. a pause command) without waking or triggering a run. The reserved transcript and a channel's `out` side are not writable here; `out` is producer-only.
+
+**Parameters:**
+- `sessionId` (required): the session id or externalId
+- `channel` (required): the named side channel to write to
+- `value` (required): the record to append. Pass an object for a structured record (e.g. `{ paused: true }`) or a string for a raw one
+
+**Example usage:**
+- `"Pause the screencast on this session"`
+- `"Send { paused: true } to the viewport channel"`
+
+
+ `write_session_channel` is a write operation and is not available in readonly mode.
+
diff --git a/docs/realtime/react-hooks/session-stream.mdx b/docs/realtime/react-hooks/session-stream.mdx
index 1d08f5f1300..c60e872d2f3 100644
--- a/docs/realtime/react-hooks/session-stream.mdx
+++ b/docs/realtime/react-hooks/session-stream.mdx
@@ -107,3 +107,20 @@ const { records, lastControl } = useSessionStream(sessionId, {
```
For an expiring token on a long-lived subscription, pass `refreshAccessToken` (see [Realtime auth](/realtime/auth)). To read a session channel outside React, use [`session.out.read()`](/ai-chat/sessions).
+
+## Named side channels
+
+`useSessionStream` reads a session's reserved channel. To read a [named side channel](/ai-chat/side-channels) — a durable, cross-run stream separate from the chat transcript — use `useSessionStreamChannel`. It takes the channel name as its first argument and has the same options and return shape, plus a channel-definition type argument that types `records`:
+
+```tsx
+import { useSessionStreamChannel } from "@trigger.dev/react-hooks";
+import type { screenshots } from "../trigger/channels";
+
+const { records } = useSessionStreamChannel("screenshots", {
+ sessionId,
+ accessToken,
+ io: "out",
+ from: "latest",
+ maxRecords: 1,
+});
+```
diff --git a/packages/cli-v3/src/mcp/config.ts b/packages/cli-v3/src/mcp/config.ts
index 227b0c5506e..a1e3f194b7c 100644
--- a/packages/cli-v3/src/mcp/config.ts
+++ b/packages/cli-v3/src/mcp/config.ts
@@ -241,4 +241,16 @@ export const toolsMetadata = {
description:
"Close an agent chat conversation. The agent exits its loop gracefully. Without this, the agent will close on its own when its idle timeout expires.",
},
+ read_session_channel: {
+ name: "read_session_channel",
+ title: "Read Session Channel",
+ description:
+ "Read records from a session's realtime stream: a named side channel (pass `channel`) or the reserved chat transcript pair (omit `channel`). By default returns whatever records exist after the optional cursor and closes (a point-in-time drain). Set `timeoutInSeconds` to wait for the next record when none exist yet. Read `out` for the producer's feed (e.g. a screencast) or `in` for what clients have sent. Use the returned nextCursor as `afterEventId` to page forward.",
+ },
+ write_session_channel: {
+ name: "write_session_channel",
+ title: "Write Session Channel",
+ description:
+ "Append one record to a named side channel's `in` stream on a session. Use this to send control input to a running agent (e.g. a pause/viewport command) without waking or triggering a run. Requires a `channel` name; the reserved transcript and the `out` side are not writable here (`out` is producer-only). Pass `value` as an object for a structured record or a string for a raw one.",
+ },
};
diff --git a/packages/cli-v3/src/mcp/tools.ts b/packages/cli-v3/src/mcp/tools.ts
index 8fa9eabf6f8..080dfff774e 100644
--- a/packages/cli-v3/src/mcp/tools.ts
+++ b/packages/cli-v3/src/mcp/tools.ts
@@ -32,6 +32,7 @@ import {
} from "./tools/prompts.js";
import { listAgentsTool } from "./tools/agents.js";
import { startAgentChatTool, sendAgentMessageTool, closeAgentChatTool } from "./tools/agentChat.js";
+import { readSessionChannelTool, writeSessionChannelTool } from "./tools/sessionChannels.js";
import { respondWithError } from "./utils.js";
/** Tool names that perform write/mutating operations. */
@@ -49,6 +50,7 @@ const WRITE_TOOLS = new Set([
startAgentChatTool.name,
sendAgentMessageTool.name,
closeAgentChatTool.name,
+ writeSessionChannelTool.name,
]);
export function registerTools(context: McpContext) {
@@ -90,6 +92,8 @@ export function registerTools(context: McpContext) {
startAgentChatTool,
sendAgentMessageTool,
closeAgentChatTool,
+ readSessionChannelTool,
+ writeSessionChannelTool,
getReportTool,
];
diff --git a/packages/cli-v3/src/mcp/tools/sessionChannels.ts b/packages/cli-v3/src/mcp/tools/sessionChannels.ts
new file mode 100644
index 00000000000..7254461f74e
--- /dev/null
+++ b/packages/cli-v3/src/mcp/tools/sessionChannels.ts
@@ -0,0 +1,177 @@
+import { z } from "zod";
+import { toolsMetadata } from "../config.js";
+import { CommonProjectsInput } from "../schemas.js";
+import { respondWithError, toolHandler } from "../utils.js";
+
+const SESSION_CHANNEL_NAME_REGEX = /^[A-Za-z0-9._-]{1,128}$/;
+
+const ReadSessionChannelInput = CommonProjectsInput.extend({
+ sessionId: z
+ .string()
+ .describe("The session id (session_* friendlyId) or the externalId it was created with."),
+ channel: z
+ .string()
+ .describe(
+ "The named side channel to read. Omit to read the session's reserved chat transcript pair."
+ )
+ .optional(),
+ io: z
+ .enum(["out", "in"])
+ .describe("Which side to read: `out` (producer feed) or `in` (client input).")
+ .default("out"),
+ afterEventId: z
+ .string()
+ .describe(
+ "Cursor: only return records after this event id. Use the nextCursor from a prior read."
+ )
+ .optional(),
+ maxRecords: z
+ .number()
+ .int()
+ .positive()
+ .max(500)
+ .describe("Maximum records to return (default 100).")
+ .default(100),
+ timeoutInSeconds: z
+ .number()
+ .int()
+ .positive()
+ .max(60)
+ .describe(
+ "Wait up to this many seconds for at least one record when none exist yet (a bounded tail). Omit for an immediate point-in-time read."
+ )
+ .optional(),
+});
+
+export const readSessionChannelTool = {
+ name: toolsMetadata.read_session_channel.name,
+ title: toolsMetadata.read_session_channel.title,
+ description: toolsMetadata.read_session_channel.description,
+ inputSchema: ReadSessionChannelInput.shape,
+ handler: toolHandler(ReadSessionChannelInput.shape, async (input, { ctx }) => {
+ ctx.logger?.log("calling read_session_channel", { input });
+
+ if (ctx.options.devOnly && input.environment !== "dev") {
+ return respondWithError(`This MCP server is only available for the dev environment.`);
+ }
+
+ if (input.channel !== undefined && !SESSION_CHANNEL_NAME_REGEX.test(input.channel)) {
+ return respondWithError(
+ `Invalid channel name "${input.channel}": use 1-128 chars from [A-Za-z0-9._-].`
+ );
+ }
+
+ const projectRef = await ctx.getProjectRef({
+ projectRef: input.projectRef,
+ cwd: input.configPath,
+ });
+
+ const apiClient = await ctx.getApiClient({
+ projectRef,
+ environment: input.environment,
+ scopes: ["read:sessions"],
+ branch: input.branch,
+ });
+
+ const drain = () =>
+ apiClient.readSessionStreamRecords(input.sessionId, input.io, {
+ channel: input.channel,
+ afterEventId: input.afterEventId,
+ });
+
+ let { records } = await drain();
+
+ if (records.length === 0 && input.timeoutInSeconds !== undefined) {
+ const deadline = Date.now() + input.timeoutInSeconds * 1000;
+ while (records.length === 0 && Date.now() < deadline) {
+ await new Promise((resolve) => setTimeout(resolve, 750));
+ ({ records } = await drain());
+ }
+ }
+
+ const limited = records.slice(0, input.maxRecords);
+ const hasMore = records.length > limited.length;
+ const nextCursor = limited.at(-1)?.seqNum;
+
+ const label = input.channel ? `channel "${input.channel}"` : "reserved pair";
+ const header = `Session ${input.sessionId} ${label} .${input.io}: ${limited.length} record${
+ limited.length === 1 ? "" : "s"
+ }${hasMore ? ` (more available)` : ""}`;
+
+ const lines = limited.map((record) => {
+ const data = typeof record.data === "string" ? record.data : JSON.stringify(record.data);
+ return `#${record.seqNum} ${data}`;
+ });
+
+ const footer =
+ nextCursor !== undefined && hasMore
+ ? `\n\nMore records available. Read again with afterEventId "${nextCursor}" to continue.`
+ : "";
+
+ return {
+ content: [
+ {
+ type: "text",
+ text: [header, "", ...lines].join("\n") + footer,
+ },
+ ],
+ };
+ }),
+};
+
+const WriteSessionChannelInput = CommonProjectsInput.extend({
+ sessionId: z
+ .string()
+ .describe("The session id (session_* friendlyId) or the externalId it was created with."),
+ channel: z.string().describe("The named side channel to write to."),
+ value: z
+ .union([z.string(), z.record(z.unknown())])
+ .describe(
+ "The record to append to the channel's `in` stream. Pass an object for a structured record (e.g. { paused: true }) or a string for a raw record."
+ ),
+});
+
+export const writeSessionChannelTool = {
+ name: toolsMetadata.write_session_channel.name,
+ title: toolsMetadata.write_session_channel.title,
+ description: toolsMetadata.write_session_channel.description,
+ inputSchema: WriteSessionChannelInput.shape,
+ handler: toolHandler(WriteSessionChannelInput.shape, async (input, { ctx }) => {
+ ctx.logger?.log("calling write_session_channel", { input });
+
+ if (ctx.options.devOnly && input.environment !== "dev") {
+ return respondWithError(`This MCP server is only available for the dev environment.`);
+ }
+
+ if (!SESSION_CHANNEL_NAME_REGEX.test(input.channel)) {
+ return respondWithError(
+ `Invalid channel name "${input.channel}": use 1-128 chars from [A-Za-z0-9._-].`
+ );
+ }
+
+ const projectRef = await ctx.getProjectRef({
+ projectRef: input.projectRef,
+ cwd: input.configPath,
+ });
+
+ const apiClient = await ctx.getApiClient({
+ projectRef,
+ environment: input.environment,
+ scopes: ["write:sessions"],
+ branch: input.branch,
+ });
+
+ const body = typeof input.value === "string" ? input.value : JSON.stringify(input.value);
+
+ await apiClient.appendToSessionStream(input.sessionId, "in", body, undefined, input.channel);
+
+ return {
+ content: [
+ {
+ type: "text",
+ text: `Wrote 1 record to session ${input.sessionId} channel "${input.channel}" .in. This does not wake or trigger a run.`,
+ },
+ ],
+ };
+ }),
+};
diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts
index c270f86ea62..585aeaed6c2 100644
--- a/packages/core/src/v3/apiClient/index.ts
+++ b/packages/core/src/v3/apiClient/index.ts
@@ -1387,15 +1387,18 @@ export class ApiClient {
async initializeSessionStream(
sessionIdOrExternalId: string,
io: "out" | "in",
- requestOptions?: ZodFetchOptions
+ requestOptions?: ZodFetchOptions,
+ channel?: string
) {
// The server returns S2 credentials in response headers alongside a tiny
// JSON body with the realtime version. Follow the same shape as
// `createStream` so downstream clients can feed them into
// `StreamsWriterV2`.
+ const base = `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}`;
+ const url = channel ? `${base}/channels/${encodeURIComponent(channel)}/${io}` : `${base}/${io}`;
return zodfetch(
CreateStreamResponseBody,
- `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}`,
+ url,
{
method: "PUT",
headers: this.#getHeaders(false),
@@ -1413,16 +1416,21 @@ export class ApiClient {
sessionIdOrExternalId: string,
io: "out" | "in",
part: TBody,
- requestOptions?: ZodFetchOptions
+ requestOptions?: ZodFetchOptions,
+ channel?: string
) {
// Generated once per logical append, outside zodfetch, so its internal
// retries reuse the same part id and the server-side dedupe collapses a
// retried POST whose first attempt actually committed. Full-length nanoid
// (~126 bits) to match the browser transport's randomUUID entropy.
const partId = nanoid();
+ const base = `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}`;
+ const appendUrl = channel
+ ? `${base}/channels/${encodeURIComponent(channel)}/${io}/append`
+ : `${base}/${io}/append`;
return zodfetch(
AppendToStreamResponseBody,
- `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}/append`,
+ appendUrl,
{
method: "POST",
headers: { ...this.#getHeaders(false), "X-Part-Id": partId },
@@ -1446,15 +1454,19 @@ export class ApiClient {
async readSessionStreamRecords(
sessionIdOrExternalId: string,
io: "out" | "in",
- options?: { afterEventId?: string; baseUrl?: string }
+ options?: { afterEventId?: string; baseUrl?: string; channel?: string }
) {
const qs = new URLSearchParams();
if (options?.afterEventId !== undefined) {
qs.set("afterEventId", options.afterEventId);
}
- const url = `${options?.baseUrl ?? this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(
+ const recordsBase = `${options?.baseUrl ?? this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(
sessionIdOrExternalId
- )}/${io}/records${qs.toString() ? `?${qs.toString()}` : ""}`;
+ )}`;
+ const recordsPath = options?.channel
+ ? `${recordsBase}/channels/${encodeURIComponent(options.channel)}/${io}/records`
+ : `${recordsBase}/${io}/records`;
+ const url = `${recordsPath}${qs.toString() ? `?${qs.toString()}` : ""}`;
return zodfetch(
ReadSessionStreamRecordsResponseBody,
url,
@@ -1477,6 +1489,11 @@ export class ApiClient {
options?: {
signal?: AbortSignal;
baseUrl?: string;
+ /**
+ * A named side channel on the session. When omitted, the session's
+ * reserved default channel (`session.in` / `session.out`) is used.
+ */
+ channel?: string;
timeoutInSeconds?: number;
onComplete?: () => void;
onError?: (error: Error) => void;
@@ -1496,7 +1513,10 @@ export class ApiClient {
onControl?: (event: ControlEvent) => void;
}
): Promise> {
- const url = `${options?.baseUrl ?? this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}`;
+ const sessionSegment = `${options?.baseUrl ?? this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}`;
+ const url = options?.channel
+ ? `${sessionSegment}/channels/${encodeURIComponent(options.channel)}/${io}`
+ : `${sessionSegment}/${io}`;
const subscription = new SSEStreamSubscription(url, {
headers: this.getHeaders(),
diff --git a/packages/core/src/v3/realtimeStreams/sessionStreamOneshot.ts b/packages/core/src/v3/realtimeStreams/sessionStreamOneshot.ts
index 9aa25fa82dd..9fd8911d511 100644
--- a/packages/core/src/v3/realtimeStreams/sessionStreamOneshot.ts
+++ b/packages/core/src/v3/realtimeStreams/sessionStreamOneshot.ts
@@ -23,8 +23,8 @@ import type { StreamWriteResult } from "./types.js";
type IO = "out" | "in";
-async function getS2Stream(apiClient: ApiClient, sessionId: string, io: IO) {
- const response = await apiClient.initializeSessionStream(sessionId, io);
+async function getS2Stream(apiClient: ApiClient, sessionId: string, io: IO, channel?: string) {
+ const response = await apiClient.initializeSessionStream(sessionId, io, undefined, channel);
const headers = response.headers ?? {};
const accessToken = headers["x-s2-access-token"];
const basin = headers["x-s2-basin"];
@@ -65,9 +65,10 @@ export async function writeSessionControlRecord(
sessionId: string,
io: IO,
subtype: TriggerControlSubtype | string,
- extraHeaders?: ReadonlyArray
+ extraHeaders?: ReadonlyArray,
+ channel?: string
): Promise {
- const stream = await getS2Stream(apiClient, sessionId, io);
+ const stream = await getS2Stream(apiClient, sessionId, io, channel);
const headers: ReadonlyArray = [
[TRIGGER_CONTROL_HEADER, subtype],
...(extraHeaders ?? []),
@@ -93,9 +94,10 @@ export async function writeSessionControlRecord(
export async function trimSessionStream(
apiClient: ApiClient,
sessionId: string,
- earliestSeqNum: number
+ earliestSeqNum: number,
+ channel?: string
): Promise {
- const stream = await getS2Stream(apiClient, sessionId, "out");
+ const stream = await getS2Stream(apiClient, sessionId, "out", channel);
await stream.append(AppendInput.create([AppendRecord.trim(earliestSeqNum)]));
}
diff --git a/packages/core/src/v3/session-streams-api.ts b/packages/core/src/v3/session-streams-api.ts
index 638a8674213..8ea970b7070 100644
--- a/packages/core/src/v3/session-streams-api.ts
+++ b/packages/core/src/v3/session-streams-api.ts
@@ -5,6 +5,7 @@ import { SessionStreamsAPI } from "./sessionStreams/index.js";
export const sessionStreams = SessionStreamsAPI.getInstance();
export * from "./sessionStreams/types.js";
+export * from "./sessionStreams/channels.js";
export * from "./sessionStreams/wireProtocol.js";
export * from "./sessionStreams/chatSnapshot.js";
export * from "./sessionStreams/router.js";
diff --git a/packages/core/src/v3/sessionStreams/channels.ts b/packages/core/src/v3/sessionStreams/channels.ts
new file mode 100644
index 00000000000..85916cd7975
--- /dev/null
+++ b/packages/core/src/v3/sessionStreams/channels.ts
@@ -0,0 +1,29 @@
+export type SessionChannelShape = { in?: unknown; out?: unknown };
+
+/**
+ * A typed declaration of a named Session channel. The channel analogue of
+ * `Task`: `TName` captures the channel's literal name and
+ * `TShape` its per-direction record types. `__shape` is a phantom carrier
+ * for `TShape` and is never read at runtime.
+ */
+export type SessionChannel<
+ TName extends string = string,
+ TShape extends SessionChannelShape = SessionChannelShape,
+> = {
+ readonly name: TName;
+ readonly __shape?: TShape;
+};
+
+export type AnySessionChannel = SessionChannel;
+
+/** Extract a channel's literal name, the analogue of `TaskIdentifier`. */
+export type SessionChannelName =
+ C extends SessionChannel ? N : never;
+
+/** Extract the `.out` record type, the analogue of `TaskOutput`. */
+export type SessionChannelOut =
+ C extends SessionChannel ? (S extends { out: infer O } ? O : unknown) : never;
+
+/** Extract the `.in` record type, the analogue of `TaskPayload`. */
+export type SessionChannelIn =
+ C extends SessionChannel ? (S extends { in: infer I } ? I : unknown) : never;
diff --git a/packages/core/src/v3/sessionStreams/index.ts b/packages/core/src/v3/sessionStreams/index.ts
index a1b6f840cf9..9f046e96089 100644
--- a/packages/core/src/v3/sessionStreams/index.ts
+++ b/packages/core/src/v3/sessionStreams/index.ts
@@ -36,110 +36,139 @@ export class SessionStreamsAPI implements SessionStreamManager {
public on(
sessionId: string,
io: SessionChannelIO,
- handler: (data: unknown) => void | boolean | Promise
+ handler: (data: unknown) => void | boolean | Promise,
+ channel?: string
): { off: () => void } {
- return this.#getManager().on(sessionId, io, handler);
+ return this.#getManager().on(sessionId, io, handler, channel);
}
public onRecord(
sessionId: string,
io: SessionChannelIO,
- handler: (record: SessionStreamRecord) => void | boolean | Promise
+ handler: (record: SessionStreamRecord) => void | boolean | Promise,
+ channel?: string
): { off: () => void } {
const manager = this.#getManager();
if (!manager.onRecord) {
throw new Error("The configured Session stream manager does not support record handlers");
}
- return manager.onRecord(sessionId, io, handler);
+ return manager.onRecord(sessionId, io, handler, channel);
}
public once(
sessionId: string,
io: SessionChannelIO,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise {
- return this.#getManager().once(sessionId, io, options);
+ return this.#getManager().once(sessionId, io, options, channel);
}
public onceRecord(
sessionId: string,
io: SessionChannelIO,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise {
const manager = this.#getManager();
if (!manager.onceRecord) {
throw new Error("The configured Session stream manager does not support record metadata");
}
- return manager.onceRecord(sessionId, io, options);
+ return manager.onceRecord(sessionId, io, options, channel);
}
public onceRecordWhere(
sessionId: string,
io: SessionChannelIO,
predicate: SessionStreamRecordPredicate,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise {
const manager = this.#getManager();
if (!manager.onceRecordWhere) {
throw new Error("The configured Session stream manager does not support selective records");
}
- return manager.onceRecordWhere(sessionId, io, predicate, options);
+ return manager.onceRecordWhere(sessionId, io, predicate, options, channel);
}
- public peek(sessionId: string, io: SessionChannelIO): unknown | undefined {
- return this.#getManager().peek(sessionId, io);
+ public peek(sessionId: string, io: SessionChannelIO, channel?: string): unknown | undefined {
+ return this.#getManager().peek(sessionId, io, channel);
}
- public peekRecord(sessionId: string, io: SessionChannelIO): SessionStreamRecord | undefined {
+ public peekRecord(
+ sessionId: string,
+ io: SessionChannelIO,
+ channel?: string
+ ): SessionStreamRecord | undefined {
const manager = this.#getManager();
if (!manager.peekRecord) {
throw new Error("The configured Session stream manager does not support record metadata");
}
- return manager.peekRecord(sessionId, io);
+ return manager.peekRecord(sessionId, io, channel);
}
- public lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
- return this.#getManager().lastSeqNum(sessionId, io);
+ public lastSeqNum(sessionId: string, io: SessionChannelIO, channel?: string): number | undefined {
+ return this.#getManager().lastSeqNum(sessionId, io, channel);
}
- public setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
- this.#getManager().setLastSeqNum(sessionId, io, seqNum);
+ public setLastSeqNum(
+ sessionId: string,
+ io: SessionChannelIO,
+ seqNum: number,
+ channel?: string
+ ): void {
+ this.#getManager().setLastSeqNum(sessionId, io, seqNum, channel);
}
- public consumeRecord(sessionId: string, io: SessionChannelIO, seqNum: number): void {
+ public consumeRecord(
+ sessionId: string,
+ io: SessionChannelIO,
+ seqNum: number,
+ channel?: string
+ ): void {
const manager = this.#getManager();
if (!manager.consumeRecord) {
throw new Error("The configured Session stream manager does not support exact consumption");
}
- manager.consumeRecord(sessionId, io, seqNum);
+ manager.consumeRecord(sessionId, io, seqNum, channel);
}
- public lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
- return this.#getManager().lastDispatchedSeqNum(sessionId, io);
+ public lastDispatchedSeqNum(
+ sessionId: string,
+ io: SessionChannelIO,
+ channel?: string
+ ): number | undefined {
+ return this.#getManager().lastDispatchedSeqNum(sessionId, io, channel);
}
- public setLastDispatchedSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
- this.#getManager().setLastDispatchedSeqNum(sessionId, io, seqNum);
+ public setLastDispatchedSeqNum(
+ sessionId: string,
+ io: SessionChannelIO,
+ seqNum: number,
+ channel?: string
+ ): void {
+ this.#getManager().setLastDispatchedSeqNum(sessionId, io, seqNum, channel);
}
public setMinTimestamp(
sessionId: string,
io: SessionChannelIO,
- minTimestamp: number | undefined
+ minTimestamp: number | undefined,
+ channel?: string
): void {
- this.#getManager().setMinTimestamp(sessionId, io, minTimestamp);
+ this.#getManager().setMinTimestamp(sessionId, io, minTimestamp, channel);
}
- public shiftBuffer(sessionId: string, io: SessionChannelIO): boolean {
- return this.#getManager().shiftBuffer(sessionId, io);
+ public shiftBuffer(sessionId: string, io: SessionChannelIO, channel?: string): boolean {
+ return this.#getManager().shiftBuffer(sessionId, io, channel);
}
- public reconnectStream(sessionId: string, io: SessionChannelIO): void {
- this.#getManager().reconnectStream?.(sessionId, io);
+ public reconnectStream(sessionId: string, io: SessionChannelIO, channel?: string): void {
+ this.#getManager().reconnectStream?.(sessionId, io, channel);
}
- public disconnectStream(sessionId: string, io: SessionChannelIO): void {
- this.#getManager().disconnectStream(sessionId, io);
+ public disconnectStream(sessionId: string, io: SessionChannelIO, channel?: string): void {
+ this.#getManager().disconnectStream(sessionId, io, channel);
}
public clearHandlers(): void {
diff --git a/packages/core/src/v3/sessionStreams/manager.test.ts b/packages/core/src/v3/sessionStreams/manager.test.ts
index 4262674bfe3..f5e7af6ee99 100644
--- a/packages/core/src/v3/sessionStreams/manager.test.ts
+++ b/packages/core/src/v3/sessionStreams/manager.test.ts
@@ -70,6 +70,87 @@ function repeatingApiClient(record: {
} as unknown as ApiClient;
}
+function channelAwareApiClient(
+ byChannel: Record>
+): ApiClient {
+ const delivered = new Set();
+ return {
+ async subscribeToSessionStream(
+ _sessionIdOrExternalId: string,
+ _io: "out" | "in",
+ options?: {
+ onPart?: (part: SSEStreamPart) => void;
+ signal?: AbortSignal;
+ channel?: string;
+ }
+ ) {
+ const channelKey = options?.channel ?? "";
+ if (!delivered.has(channelKey)) {
+ delivered.add(channelKey);
+ for (const record of byChannel[channelKey] ?? []) {
+ options?.onPart?.(record as SSEStreamPart);
+ }
+ }
+ const signal = options?.signal;
+ // eslint-disable-next-line require-yield
+ return (async function* () {
+ if (signal?.aborted) return;
+ await new Promise((resolve) => {
+ signal?.addEventListener("abort", () => resolve(), { once: true });
+ });
+ })() as unknown as Awaited>;
+ },
+ } as unknown as ApiClient;
+}
+
+describe("StandardSessionStreamManager — named channels", () => {
+ const sessionId = "session-1";
+ const io = "in" as const;
+
+ it("routes records to the addressed channel and never across channels", async () => {
+ const manager = new StandardSessionStreamManager(
+ channelAwareApiClient({
+ a: [{ id: "0", chunk: { v: "a-record" }, timestamp: 1000 }],
+ b: [{ id: "0", chunk: { v: "b-record" }, timestamp: 1000 }],
+ }),
+ "http://localhost"
+ );
+
+ const fromA = await manager.once(sessionId, io, { timeoutMs: 500 }, "a");
+ const fromB = await manager.once(sessionId, io, { timeoutMs: 500 }, "b");
+
+ expect(fromA).toEqual({ ok: true, output: { v: "a-record" } });
+ expect(fromB).toEqual({ ok: true, output: { v: "b-record" } });
+
+ manager.disconnectStream(sessionId, io, "a");
+ manager.disconnectStream(sessionId, io, "b");
+ manager.disconnect();
+ });
+
+ it("keeps the reserved channel isolated from a named channel", async () => {
+ const manager = new StandardSessionStreamManager(
+ channelAwareApiClient({
+ "": [{ id: "0", chunk: { v: "reserved" }, timestamp: 1000 }],
+ screenshots: [{ id: "0", chunk: { v: "named" }, timestamp: 1000 }],
+ }),
+ "http://localhost"
+ );
+
+ const reserved = await manager.once(sessionId, io, { timeoutMs: 500 });
+ const named = await manager.once(sessionId, io, { timeoutMs: 500 }, "screenshots");
+
+ expect(reserved).toEqual({ ok: true, output: { v: "reserved" } });
+ expect(named).toEqual({ ok: true, output: { v: "named" } });
+
+ expect(manager.peek(sessionId, io)).toBeUndefined();
+ expect(manager.peek(sessionId, io, "screenshots")).toBeUndefined();
+
+ manager.disconnectStream(sessionId, io);
+ manager.disconnectStream(sessionId, io, "screenshots");
+ manager.disconnect();
+ });
+});
+
describe("StandardSessionStreamManager — minTimestamp filter", () => {
const sessionId = "session-1";
const io = "in" as const;
diff --git a/packages/core/src/v3/sessionStreams/manager.ts b/packages/core/src/v3/sessionStreams/manager.ts
index 73c85f972e2..0e833315813 100644
--- a/packages/core/src/v3/sessionStreams/manager.ts
+++ b/packages/core/src/v3/sessionStreams/manager.ts
@@ -47,8 +47,8 @@ type TailState = {
promise: Promise;
};
-function keyFor(sessionId: string, io: SessionChannelIO): string {
- return `${sessionId}:${io}`;
+function keyFor(sessionId: string, io: SessionChannelIO, channel?: string): string {
+ return `${sessionId}:${channel ?? ""}:${io}`;
}
/**
@@ -103,8 +103,13 @@ export class StandardSessionStreamManager implements SessionStreamManager {
private debug: boolean = false
) {}
- on(sessionId: string, io: SessionChannelIO, handler: SessionStreamHandler): { off: () => void } {
- return this.#register(sessionId, io, { kind: "data", fn: handler });
+ on(
+ sessionId: string,
+ io: SessionChannelIO,
+ handler: SessionStreamHandler,
+ channel?: string
+ ): { off: () => void } {
+ return this.#register(sessionId, io, { kind: "data", fn: handler }, channel);
}
/**
@@ -114,17 +119,19 @@ export class StandardSessionStreamManager implements SessionStreamManager {
onRecord(
sessionId: string,
io: SessionChannelIO,
- handler: SessionStreamRecordHandler
+ handler: SessionStreamRecordHandler,
+ channel?: string
): { off: () => void } {
- return this.#register(sessionId, io, { kind: "record", fn: handler });
+ return this.#register(sessionId, io, { kind: "record", fn: handler }, channel);
}
#register(
sessionId: string,
io: SessionChannelIO,
- handler: RegisteredHandler
+ handler: RegisteredHandler,
+ channel?: string
): { off: () => void } {
- const key = keyFor(sessionId, io);
+ const key = keyFor(sessionId, io, channel);
let handlerSet = this.handlers.get(key);
if (!handlerSet) {
@@ -136,7 +143,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
// Explicit re-attach clears the "explicitly disconnected" suppression
// so the tail can subscribe again now that callers want delivery back.
this.explicitlyDisconnected.delete(key);
- this.#ensureTailConnected(sessionId, io);
+ this.#ensureTailConnected(sessionId, io, channel);
// Selective drain: offer each buffered record to the new handler and
// remove ONLY the ones it consumed (returned `true` — e.g. the
@@ -181,9 +188,10 @@ export class StandardSessionStreamManager implements SessionStreamManager {
once(
sessionId: string,
io: SessionChannelIO,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise {
- const recordPromise = this.onceRecord(sessionId, io, options);
+ const recordPromise = this.onceRecord(sessionId, io, options, channel);
return new InputStreamOncePromise((resolve, reject) => {
recordPromise.then((result) => {
resolve(result.ok ? { ok: true, output: result.output.data } : result);
@@ -194,27 +202,30 @@ export class StandardSessionStreamManager implements SessionStreamManager {
onceRecord(
sessionId: string,
io: SessionChannelIO,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise {
- return this.#onceRecord(sessionId, io, undefined, options);
+ return this.#onceRecord(sessionId, io, undefined, options, channel);
}
onceRecordWhere(
sessionId: string,
io: SessionChannelIO,
predicate: SessionStreamRecordPredicate,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise {
- return this.#onceRecord(sessionId, io, predicate, options);
+ return this.#onceRecord(sessionId, io, predicate, options, channel);
}
#onceRecord(
sessionId: string,
io: SessionChannelIO,
predicate: SessionStreamRecordPredicate | undefined,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise {
- const key = keyFor(sessionId, io);
+ const key = keyFor(sessionId, io, channel);
if (options?.timeoutMs === 0) {
const record = this.#takeBufferedRecord(key, predicate);
@@ -228,7 +239,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
}
this.explicitlyDisconnected.delete(key);
- this.#ensureTailConnected(sessionId, io);
+ this.#ensureTailConnected(sessionId, io, channel);
const record = this.#takeBufferedRecord(key, predicate);
if (record) {
@@ -293,28 +304,32 @@ export class StandardSessionStreamManager implements SessionStreamManager {
return record;
}
- peek(sessionId: string, io: SessionChannelIO): unknown | undefined {
- return this.peekRecord(sessionId, io)?.data;
+ peek(sessionId: string, io: SessionChannelIO, channel?: string): unknown | undefined {
+ return this.peekRecord(sessionId, io, channel)?.data;
}
- peekRecord(sessionId: string, io: SessionChannelIO): SessionStreamRecord | undefined {
- return this.buffer.get(keyFor(sessionId, io))?.[0];
+ peekRecord(
+ sessionId: string,
+ io: SessionChannelIO,
+ channel?: string
+ ): SessionStreamRecord | undefined {
+ return this.buffer.get(keyFor(sessionId, io, channel))?.[0];
}
- lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
- return this.seqNums.get(keyFor(sessionId, io));
+ lastSeqNum(sessionId: string, io: SessionChannelIO, channel?: string): number | undefined {
+ return this.seqNums.get(keyFor(sessionId, io, channel));
}
- setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
- const key = keyFor(sessionId, io);
+ setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number, channel?: string): void {
+ const key = keyFor(sessionId, io, channel);
const current = this.seqNums.get(key);
if (current === undefined || seqNum > current) {
this.seqNums.set(key, seqNum);
}
}
- consumeRecord(sessionId: string, io: SessionChannelIO, seqNum: number): void {
- const key = keyFor(sessionId, io);
+ consumeRecord(sessionId: string, io: SessionChannelIO, seqNum: number, channel?: string): void {
+ const key = keyFor(sessionId, io, channel);
const buffered = this.buffer.get(key);
const index = buffered?.findIndex((record) => record.seqNum === seqNum) ?? -1;
@@ -329,8 +344,12 @@ export class StandardSessionStreamManager implements SessionStreamManager {
this.#drainOnceWaitersFromBuffer(key);
}
- lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
- const key = keyFor(sessionId, io);
+ lastDispatchedSeqNum(
+ sessionId: string,
+ io: SessionChannelIO,
+ channel?: string
+ ): number | undefined {
+ const key = keyFor(sessionId, io, channel);
const highWatermark = this.lastDispatchedSeqNums.get(key);
if (highWatermark === undefined) return undefined;
@@ -346,10 +365,15 @@ export class StandardSessionStreamManager implements SessionStreamManager {
return safeCursor >= 0 ? safeCursor : undefined;
}
- setLastDispatchedSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
+ setLastDispatchedSeqNum(
+ sessionId: string,
+ io: SessionChannelIO,
+ seqNum: number,
+ channel?: string
+ ): void {
if (!Number.isFinite(seqNum)) return;
- this.#advanceLastDispatched(keyFor(sessionId, io), seqNum);
+ this.#advanceLastDispatched(keyFor(sessionId, io, channel), seqNum);
}
#advanceLastDispatched(key: string, seqNum: number): void {
@@ -380,8 +404,13 @@ export class StandardSessionStreamManager implements SessionStreamManager {
}
}
- setMinTimestamp(sessionId: string, io: SessionChannelIO, minTimestamp: number | undefined): void {
- const key = keyFor(sessionId, io);
+ setMinTimestamp(
+ sessionId: string,
+ io: SessionChannelIO,
+ minTimestamp: number | undefined,
+ channel?: string
+ ): void {
+ const key = keyFor(sessionId, io, channel);
if (minTimestamp === undefined) {
this.minTimestamps.delete(key);
} else {
@@ -389,8 +418,8 @@ export class StandardSessionStreamManager implements SessionStreamManager {
}
}
- shiftBuffer(sessionId: string, io: SessionChannelIO): boolean {
- const key = keyFor(sessionId, io);
+ shiftBuffer(sessionId: string, io: SessionChannelIO, channel?: string): boolean {
+ const key = keyFor(sessionId, io, channel);
const buffered = this.buffer.get(key);
if (buffered && buffered.length > 0) {
const record = buffered.shift()!;
@@ -404,8 +433,8 @@ export class StandardSessionStreamManager implements SessionStreamManager {
return false;
}
- disconnectStream(sessionId: string, io: SessionChannelIO): void {
- const key = keyFor(sessionId, io);
+ disconnectStream(sessionId: string, io: SessionChannelIO, channel?: string): void {
+ const key = keyFor(sessionId, io, channel);
const tail = this.tails.get(key);
// Mark as explicitly disconnected BEFORE we abort, so the tail's
// `.finally` reconnect path sees the flag when it runs (which can be
@@ -429,10 +458,10 @@ export class StandardSessionStreamManager implements SessionStreamManager {
* its handler just to clear the suppression flag would replay the buffer at
* it.
*/
- reconnectStream(sessionId: string, io: SessionChannelIO): void {
- const key = keyFor(sessionId, io);
+ reconnectStream(sessionId: string, io: SessionChannelIO, channel?: string): void {
+ const key = keyFor(sessionId, io, channel);
this.explicitlyDisconnected.delete(key);
- this.#ensureTailConnected(sessionId, io);
+ this.#ensureTailConnected(sessionId, io, channel);
}
clearHandlers(): void {
@@ -485,12 +514,12 @@ export class StandardSessionStreamManager implements SessionStreamManager {
this.buffer.clear();
}
- #ensureTailConnected(sessionId: string, io: SessionChannelIO): void {
- const key = keyFor(sessionId, io);
+ #ensureTailConnected(sessionId: string, io: SessionChannelIO, channel?: string): void {
+ const key = keyFor(sessionId, io, channel);
if (this.tails.has(key)) return;
const abortController = new AbortController();
- const promise = this.#runTail(sessionId, io, abortController.signal)
+ const promise = this.#runTail(sessionId, io, abortController.signal, channel)
.catch((error) => {
if (this.debug) {
console.error(`[SessionStreamManager] Tail error for "${key}":`, error);
@@ -530,15 +559,20 @@ export class StandardSessionStreamManager implements SessionStreamManager {
const stillHasWaiters =
this.onceWaiters.has(key) && this.onceWaiters.get(key)!.length > 0;
if (!stillHasHandlers && !stillHasWaiters) return;
- this.#ensureTailConnected(sessionId, io);
+ this.#ensureTailConnected(sessionId, io, channel);
}, delayMs);
}
});
this.tails.set(key, { abortController, promise });
}
- async #runTail(sessionId: string, io: SessionChannelIO, signal: AbortSignal): Promise {
- const key = keyFor(sessionId, io);
+ async #runTail(
+ sessionId: string,
+ io: SessionChannelIO,
+ signal: AbortSignal,
+ channel?: string
+ ): Promise {
+ const key = keyFor(sessionId, io, channel);
try {
const lastSeq = this.seqNums.get(key);
// Dispatch is driven from `onPart` (not the for-await loop) so each
@@ -549,6 +583,7 @@ export class StandardSessionStreamManager implements SessionStreamManager {
const stream = await this.apiClient.subscribeToSessionStream(sessionId, io, {
signal,
baseUrl: this.baseUrl,
+ channel,
timeoutInSeconds: 600,
lastEventId: lastSeq !== undefined ? String(lastSeq) : undefined,
onPart: (part) => {
diff --git a/packages/core/src/v3/sessionStreams/types.ts b/packages/core/src/v3/sessionStreams/types.ts
index cc6bde884cc..ccf394829fc 100644
--- a/packages/core/src/v3/sessionStreams/types.ts
+++ b/packages/core/src/v3/sessionStreams/types.ts
@@ -47,7 +47,8 @@ export interface SessionStreamManager {
on(
sessionId: string,
io: SessionChannelIO,
- handler: (data: unknown) => void | boolean | Promise
+ handler: (data: unknown) => void | boolean | Promise,
+ channel?: string
): { off: () => void };
/**
@@ -57,21 +58,24 @@ export interface SessionStreamManager {
onRecord?(
sessionId: string,
io: SessionChannelIO,
- handler: (record: SessionStreamRecord) => void | boolean | Promise
+ handler: (record: SessionStreamRecord) => void | boolean | Promise,
+ channel?: string
): { off: () => void };
/** Wait for the next record on the given channel (buffered or live). */
once(
sessionId: string,
io: SessionChannelIO,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise;
/** Wait for and consume the next record, including its durable metadata. */
onceRecord?(
sessionId: string,
io: SessionChannelIO,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise;
/**
@@ -83,23 +87,28 @@ export interface SessionStreamManager {
sessionId: string,
io: SessionChannelIO,
predicate: SessionStreamRecordPredicate,
- options?: InputStreamOnceOptions
+ options?: InputStreamOnceOptions,
+ channel?: string
): InputStreamOncePromise;
/** Non-blocking peek at the head of the channel buffer. */
- peek(sessionId: string, io: SessionChannelIO): unknown | undefined;
+ peek(sessionId: string, io: SessionChannelIO, channel?: string): unknown | undefined;
/** Non-blocking peek at the head record, including its durable metadata. */
- peekRecord?(sessionId: string, io: SessionChannelIO): SessionStreamRecord | undefined;
+ peekRecord?(
+ sessionId: string,
+ io: SessionChannelIO,
+ channel?: string
+ ): SessionStreamRecord | undefined;
/** Last S2 sequence number seen on the given channel. */
- lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined;
+ lastSeqNum(sessionId: string, io: SessionChannelIO, channel?: string): number | undefined;
/** Advance the last-seen sequence number (prevents SSE replay after `.wait` resume). */
- setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void;
+ setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number, channel?: string): void;
/** Consume one exact record delivered through the waitpoint path. */
- consumeRecord?(sessionId: string, io: SessionChannelIO, seqNum: number): void;
+ consumeRecord?(sessionId: string, io: SessionChannelIO, seqNum: number, channel?: string): void;
/**
* Highest sequence number that is safe to persist as consumed. When a later
@@ -111,7 +120,11 @@ export interface SessionStreamManager {
* `turn-complete` control record so the next worker boot can resume
* the channel from this point without replaying processed messages.
*/
- lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined;
+ lastDispatchedSeqNum(
+ sessionId: string,
+ io: SessionChannelIO,
+ channel?: string
+ ): number | undefined;
/**
* Seed the committed-consume cursor at worker boot — e.g. from the
@@ -119,7 +132,12 @@ export interface SessionStreamManager {
* `.out`. Monotonic: only ever advances forward, never backwards. Existing
* unconsumed records still constrain {@link lastDispatchedSeqNum}.
*/
- setLastDispatchedSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void;
+ setLastDispatchedSeqNum(
+ sessionId: string,
+ io: SessionChannelIO,
+ seqNum: number,
+ channel?: string
+ ): void;
/**
* Set a per-stream lower-bound SSE timestamp. Records whose timestamp
@@ -129,16 +147,21 @@ export interface SessionStreamManager {
*
* Pass `undefined` to clear the filter.
*/
- setMinTimestamp(sessionId: string, io: SessionChannelIO, minTimestamp: number | undefined): void;
+ setMinTimestamp(
+ sessionId: string,
+ io: SessionChannelIO,
+ minTimestamp: number | undefined,
+ channel?: string
+ ): void;
/** Remove and discard the first buffered record. Returns true if one was removed. */
- shiftBuffer(sessionId: string, io: SessionChannelIO): boolean;
+ shiftBuffer(sessionId: string, io: SessionChannelIO, channel?: string): boolean;
/** Abort the SSE tail while preserving buffered records. Called before `.wait` suspends. */
- disconnectStream(sessionId: string, io: SessionChannelIO): void;
+ disconnectStream(sessionId: string, io: SessionChannelIO, channel?: string): void;
/** Re-open a channel closed by {@link disconnectStream}, registering nothing. */
- reconnectStream?(sessionId: string, io: SessionChannelIO): void;
+ reconnectStream?(sessionId: string, io: SessionChannelIO, channel?: string): void;
/** Clear all `.on` handlers; abort tails without pending once-waiters. */
clearHandlers(): void;
diff --git a/packages/react-hooks/src/hooks/useSessionStreamChannel.ts b/packages/react-hooks/src/hooks/useSessionStreamChannel.ts
new file mode 100644
index 00000000000..5258a0d5777
--- /dev/null
+++ b/packages/react-hooks/src/hooks/useSessionStreamChannel.ts
@@ -0,0 +1,378 @@
+"use client";
+
+import type {
+ AnySessionChannel,
+ ApiClient,
+ ControlEvent,
+ SessionChannelIn,
+ SessionChannelName,
+ SessionChannelOut,
+ SSEStreamPart,
+} from "@trigger.dev/core/v3";
+import { useCallback, useEffect, useId, useRef, useState } from "react";
+import { createThrottledQueue } from "../utils/throttle.js";
+import type { KeyedMutator } from "../utils/trigger-swr.js";
+import { useSWR } from "../utils/trigger-swr.js";
+import { useStableRequestCallback } from "../utils/useStableRequestCallback.js";
+import type { UseApiClientOptions } from "./useApiClient.js";
+import { useApiClient } from "./useApiClient.js";
+
+type ChannelRecord = S extends "out"
+ ? SessionChannelOut
+ : SessionChannelIn;
+
+export type UseSessionStreamChannelInstance = {
+ /** The records received so far on the channel, in arrival order. */
+ records: Array;
+ /** The cursor of the last record seen; pass back as `lastEventId` to resume. */
+ lastEventId: string | undefined;
+ /** The last control record seen on the channel. */
+ lastControl: ControlEvent | undefined;
+ error: Error | undefined;
+ /** Abort the current request immediately, keep the records received so far. */
+ stop: () => void;
+};
+
+export type UseSessionStreamChannelOptions<
+ TChannel extends AnySessionChannel,
+ S extends "in" | "out",
+> = UseApiClientOptions & {
+ /**
+ * The id or external id of the session that owns the channel. May be
+ * undefined while it resolves; the subscription starts once it is set.
+ */
+ sessionId?: string;
+ id?: string;
+ enabled?: boolean;
+ /**
+ * Which side of the channel to read.
+ *
+ * @default "out"
+ */
+ io?: S;
+ /**
+ * The number of milliseconds to throttle the record updates.
+ *
+ * @default 16
+ */
+ throttleInMs?: number;
+ /**
+ * The number of seconds to wait for new data before the stream closes.
+ *
+ * @default 60 seconds
+ */
+ timeoutInSeconds?: number;
+ /** The cursor to resume from. If not provided, reads per `from`. */
+ lastEventId?: string | number;
+ /**
+ * Where a fresh subscription (no `lastEventId`) starts reading.
+ *
+ * - `"beginning"` (default): replay the full channel history, then live-tail.
+ * - `"latest"`: start at the current tail, for a last-value / live view.
+ *
+ * Ignored when `lastEventId` is set.
+ */
+ from?: "beginning" | "latest";
+ /**
+ * Cap the number of records kept in `records`. Use `maxRecords: 1` with
+ * `from: "latest"` for a bounded last-value view.
+ */
+ maxRecords?: number;
+ /** Invoked once per throttled flush with the batch of records (control records included). */
+ onRecords?: (records: Array>>) => void;
+ /** Called when a control record is received on the channel. */
+ onControl?: (event: ControlEvent) => void;
+};
+
+/**
+ * Read one side of a named Session side channel, with record types inferred
+ * from a `defineSessionChannel` declaration passed as the type argument.
+ *
+ * The channel name is typesafe (`SessionChannelName`) and `records`
+ * is typed from the channel's `.out` / `.in` record type. Called without the
+ * type argument, the channel name is any string and `records` is `unknown`.
+ *
+ * Requires a Public Access Token scoped to the session (or to the channel).
+ *
+ * @example
+ * ```tsx
+ * import type { screenshotsChannel } from "./shared/channels";
+ *
+ * const { records } = useSessionStreamChannel("screenshots", {
+ * sessionId,
+ * accessToken,
+ * io: "out",
+ * from: "latest",
+ * maxRecords: 1,
+ * });
+ * ```
+ */
+export function useSessionStreamChannel<
+ TChannel extends AnySessionChannel = AnySessionChannel,
+ S extends "in" | "out" = "out",
+>(
+ channel: SessionChannelName,
+ options: UseSessionStreamChannelOptions
+): UseSessionStreamChannelInstance> {
+ type TRecord = ChannelRecord;
+
+ const hookId = useId();
+ const idKey = options.id ?? hookId;
+ const io = (options.io ?? "out") as "out" | "in";
+ const sessionId = options.sessionId;
+ const channelName = channel as string;
+
+ const [initialRecordsFallback] = useState([] as Array);
+
+ const { data: records, mutate: mutateRecords } = useSWR>(
+ [idKey, sessionId, channelName, io, "records"],
+ null,
+ { fallbackData: initialRecordsFallback }
+ );
+
+ const recordsRef = useRef>(records ?? ([] as Array));
+ useEffect(() => {
+ recordsRef.current = records || ([] as Array);
+ }, [records]);
+
+ const { data: lastEventId = undefined, mutate: setLastEventId } = useSWR(
+ [idKey, sessionId, channelName, io, "lastEventId"],
+ null
+ );
+ const lastEventIdRef = useRef(lastEventId);
+ const channelIdentityRef = useRef(`${idKey}:${sessionId}:${channelName}:${io}`);
+ useEffect(() => {
+ const identity = `${idKey}:${sessionId}:${channelName}:${io}`;
+ if (channelIdentityRef.current !== identity) {
+ channelIdentityRef.current = identity;
+ lastEventIdRef.current = lastEventId;
+ }
+ }, [idKey, sessionId, channelName, io, lastEventId]);
+
+ const { data: lastControl = undefined, mutate: setLastControl } = useSWR<
+ undefined | ControlEvent
+ >([idKey, sessionId, channelName, io, "lastControl"], null);
+
+ const { data: _isComplete = false, mutate: setIsComplete } = useSWR(
+ [idKey, sessionId, channelName, io, "complete"],
+ null
+ );
+
+ const { data: error = undefined, mutate: setError } = useSWR(
+ [idKey, sessionId, channelName, io, "error"],
+ null
+ );
+
+ const abortControllerRef = useRef(null);
+
+ const stop = useCallback(() => {
+ if (abortControllerRef.current) {
+ abortControllerRef.current.abort();
+ abortControllerRef.current = null;
+ }
+ }, []);
+
+ const onRecordsCallback = options.onRecords;
+ const onRecords = useCallback(
+ (recordsBatch: Array>) => {
+ if (onRecordsCallback) {
+ onRecordsCallback(recordsBatch);
+ }
+ },
+ [onRecordsCallback]
+ );
+
+ const onControlCallback = options.onControl;
+ const onControl = useCallback(
+ (event: ControlEvent) => {
+ if (onControlCallback) {
+ onControlCallback(event);
+ }
+ },
+ [onControlCallback]
+ );
+
+ const apiClient = useApiClient(options);
+ const timeoutInSeconds = options.timeoutInSeconds;
+ const startEventId = options.lastEventId;
+ const throttleInMs = options.throttleInMs;
+ const from = options.from;
+ const maxRecords = options.maxRecords;
+
+ useEffect(() => {
+ if (maxRecords != null && maxRecords >= 0) {
+ const current = recordsRef.current;
+ if (current.length > maxRecords) {
+ mutateRecords(current.slice(current.length - maxRecords));
+ }
+ }
+ }, [maxRecords, mutateRecords]);
+
+ const triggerRequest = useCallback(async () => {
+ let abortController: AbortController | null = null;
+ try {
+ if (!sessionId || !apiClient) {
+ return;
+ }
+
+ abortController = new AbortController();
+ abortControllerRef.current = abortController;
+
+ await processSessionChannelStream(
+ sessionId,
+ io,
+ channelName,
+ apiClient,
+ mutateRecords,
+ recordsRef,
+ setLastEventId,
+ setLastControl,
+ setError,
+ onRecords,
+ onControl,
+ abortControllerRef,
+ timeoutInSeconds,
+ startEventId !== undefined ? String(startEventId) : lastEventIdRef.current,
+ throttleInMs ?? 16,
+ from,
+ maxRecords
+ );
+ } catch (err) {
+ if ((err as any).name === "AbortError") {
+ return;
+ }
+
+ setError(err as Error);
+ } finally {
+ if (abortControllerRef.current === abortController) {
+ abortControllerRef.current = null;
+ }
+
+ setIsComplete(true);
+ }
+ }, [
+ sessionId,
+ io,
+ channelName,
+ apiClient,
+ mutateRecords,
+ setLastEventId,
+ setLastControl,
+ setError,
+ setIsComplete,
+ onRecords,
+ onControl,
+ timeoutInSeconds,
+ startEventId,
+ throttleInMs,
+ from,
+ maxRecords,
+ ]);
+ const requestSubscription = useStableRequestCallback(triggerRequest);
+
+ useEffect(() => {
+ if (typeof options.enabled === "boolean" && !options.enabled) {
+ return;
+ }
+
+ if (!sessionId) {
+ return;
+ }
+
+ requestSubscription().finally(() => {});
+
+ return () => {
+ stop();
+ };
+ }, [sessionId, channelName, io, stop, options.enabled, requestSubscription]);
+
+ return { records: records ?? initialRecordsFallback, lastEventId, lastControl, error, stop };
+}
+
+async function processSessionChannelStream(
+ sessionIdOrExternalId: string,
+ io: "out" | "in",
+ channel: string,
+ apiClient: ApiClient,
+ mutateRecordsData: KeyedMutator>,
+ existingRecordsRef: React.MutableRefObject>,
+ setLastEventId: KeyedMutator,
+ setLastControl: KeyedMutator,
+ onError: (e: Error) => void,
+ onRecords: (records: Array>) => void,
+ onControl: (event: ControlEvent) => void,
+ abortControllerRef: React.MutableRefObject,
+ timeoutInSeconds?: number,
+ lastEventId?: string,
+ throttleInMs?: number,
+ from?: "beginning" | "latest",
+ maxRecords?: number
+) {
+ let lastSeenEventId: string | undefined;
+ let publishedEventId: string | undefined;
+ let partsBatch: Array> = [];
+
+ const publishLastEventId = () => {
+ if (lastSeenEventId !== publishedEventId) {
+ publishedEventId = lastSeenEventId;
+ setLastEventId(lastSeenEventId);
+ }
+ };
+
+ const flushParts = () => {
+ if (partsBatch.length === 0) return;
+ const batch = partsBatch;
+ partsBatch = [];
+ onRecords(batch);
+ };
+
+ try {
+ const stream = await apiClient.subscribeToSessionStream(sessionIdOrExternalId, io, {
+ signal: abortControllerRef.current?.signal,
+ channel,
+ timeoutInSeconds,
+ lastEventId,
+ from,
+ onPart: (part) => {
+ lastSeenEventId = part.id;
+ partsBatch.push(part);
+ },
+ onControl: (event) => {
+ setLastControl(event);
+ onControl(event);
+ },
+ });
+
+ const recordsQueue = createThrottledQueue(async (newRecords) => {
+ const combined = [...existingRecordsRef.current, ...newRecords];
+ const bounded =
+ maxRecords != null && maxRecords >= 0 && combined.length > maxRecords
+ ? combined.slice(combined.length - maxRecords)
+ : combined;
+ existingRecordsRef.current = bounded;
+ mutateRecordsData(bounded);
+ publishLastEventId();
+ flushParts();
+ }, throttleInMs);
+
+ for await (const record of stream) {
+ recordsQueue.add(record);
+ }
+
+ await recordsQueue.flush();
+ publishLastEventId();
+ flushParts();
+ } catch (err) {
+ if ((err as any).name === "AbortError") {
+ return;
+ }
+
+ if (err instanceof Error) {
+ onError(err);
+ } else {
+ onError(new Error(String(err)));
+ }
+
+ throw err;
+ }
+}
diff --git a/packages/react-hooks/src/index.ts b/packages/react-hooks/src/index.ts
index 57aa3b16877..6f6c967409b 100644
--- a/packages/react-hooks/src/index.ts
+++ b/packages/react-hooks/src/index.ts
@@ -6,3 +6,4 @@ export * from "./hooks/useTaskTrigger.js";
export * from "./hooks/useWaitToken.js";
export * from "./hooks/useInputStreamSend.js";
export * from "./hooks/useSessionStream.js";
+export * from "./hooks/useSessionStreamChannel.js";
diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts
index bcc70fa9ce0..2cfd6b22b43 100644
--- a/packages/trigger-sdk/src/v3/ai.ts
+++ b/packages/trigger-sdk/src/v3/ai.ts
@@ -41,6 +41,8 @@ import {
type RouterCheckpoint,
type SessionRouteTable,
type SessionStreamRecord,
+ type AnySessionChannel,
+ type SessionChannelName,
} from "@trigger.dev/core/v3";
import type {
FinishReason,
@@ -107,6 +109,7 @@ type ToolCallOptions = {
import { readFileInSkill, runBashInSkill } from "./agentSkillsRuntime.js";
import { ensureAiSdkTelemetry } from "./aiAutoTelemetry.js";
import {
+ type SessionChannelHandleFor,
type SessionHandle,
type SessionPipeStreamOptions,
sessions,
@@ -11849,6 +11852,17 @@ export const chat = {
response: chatResponse,
/** Pre-built input stream for receiving messages from the transport. */
messages: messagesInput,
+ /** The current chat.agent run's Session handle. See {@link SessionHandle}. */
+ session: getChatSession,
+ /**
+ * Open a named side channel on the current chat.agent run's Session: a
+ * durable, cross-run `.in`/`.out` pair addressed by `name`, separate from the
+ * chat transcript. Writing its `.in` does not wake a run. Shortcut for
+ * `chat.session().channel(name)`.
+ */
+ channel: (
+ channel: SessionChannelName | C
+ ): SessionChannelHandleFor => getChatSession().channel(channel),
/** Create a managed stop signal wired to the stop input stream. See {@link createStopSignal}. */
createStopSignal,
/** Signal the frontend that the current turn is complete. See {@link chatWriteTurnComplete}. */
diff --git a/packages/trigger-sdk/src/v3/sessions.test.ts b/packages/trigger-sdk/src/v3/sessions.test.ts
index abeccb0c12d..8a2b7062d26 100644
--- a/packages/trigger-sdk/src/v3/sessions.test.ts
+++ b/packages/trigger-sdk/src/v3/sessions.test.ts
@@ -96,7 +96,7 @@ describe("SessionOutputChannel initializeSessionStream cache", () => {
await Promise.all([p1.waitUntilComplete(), p2.waitUntilComplete(), p3.waitUntilComplete()]);
expect(spy).toHaveBeenCalledTimes(1);
- expect(spy).toHaveBeenCalledWith("session-1", "out", undefined);
+ expect(spy).toHaveBeenCalledWith("session-1", "out", undefined, undefined);
});
it("evicts on initialize failure so the next call retries instead of returning a poisoned entry", async () => {
@@ -150,8 +150,8 @@ describe("SessionOutputChannel initializeSessionStream cache", () => {
]);
expect(spy).toHaveBeenCalledTimes(2);
- expect(spy).toHaveBeenCalledWith("session-a", "out", undefined);
- expect(spy).toHaveBeenCalledWith("session-b", "out", undefined);
+ expect(spy).toHaveBeenCalledWith("session-a", "out", undefined, undefined);
+ expect(spy).toHaveBeenCalledWith("session-b", "out", undefined, undefined);
});
it("evicts the cache when a writer's wait() rejects (simulated stale-token failure)", async () => {
diff --git a/packages/trigger-sdk/src/v3/sessions.ts b/packages/trigger-sdk/src/v3/sessions.ts
index 9758d534f21..4638dae27ca 100644
--- a/packages/trigger-sdk/src/v3/sessions.ts
+++ b/packages/trigger-sdk/src/v3/sessions.ts
@@ -21,6 +21,12 @@ import type {
UpdateSessionRequestBody,
WriterStreamOptions,
CursorPagePromise,
+ AnySessionChannel,
+ SessionChannel,
+ SessionChannelIn,
+ SessionChannelName,
+ SessionChannelOut,
+ SessionChannelShape,
} from "@trigger.dev/core/v3";
import {
InputStreamOncePromise,
@@ -58,6 +64,7 @@ export const sessions = {
close: closeSession,
list: listSessions,
open,
+ defineChannel,
};
// Test hook: lets `@trigger.dev/sdk/ai/test` replace `sessions.open()` with
@@ -252,6 +259,59 @@ export class SessionHandle {
this.out = overrides?.out ?? new SessionOutputChannel(id);
this.in = overrides?.in ?? new SessionInputChannel(id);
}
+
+ /**
+ * Open a named side channel on this session: a durable, cross-run `.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; a run observes it via
+ * `.in.on()` / `.in.once()`. Records outlive any single run and are bounded by
+ * the org's stream retention, the same as the reserved chat streams.
+ *
+ * Pass a `sessions.defineChannel(...)` definition to type `.in`/`.out` records;
+ * a bare name string works too, with records typed `unknown`.
+ */
+ channel(
+ channel: SessionChannelName | C
+ ): SessionChannelHandleFor {
+ const name = typeof channel === "string" ? channel : channel.name;
+ if (!SESSION_CHANNEL_NAME_REGEX.test(name)) {
+ throw new Error(
+ `Invalid session channel name "${name}": use 1-128 chars from [A-Za-z0-9._-].`
+ );
+ }
+ return {
+ name,
+ out: new SessionOutputChannel(this.id, name),
+ in: new SessionInputChannel(this.id, name),
+ } as SessionChannelHandleFor;
+ }
+}
+
+export type SessionChannelHandleFor = {
+ readonly name: string;
+ readonly out: SessionOutputChannel>;
+ readonly in: SessionInputChannel>;
+};
+
+export type SessionChannelHandle = SessionChannelHandleFor;
+
+const SESSION_CHANNEL_NAME_REGEX = /^[A-Za-z0-9._-]{1,128}$/;
+
+/**
+ * Declare a named Session channel with typed `.in` / `.out` records, inferred
+ * on both the producer and the consumer. The channel analogue of a task
+ * definition: pass the result to `session.channel(...)` / `chat.channel(...)`
+ * and to `useSessionStreamChannel` so the record types line up
+ * on every side.
+ */
+function defineChannel<
+ TShape extends SessionChannelShape = SessionChannelShape,
+ const TName extends string = string,
+>(name: TName): SessionChannel {
+ if (!SESSION_CHANNEL_NAME_REGEX.test(name)) {
+ throw new Error(`Invalid session channel name "${name}": use 1-128 chars from [A-Za-z0-9._-].`);
+ }
+ return { name };
}
/**
@@ -268,7 +328,7 @@ export type SessionPipeStreamOptions = Omit;
* consume via SSE. S2 credentials for direct writes are fetched
* internally by `pipe`/`writer` — there's no public `initialize()`.
*/
-export class SessionOutputChannel {
+export class SessionOutputChannel {
// Cache of the in-flight / resolved `initializeSessionStream` PUT for
// this channel. Every `pipe()` / `writer()` call needs the same S2
// credentials, so we share a single promise instead of re-PUTing on
@@ -279,7 +339,10 @@ export class SessionOutputChannel {
// Evicts on failure (so the next call retries) and on `reset()`.
#initPromise?: Promise;
- constructor(public readonly sessionId: string) {}
+ constructor(
+ public readonly sessionId: string,
+ public readonly channel?: string
+ ) {}
/**
* Drop the cached `initializeSessionStream` response. Surfaces for
@@ -300,8 +363,8 @@ export class SessionOutputChannel {
* which would give SSE consumers a JSON-string instead of an object.
* Mirrors how `streams.define.append` delegates to `streams.writer`.
*/
- async append(value: T, options?: SessionPipeStreamOptions): Promise {
- const { waitUntilComplete } = this.writer({
+ async append(value: TOut, options?: SessionPipeStreamOptions): Promise {
+ const { waitUntilComplete } = this.writer({
...options,
spanName: "sessions.append()",
execute: ({ write }) => {
@@ -317,7 +380,7 @@ export class SessionOutputChannel {
* {@link SessionStreamInstance}. Parallel to {@link streams.pipe} but
* session-scoped — no `target` option because the session is the target.
*/
- pipe(
+ pipe(
value: AsyncIterable | ReadableStream,
options?: SessionPipeStreamOptions
): PipeStreamResult {
@@ -331,7 +394,7 @@ export class SessionOutputChannel {
* stream and await completion. Span is collapsible via `options.spanName`
* / `options.collapsed`.
*/
- writer(options: WriterStreamOptions): PipeStreamResult {
+ writer(options: WriterStreamOptions): PipeStreamResult {
let controller!: ReadableStreamDefaultController;
const ongoingStreamPromises: Promise[] = [];
@@ -407,11 +470,12 @@ export class SessionOutputChannel {
* shared {@link SSEStreamSubscription} plumbing used by run-scoped
* realtime streams.
*/
- async read(options?: SessionSubscribeOptions): Promise> {
+ async read(options?: SessionSubscribeOptions): Promise> {
const apiClient = apiClientManager.clientOrThrow();
return apiClient.subscribeToSessionStream(this.sessionId, "out", {
signal: options?.signal,
+ channel: this.channel,
timeoutInSeconds: options?.timeoutInSeconds,
lastEventId: options?.lastEventId != null ? String(options.lastEventId) : undefined,
onPart: options?.onPart,
@@ -433,12 +497,18 @@ export class SessionOutputChannel {
attributes: {
session: this.sessionId,
io: "out",
+ ...(this.channel ? { channel: this.channel } : {}),
[SemanticInternalAttributes.ENTITY_TYPE]: "session-stream",
- [SemanticInternalAttributes.ENTITY_ID]: `${this.sessionId}:out`,
+ [SemanticInternalAttributes.ENTITY_ID]: `${this.sessionId}:${this.channel ?? ""}:out`,
[SemanticInternalAttributes.STYLE_ICON]: "sessions",
...(collapsed ? { [SemanticInternalAttributes.COLLAPSED]: true } : {}),
...accessoryAttributes({
- items: [{ text: `${this.sessionId}.out`, variant: "normal" }],
+ items: this.channel
+ ? [
+ { text: this.channel, variant: "normal" },
+ { text: "out", variant: "normal" },
+ ]
+ : [{ text: `${this.sessionId}.out`, variant: "normal" }],
style: "codepath",
}),
},
@@ -481,7 +551,8 @@ export class SessionOutputChannel {
const fresh = apiClient.initializeSessionStream(
this.sessionId,
"out",
- options?.requestOptions
+ options?.requestOptions,
+ this.channel
);
this.#initPromise = fresh;
// Evict on failure so the next call retries instead of returning a
@@ -569,7 +640,14 @@ export class SessionOutputChannel {
extraHeaders?: ReadonlyArray
): Promise {
const apiClient = apiClientManager.clientOrThrow();
- return writeSessionControlRecord(apiClient, this.sessionId, "out", subtype, extraHeaders);
+ return writeSessionControlRecord(
+ apiClient,
+ this.sessionId,
+ "out",
+ subtype,
+ extraHeaders,
+ this.channel
+ );
}
/**
@@ -583,7 +661,7 @@ export class SessionOutputChannel {
*/
async trimTo(earliestSeqNum: number): Promise {
const apiClient = apiClientManager.clientOrThrow();
- await trimSessionStream(apiClient, this.sessionId, earliestSeqNum);
+ await trimSessionStream(apiClient, this.sessionId, earliestSeqNum, this.channel);
}
}
@@ -594,8 +672,19 @@ export class SessionOutputChannel {
* external clients. Keyed on the session rather than the run so a
* conversation can survive across run boundaries.
*/
-export class SessionInputChannel {
- constructor(public readonly sessionId: string) {}
+export class SessionInputChannel {
+ constructor(
+ public readonly sessionId: string,
+ public readonly channel?: string
+ ) {}
+
+ #assertReservedChannelForWait(method: string): void {
+ if (this.channel) {
+ throw new Error(
+ `session.channel("${this.channel}").in.${method} is not supported: a named side channel does not wake a run. Use .in.on() / .in.once() to observe it instead.`
+ );
+ }
+ }
/**
* Send a single record to the channel. Called by external clients
@@ -603,21 +692,34 @@ export class SessionInputChannel {
* Matches {@link streams.input.send} but session-scoped — the session
* is the address, no `runId` required.
*/
- async send(value: unknown, requestOptions?: ApiRequestOptions): Promise {
+ async send(value: TIn, requestOptions?: ApiRequestOptions): Promise {
const apiClient = apiClientManager.clientOrThrow();
const body = typeof value === "string" ? value : JSON.stringify(value);
+ const spanName = this.channel
+ ? `sessions.open(${this.sessionId}).channel(${this.channel}).in.send()`
+ : `sessions.open(${this.sessionId}).in.send()`;
+
const $requestOptions = mergeRequestOptions(
{
tracer,
- name: `sessions.open(${this.sessionId}).in.send()`,
+ name: spanName,
icon: "sessions",
- attributes: sessionAttributes(this.sessionId, { io: "in" }),
+ attributes: sessionAttributes(this.sessionId, {
+ io: "in",
+ ...(this.channel ? { channel: this.channel } : {}),
+ }),
},
requestOptions
);
- await apiClient.appendToSessionStream(this.sessionId, "in", body, $requestOptions);
+ await apiClient.appendToSessionStream(
+ this.sessionId,
+ "in",
+ body,
+ $requestOptions,
+ this.channel
+ );
}
/**
@@ -630,11 +732,12 @@ export class SessionInputChannel {
* won't be buffered for a later `once()` and won't be re-delivered on a
* future `on()` attach. Plain observers should return nothing.
*/
- on(handler: (data: T) => void | boolean | Promise): { off: () => void } {
+ on(handler: (data: T) => void | boolean | Promise): { off: () => void } {
return sessionStreams.on(
this.sessionId,
"in",
- handler as (data: unknown) => void | boolean | Promise
+ handler as (data: unknown) => void | boolean | Promise,
+ this.channel
);
}
@@ -643,11 +746,11 @@ export class SessionInputChannel {
* Returns `{ ok: true, output }` on arrival or `{ ok: false, error }`
* when the timeout fires. Chain `.unwrap()` to get the data directly.
*/
- once(options?: InputStreamOnceOptions): InputStreamOncePromise {
+ once(options?: InputStreamOnceOptions): InputStreamOncePromise {
const ctx = taskContext.ctx;
const runId = ctx?.run.id;
- const innerPromise = sessionStreams.once(this.sessionId, "in", options);
+ const innerPromise = sessionStreams.once(this.sessionId, "in", options, this.channel);
return new InputStreamOncePromise((resolve, reject) => {
tracer
@@ -662,12 +765,22 @@ export class SessionInputChannel {
[SemanticInternalAttributes.STYLE_ICON]: "sessions",
[SemanticInternalAttributes.ENTITY_TYPE]: "session-stream",
...(runId
- ? { [SemanticInternalAttributes.ENTITY_ID]: `${runId}:${this.sessionId}:in` }
+ ? {
+ [SemanticInternalAttributes.ENTITY_ID]: `${runId}:${this.sessionId}:${
+ this.channel ?? ""
+ }:in`,
+ }
: {}),
session: this.sessionId,
io: "in",
+ ...(this.channel ? { channel: this.channel } : {}),
...accessoryAttributes({
- items: [{ text: `${this.sessionId}.in`, variant: "normal" }],
+ items: this.channel
+ ? [
+ { text: this.channel, variant: "normal" },
+ { text: "in", variant: "normal" },
+ ]
+ : [{ text: `${this.sessionId}.in`, variant: "normal" }],
style: "codepath",
}),
},
@@ -678,8 +791,8 @@ export class SessionInputChannel {
}
/** Non-blocking peek at the head of the `.in` buffer. */
- peek(): T | undefined {
- return sessionStreams.peek(this.sessionId, "in") as T | undefined;
+ peek(): T | undefined {
+ return sessionStreams.peek(this.sessionId, "in", this.channel) as T | undefined;
}
/**
@@ -693,7 +806,7 @@ export class SessionInputChannel {
* past already-processed user messages.
*/
lastDispatchedSeqNum(): number | undefined {
- return sessionStreams.lastDispatchedSeqNum(this.sessionId, "in");
+ return sessionStreams.lastDispatchedSeqNum(this.sessionId, "in", this.channel);
}
/**
@@ -717,6 +830,7 @@ export class SessionInputChannel {
async awaitWake(
options?: InputStreamWaitOptions & { lastSeqNum?: number }
): Promise<{ ok: true; waitpointId: string } | { ok: false; error: Error }> {
+ this.#assertReservedChannelForWait("awaitWake()");
const ctx = taskContext.ctx;
if (!ctx) {
@@ -774,6 +888,7 @@ export class SessionInputChannel {
wait(options?: InputStreamWaitOptions): ManualWaitpointPromise {
return new ManualWaitpointPromise(async (resolve, reject) => {
try {
+ this.#assertReservedChannelForWait("wait()");
const apiClient = apiClientManager.clientOrThrow();
const result = await tracer.startActiveSpan(
@@ -843,6 +958,7 @@ export class SessionInputChannel {
async waitWithIdleTimeout(
options: InputStreamWaitWithIdleTimeoutOptions
): Promise<{ ok: true; output: T } | { ok: false; error?: Error }> {
+ this.#assertReservedChannelForWait("waitWithIdleTimeout()");
// eslint-disable-next-line no-this-alias
const self = this;
const spanName =
Waiting…
; +} +``` + +`useSessionStreamChannel` has the same options and return shape as [`useSessionStream`](/realtime/react-hooks/session-stream) (`io`, `from`, `maxRecords`, `lastEventId`, `onRecords`, `onControl`, `throttleInMs`, `timeoutInSeconds`), plus the typed channel generic. A bare name string works without the generic, with `records` typed `unknown`. + +The client writes the `.in` control with a session handle: `sessions.open(sessionId).channel(screenshots).in.send({ paused: true })`. This appends to the channel and does not wake a run. + +## From MCP + +An MCP client can read and write a session's channels with two [MCP tools](/mcp-tools): `read_session_channel` drains a channel's records (with an optional `timeoutInSeconds` to wait for the next one), and `write_session_channel` appends a record to a channel's `.in` to send control input to a running agent. Reading `.out` gives the producer feed (e.g. the screencast); writing `.in` does not wake a run, and `.out` stays producer-only. + +## Retention + +A side channel's streams are bounded by the same retention as the rest of your realtime streams: streams are created on demand when first written and age out on your plan's retention window, with empty streams cleaned up automatically. A channel needs no separate setup or trimming. + +