feat(sdk,core,webapp,react-hooks): named side channels on a Session (#4815)
## Summary Adds **named side channels** to a Session: durable, two-way realtime streams that outlive a single run and are shared across every run of the session. Today a Session has exactly one reserved `.in`/`.out` pair (the chat transcript). This lets a session hold any number of *named* channels alongside it, each its own `.in`/`.out` pair, so an agent can stream out-of-band data (a feed of frames, telemetry, a control channel) on a stream separate from the transcript while many clients read it live. The two properties a named channel adds over the reserved pair: 1. It is addressed by a name that outlives a run and is shared across runs, not welded to the chat turn loop. 2. Writing its `.in` does **not** wake or trigger a run. A run observes it by subscribing; an external client writes it without spawning anything. This is the generalization half of the Momentic ask (stream browser screenshots from a `chat.agent` to the frontend on a channel separate from the chat). It builds directly on the start-from-latest / `useSessionStream` subscribe seam from #4811. ## Usage Declare the channel's record types once and infer them on both sides: ```ts // channels.ts (shared, client imports it type-only) import { sessions } from "@trigger.dev/sdk"; export const screenshots = sessions.defineChannel<{ out: ScreenshotFrame; in: ViewportControl }>( "screenshots" ); ``` Open a channel from a session handle (`sessions.open(id)` returns one for a known session id). Writing its `.out` is durable, cross-run, and wakes nothing; a run observes its `.in` by tailing, without suspending: ```ts import { sessions } from "@trigger.dev/sdk"; import { screenshots } from "./channels"; const channel = sessions.open(sessionId).channel(screenshots); await channel.out.append(frame); // frame: ScreenshotFrame (typed from the definition) channel.in.on((control) => { /* ... */ }); // control: ViewportControl, tail, no suspend ``` Passing the definition types `.out.append` / `.in.on` on the producer side; a bare name string also works, with records typed `unknown`. An external client writes the `.in` without waking a run, and reads the `.out` from React: ```ts sessions.open(sessionId).channel("screenshots").in.send({ paused: true }); const { records } = useSessionStreamChannel<typeof screenshots>("screenshots", { sessionId, accessToken, io: "out", from: "latest", maxRecords: 1, }); ``` `session.channel(name)` returns the same `{ in, out }` handle shape as the reserved pair, so `append` / `pipe` / `writer` / `read` / `writeControl` / `trimTo` on `.out` and `send` / `on` / `once` / `peek` on `.in` all carry over. Passing a name other than the declared one is a type error; a bare-string call without the generic stays valid with `records` typed `unknown`. ### With `chat.agent` This is the motivating case: a `chat.agent` answers on the reserved transcript as usual, and streams screenshot frames on a side channel in parallel. `chat.channel(name)` opens a channel on the current run's own Session, so there's no id to thread: ```ts import { chat } from "@trigger.dev/sdk/ai"; import { streamText } from "ai"; import { screenshots } from "./channels"; export const browserAgent = chat.agent({ id: "browser-agent", run: async ({ messages, signal }) => { const frames = chat.channel(screenshots); // client pause/resume arrives here without waking a turn frames.in.on((control: ViewportControl) => applyViewport(control)); // frames stream on their own channel, not the chat transcript driveBrowser({ signal, onFrame: (frame) => frames.out.append(frame) }); // the assistant reply still goes to the reserved transcript return streamText({ model: openai("gpt-4o"), messages, abortSignal: signal }); }, }); ``` `chat.channel(name)` is a shortcut for `chat.session().channel(name)`; `chat.session()` returns the current run's full `SessionHandle` if you need it. The frontend renders the transcript with `useChat` as before, and the screenshots with `useSessionStreamChannel<typeof screenshots>("screenshots", { sessionId: chatId, io: "out", from: "latest", maxRecords: 1 })`: a live view of the newest frame that survives across turns (each turn is a new run), because the channel is keyed on the session, not the run. ### From MCP An MCP client can observe and write a session's channels with two tools, built on the same apiClient surface as the hook and the dashboard viewer: - `read_session_channel` reads records from a channel (or the reserved pair). It is a point-in-time drain with cursor pagination (`afterEventId` / `nextCursor`, `maxRecords`); pass `timeoutInSeconds` to wait for the next record when none exist yet. - `write_session_channel` appends one record to a channel's `.in` (an object or a raw string), so an agent can send control input without waking a run. `.out` is producer-only, so it is not writable here. ### On the session page The session detail page lists a session's channels (via an S2 prefix list in the loader) and shows each as a tab beside `Rendered` and `Raw`. Selecting a channel renders its records in the same table as the Raw transcript view, sourced from that channel's `out` and `in` streams. ## How it works **Addressing.** A channel is a stream name segment: `sessions/{id}/channels/{name}/{io}`. The reserved pair keeps its two-part `sessions/{id}/{io}` name for back-compat, and the `channels/` segment means a user channel named `in`/`out` can never collide with it. The channel dimension is threaded through the session stream manager (keyed on `(session, channel, io)`, reserved = absent), `subscribeToSessionStream`, the session apiClient methods, and the `realtime.v1.sessions.$session.channels.$channel.$io.{ts,append,records}` routes. The reserved-pair routes are untouched. The start-from-latest tail path from #4811 is channel-agnostic, so `from: "latest"` and `maxRecords` compose unchanged. **No-wake.** The reserved `.in` append route ensures a run and drains waitpoints so a chat turn advances. The channel `.in` append route deliberately does neither: the record lands durably and a run picks it up when it next subscribes, so writing a side channel can't spawn or resume a run. A named channel's `.in` is therefore subscribe-only from the run side (`.on` / `.once` / `.peek`); `.wait()` / `waitWithIdleTimeout()` throw with a message pointing at the observe methods. **Auth.** Channel scope folds into the existing resource id (`sessions:<key>:channels:<channel>`), so no RBAC grammar change. A channel route authorizes both the channel-folded id and the bare session id, which means a session-wide token grants every channel while a channel-scoped token grants only its own. The per-io rule is preserved per channel: writing `.out` requires secret-key auth so a browser can't forge frames; `.in` is writable with the session token. **Retention.** Channel streams are created on demand on first write and inherit the org's stream retention (bounded age plus delete-on-empty from the store's default config), the same as the reserved chat streams. There is no per-channel control-plane call on the write path. Custom per-channel retention is deferred until the stream store can set config inline on the on-demand create, which avoids a control-plane round trip. **Spans.** Channel writes carry `channel` and `io` attributes, an accessory chip, and the session icon. Clicking a channel span in the run's span inspector renders the channel's actual records with the same viewer the run realtime streams use, rather than the raw properties JSON. ## Verification - **Unit (core):** the stream manager isolates channels: two channels on the same `(session, io)` never cross buffers, and a named channel is isolated from the reserved pair. - **Full-stack e2e** against a real stack (webapp, stream store, Postgres, real runs): - a named `.out` record is readable back **after the triggering run has gone terminal** (durable, cross-run); - a channel `.in` append creates **no** run, while a reserved `.in` append **does** wake one (the differential is the red/green); - `from: "latest"` on a named channel delivers the live record and does **not** replay the backlog from the start; - the span inspector renders a channel span's records, and the MCP read/write tools round-trip records on a real session; - an invalid channel name is rejected. ## Notes - **Channel listing works on the self-hosted store too.** The stream store's list operation is available on s2-lite, so the session page's channel list is an OSS feature. It is a control-plane call made once per session-page load (best-effort; a failure just hides the tabs), not on the write path. - **The ~1 MiB per-record cap is unchanged.** Large payloads (e.g. raw screenshots) still need object-store pointers on the channel rather than inline bytes; that's independent of this change. - Docs ride this branch: the side channels guide, the `useSessionStreamChannel` reference, and the MCP tools list are all updated here. ## Screenshots <img width="3444" height="1870" alt="CleanShot 2026-08-28 at 21 46 27@2x" src="https://github.com/user-attachments/assets/c192aaee-b946-4824-87b7-ca057514d25e" />
This commit is contained in:
Eric Allam
@@ -0,0 +1,16 @@
|
||||
---
|
||||
"@trigger.dev/react-hooks": patch
|
||||
"@trigger.dev/core": patch
|
||||
"@trigger.dev/sdk": patch
|
||||
---
|
||||
|
||||
Named side channels on a Session: durable, two-way realtime streams that outlive a single run and are shared across runs. Open a channel with `sessions.open(id).channel(name)` (or `chat.channel(name)` inside a `chat.agent`) to get an `.in`/`.out` pair addressed by name rather than the reserved default pair. Writing a side channel's `.in` does not wake or trigger a run, so a channel can carry out-of-band data (a stream of frames, a control signal) that many clients read while the agent produces it.
|
||||
|
||||
```ts
|
||||
// Inside a chat.agent: stream frames on a named channel, wakes nothing
|
||||
const frames = chat.channel("screenshots");
|
||||
await frames.out.append(frame);
|
||||
frames.in.on((control) => { /* client control, no suspend */ });
|
||||
```
|
||||
|
||||
Declare channel record types once with `sessions.defineChannel(...)` and infer them on both the producer and the consumer, including `useSessionStreamChannel` in React. Channels get a default retention that keeps them bounded, overridable per channel.
|
||||
@@ -36,6 +36,7 @@ import { PythonLogoIcon } from "~/assets/icons/PythonLogoIcon";
|
||||
import { TraceIcon } from "~/assets/icons/TraceIcon";
|
||||
import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";
|
||||
import { StreamsIcon } from "~/assets/icons/StreamsIcon";
|
||||
import { AIChatIcon } from "~/assets/icons/AIChatIcon";
|
||||
|
||||
type TaskIconProps = {
|
||||
name: string | undefined;
|
||||
@@ -169,6 +170,8 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
|
||||
className={cn(className, "text-text-dimmed group-hover/spannode:text-text-bright")}
|
||||
/>
|
||||
);
|
||||
case "sessions":
|
||||
return <AIChatIcon className={cn(className, "text-sessions")} />;
|
||||
case "hero-sparkles":
|
||||
return (
|
||||
<SparklesIcon
|
||||
|
||||
@@ -840,6 +840,45 @@ export class SpanPresenter extends BasePresenter {
|
||||
},
|
||||
};
|
||||
}
|
||||
case "session-stream": {
|
||||
if (!span.entity.id) {
|
||||
logger.error(`SpanPresenter: No session stream id`, {
|
||||
spanId,
|
||||
sessionStreamId: span.entity.id,
|
||||
});
|
||||
return { ...data, entity: null };
|
||||
}
|
||||
|
||||
const parts = span.entity.id.split(":");
|
||||
const io = parts.at(-1);
|
||||
const channel = parts.at(-2) ?? "";
|
||||
const sessionId = parts.at(-3);
|
||||
|
||||
if (!sessionId || (io !== "out" && io !== "in")) {
|
||||
logger.error(`SpanPresenter: Invalid session stream id`, {
|
||||
spanId,
|
||||
sessionStreamId: span.entity.id,
|
||||
});
|
||||
return { ...data, entity: null };
|
||||
}
|
||||
|
||||
const metadata = span.entity.metadata
|
||||
? (safeJsonParse(span.entity.metadata) as Record<string, unknown> | undefined)
|
||||
: undefined;
|
||||
|
||||
return {
|
||||
...data,
|
||||
entity: {
|
||||
type: "session-stream" as const,
|
||||
object: {
|
||||
sessionId,
|
||||
channel: channel.length > 0 ? channel : undefined,
|
||||
io,
|
||||
metadata,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
case "prompt": {
|
||||
const promptData = extractPromptSpanData(span.properties as Record<string, unknown>);
|
||||
|
||||
|
||||
+101
-9
@@ -57,6 +57,14 @@ import { redirectWithErrorMessage } from "~/models/message.server";
|
||||
import { findProjectBySlug } from "~/models/project.server";
|
||||
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
|
||||
import { SessionPresenter } from "~/presenters/v3/SessionPresenter.server";
|
||||
import { tryCatch } from "@trigger.dev/core/utils";
|
||||
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
|
||||
import {
|
||||
canonicalSessionAddressingKey,
|
||||
resolveSessionByIdOrExternalId,
|
||||
} from "~/services/realtime/sessions.server";
|
||||
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import {
|
||||
type StreamChunk,
|
||||
useRealtimeStream,
|
||||
@@ -115,11 +123,34 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
|
||||
throw new Response("Session not found", { status: 404 });
|
||||
}
|
||||
|
||||
return typedjson({ session, loadedAt: Date.now() });
|
||||
let channels: string[] = [];
|
||||
const streamSessionId = session.agentView?.sessionId;
|
||||
if (streamSessionId) {
|
||||
const [channelsError, listed] = await tryCatch(
|
||||
(async () => {
|
||||
const row = await resolveSessionByIdOrExternalId($replica, environment.id, streamSessionId);
|
||||
if (!row) return [] as string[];
|
||||
const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session: row });
|
||||
if (!(realtimeStream instanceof S2RealtimeStreams)) return [] as string[];
|
||||
const addressingKey = canonicalSessionAddressingKey(row, streamSessionId);
|
||||
return realtimeStream.listSessionChannels(addressingKey);
|
||||
})()
|
||||
);
|
||||
if (channelsError) {
|
||||
logger.warn("Failed to list session channels", {
|
||||
sessionId: streamSessionId,
|
||||
error: channelsError,
|
||||
});
|
||||
} else {
|
||||
channels = listed ?? [];
|
||||
}
|
||||
}
|
||||
|
||||
return typedjson({ session, channels, loadedAt: Date.now() });
|
||||
};
|
||||
|
||||
export default function Page() {
|
||||
const { session, loadedAt } = useTypedLoaderData<typeof loader>();
|
||||
const { session, channels, loadedAt } = useTypedLoaderData<typeof loader>();
|
||||
const organization = useOrganization();
|
||||
const project = useProject();
|
||||
const environment = useEnvironment();
|
||||
@@ -158,7 +189,7 @@ export default function Page() {
|
||||
<PageBody scrollable={false}>
|
||||
<ResizablePanelGroup orientation="horizontal" className="max-h-full">
|
||||
<ResizablePanel id="session-conversation" min={"300px"}>
|
||||
<ConversationPane session={session} />
|
||||
<ConversationPane session={session} channels={channels} />
|
||||
</ResizablePanel>
|
||||
<ResizableHandle id="session-handle" />
|
||||
<ResizablePanel
|
||||
@@ -177,18 +208,49 @@ export default function Page() {
|
||||
|
||||
type LoadedSession = ReturnType<typeof useTypedLoaderData<typeof loader>>["session"];
|
||||
|
||||
function ConversationPane({ session }: { session: LoadedSession }) {
|
||||
function ConversationPane({ session, channels }: { session: LoadedSession; channels: string[] }) {
|
||||
const organization = useOrganization();
|
||||
const project = useProject();
|
||||
const environment = useEnvironment();
|
||||
const { value, replace } = useSearchParams();
|
||||
const isRaw = value("raw") === "1";
|
||||
const channelParam = value("channel");
|
||||
const activeChannel = channelParam && channels.includes(channelParam) ? channelParam : undefined;
|
||||
|
||||
const sessionId = session.agentView.sessionId;
|
||||
const encodedSession = encodeURIComponent(sessionId);
|
||||
const sessionResourceBase = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/sessions/${encodedSession}/realtime/v1`;
|
||||
|
||||
const setView = useCallback((raw: boolean) => replace({ raw: raw ? "1" : undefined }), [replace]);
|
||||
const setView = useCallback(
|
||||
(raw: boolean) => replace({ raw: raw ? "1" : undefined, channel: undefined }),
|
||||
[replace]
|
||||
);
|
||||
const selectChannel = useCallback(
|
||||
(channel: string) => replace({ channel, raw: undefined }),
|
||||
[replace]
|
||||
);
|
||||
|
||||
const utilityBarProps = {
|
||||
channels,
|
||||
activeChannel,
|
||||
onSelectChannel: selectChannel,
|
||||
};
|
||||
|
||||
if (activeChannel) {
|
||||
const channelBase = `${sessionResourceBase}/channels/${encodeURIComponent(activeChannel)}`;
|
||||
return (
|
||||
<div className="flex h-full max-h-full flex-col overflow-hidden bg-background-bright">
|
||||
<RawConversationView
|
||||
key={activeChannel}
|
||||
inResourcePath={`${channelBase}/in`}
|
||||
outResourcePath={`${channelBase}/out`}
|
||||
isRaw={isRaw}
|
||||
onChangeView={setView}
|
||||
{...utilityBarProps}
|
||||
/>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
return (
|
||||
<div className="flex h-full max-h-full flex-col overflow-hidden bg-background-bright">
|
||||
@@ -198,10 +260,11 @@ function ConversationPane({ session }: { session: LoadedSession }) {
|
||||
outResourcePath={`${sessionResourceBase}/out`}
|
||||
isRaw={isRaw}
|
||||
onChangeView={setView}
|
||||
{...utilityBarProps}
|
||||
/>
|
||||
) : (
|
||||
<>
|
||||
<ConversationUtilityBar isRaw={isRaw} onChangeView={setView} />
|
||||
<ConversationUtilityBar isRaw={isRaw} onChangeView={setView} {...utilityBarProps} />
|
||||
<div className="min-w-0 flex-1 overflow-y-auto overflow-x-hidden px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-surface-control">
|
||||
<AgentView agentView={session.agentView} />
|
||||
</div>
|
||||
@@ -214,29 +277,45 @@ function ConversationPane({ session }: { session: LoadedSession }) {
|
||||
function ConversationUtilityBar({
|
||||
isRaw,
|
||||
onChangeView,
|
||||
channels = [],
|
||||
activeChannel,
|
||||
onSelectChannel,
|
||||
right,
|
||||
}: {
|
||||
isRaw: boolean;
|
||||
onChangeView: (raw: boolean) => void;
|
||||
channels?: string[];
|
||||
activeChannel?: string;
|
||||
onSelectChannel?: (channel: string) => void;
|
||||
right?: React.ReactNode;
|
||||
}) {
|
||||
return (
|
||||
<div className="flex h-9 items-center justify-between gap-3 border-b border-grid-bright px-3">
|
||||
<TabContainer className="-mb-2">
|
||||
<TabButton
|
||||
isActive={!isRaw}
|
||||
isActive={!isRaw && !activeChannel}
|
||||
layoutId="conversation-view-mode"
|
||||
onClick={() => onChangeView(false)}
|
||||
>
|
||||
Rendered
|
||||
</TabButton>
|
||||
<TabButton
|
||||
isActive={isRaw}
|
||||
isActive={isRaw && !activeChannel}
|
||||
layoutId="conversation-view-mode"
|
||||
onClick={() => onChangeView(true)}
|
||||
>
|
||||
Raw
|
||||
</TabButton>
|
||||
{channels.map((channel) => (
|
||||
<TabButton
|
||||
key={channel}
|
||||
isActive={activeChannel === channel}
|
||||
layoutId="conversation-view-mode"
|
||||
onClick={() => onSelectChannel?.(channel)}
|
||||
>
|
||||
{channel}
|
||||
</TabButton>
|
||||
))}
|
||||
</TabContainer>
|
||||
{right}
|
||||
</div>
|
||||
@@ -266,11 +345,17 @@ function RawConversationView({
|
||||
outResourcePath,
|
||||
isRaw,
|
||||
onChangeView,
|
||||
channels,
|
||||
activeChannel,
|
||||
onSelectChannel,
|
||||
}: {
|
||||
inResourcePath: string;
|
||||
outResourcePath: string;
|
||||
isRaw: boolean;
|
||||
onChangeView: (raw: boolean) => void;
|
||||
channels?: string[];
|
||||
activeChannel?: string;
|
||||
onSelectChannel?: (channel: string) => void;
|
||||
}) {
|
||||
const {
|
||||
chunks: inChunks,
|
||||
@@ -496,7 +581,14 @@ function RawConversationView({
|
||||
|
||||
return (
|
||||
<>
|
||||
<ConversationUtilityBar isRaw={isRaw} onChangeView={onChangeView} right={controls} />
|
||||
<ConversationUtilityBar
|
||||
isRaw={isRaw}
|
||||
onChangeView={onChangeView}
|
||||
channels={channels}
|
||||
activeChannel={activeChannel}
|
||||
onSelectChannel={onSelectChannel}
|
||||
right={controls}
|
||||
/>
|
||||
<div className="flex min-h-0 flex-1 flex-col bg-background-deep">
|
||||
<div
|
||||
ref={scrollRef}
|
||||
|
||||
@@ -12,6 +12,10 @@ import { $replica, prisma, type PrismaClient } from "~/db.server";
|
||||
import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { mintSessionToken } from "~/services/realtime/mintSessionToken.server";
|
||||
import {
|
||||
isSafeSessionExternalId,
|
||||
SESSION_CHANNEL_SCOPE_INFIX,
|
||||
} from "~/services/realtime/sessionChannels.server";
|
||||
import {
|
||||
ensureRunForSession,
|
||||
type SessionTriggerConfig,
|
||||
@@ -168,6 +172,13 @@ const { action } = createActionApiRoute(
|
||||
},
|
||||
async ({ authentication, body }) => {
|
||||
try {
|
||||
if (body.externalId && !isSafeSessionExternalId(body.externalId)) {
|
||||
return json(
|
||||
{ error: `externalId cannot contain "${SESSION_CHANNEL_SCOPE_INFIX}"` },
|
||||
{ status: 422 }
|
||||
);
|
||||
}
|
||||
|
||||
// Idempotent on (env, externalId): two concurrent POSTs converge to the same row, and
|
||||
// `triggerConfig` is refreshed on the cached path so a redeployed config reaches the next run.
|
||||
const { session, isCached } = await findOrCreateSession({
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
import { json } from "@remix-run/server-runtime";
|
||||
import { tryCatch } from "@trigger.dev/core/utils";
|
||||
import { nanoid } from "nanoid";
|
||||
import { z } from "zod";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
|
||||
import {
|
||||
SESSION_CHANNEL_NAME_REGEX,
|
||||
sessionChannelResources,
|
||||
} from "~/services/realtime/sessionChannels.server";
|
||||
import {
|
||||
canonicalSessionAddressingKey,
|
||||
resolveSessionWithWriterFallback,
|
||||
} from "~/services/realtime/sessions.server";
|
||||
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
|
||||
import {
|
||||
claimSessionStreamPart,
|
||||
releaseSessionStreamPart,
|
||||
} from "~/services/sessionStreamWaitpointCache.server";
|
||||
import { anyResource, createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
|
||||
import { ServiceValidationError } from "~/v3/services/common.server";
|
||||
|
||||
const ParamsSchema = z.object({
|
||||
session: z.string(),
|
||||
channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
|
||||
io: z.enum(["out", "in"]),
|
||||
});
|
||||
|
||||
const MAX_APPEND_BODY_BYTES = 1024 * 1024;
|
||||
|
||||
const { action, loader } = createActionApiRoute(
|
||||
{
|
||||
params: ParamsSchema,
|
||||
method: "POST",
|
||||
maxContentLength: MAX_APPEND_BODY_BYTES,
|
||||
allowJWT: true,
|
||||
corsStrategy: "all",
|
||||
findResource: async (params, auth) =>
|
||||
resolveSessionWithWriterFallback(auth.environment.id, params.session),
|
||||
authorization: {
|
||||
action: "write",
|
||||
resource: (params, _s, _h, _b, session) => {
|
||||
const ids = new Set<string>([params.session]);
|
||||
if (session) {
|
||||
ids.add(session.friendlyId);
|
||||
if (session.externalId) ids.add(session.externalId);
|
||||
}
|
||||
return anyResource(sessionChannelResources(params.channel, ids));
|
||||
},
|
||||
},
|
||||
},
|
||||
async ({ request, params, authentication, resource: session }) => {
|
||||
if (!session) {
|
||||
return new Response("Session not found", { status: 404 });
|
||||
}
|
||||
|
||||
if (session.closedAt) {
|
||||
return json({ ok: false, error: "Cannot append to a closed session" }, { status: 400 });
|
||||
}
|
||||
|
||||
if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
|
||||
return json({ ok: false, error: "Cannot append to an expired session" }, { status: 400 });
|
||||
}
|
||||
|
||||
if (params.io === "out" && authentication.type !== "PRIVATE") {
|
||||
return json(
|
||||
{ ok: false, error: "Appending to the out channel requires secret key authentication" },
|
||||
{ status: 403 }
|
||||
);
|
||||
}
|
||||
|
||||
const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
|
||||
session,
|
||||
});
|
||||
|
||||
if (!(realtimeStream instanceof S2RealtimeStreams)) {
|
||||
return json(
|
||||
{ ok: false, error: "Session channels require the S2 realtime backend" },
|
||||
{ status: 501 }
|
||||
);
|
||||
}
|
||||
|
||||
const addressingKey = canonicalSessionAddressingKey(session, params.session);
|
||||
const claimKey = `${addressingKey}:channels:${params.channel}`;
|
||||
|
||||
const part = await request.text();
|
||||
|
||||
const clientPartId = request.headers.get("X-Part-Id");
|
||||
const partId = clientPartId ?? nanoid(7);
|
||||
|
||||
const wonClaim = clientPartId
|
||||
? await claimSessionStreamPart(
|
||||
authentication.environment.id,
|
||||
claimKey,
|
||||
params.io,
|
||||
clientPartId
|
||||
)
|
||||
: true;
|
||||
|
||||
let appendSeq: number | undefined;
|
||||
if (wonClaim) {
|
||||
const [appendError, seq] = await tryCatch(
|
||||
realtimeStream.appendPartToSessionStream(
|
||||
part,
|
||||
partId,
|
||||
addressingKey,
|
||||
params.io,
|
||||
params.channel
|
||||
)
|
||||
);
|
||||
appendSeq = seq ?? undefined;
|
||||
|
||||
if (appendError) {
|
||||
if (clientPartId) {
|
||||
await releaseSessionStreamPart(
|
||||
authentication.environment.id,
|
||||
claimKey,
|
||||
params.io,
|
||||
clientPartId
|
||||
);
|
||||
}
|
||||
if (appendError instanceof ServiceValidationError) {
|
||||
return json(
|
||||
{ ok: false, error: appendError.message },
|
||||
{ status: appendError.status ?? 422 }
|
||||
);
|
||||
}
|
||||
logger.error("Failed to append to session channel stream", {
|
||||
sessionId: session.id,
|
||||
io: params.io,
|
||||
channel: params.channel,
|
||||
error: appendError,
|
||||
});
|
||||
return json(
|
||||
{ ok: false, error: "Something went wrong, please try again." },
|
||||
{ status: 500 }
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return json({ ok: true, seq: appendSeq }, { status: 200 });
|
||||
}
|
||||
);
|
||||
|
||||
export { action, loader };
|
||||
@@ -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<string>([addressingKey]);
|
||||
if (row) {
|
||||
ids.add(row.friendlyId);
|
||||
if (row.externalId) ids.add(row.externalId);
|
||||
}
|
||||
return anyResource(sessionChannelResources(params.channel, ids));
|
||||
},
|
||||
},
|
||||
},
|
||||
async ({ params, authentication, resource, searchParams }) => {
|
||||
const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
|
||||
session: resource.row,
|
||||
organization: resource.row ? null : authentication.environment.organization,
|
||||
});
|
||||
|
||||
if (!(realtimeStream instanceof S2RealtimeStreams)) {
|
||||
return new Response("Session channels require the S2 realtime backend", { status: 501 });
|
||||
}
|
||||
|
||||
const afterSeqNum =
|
||||
searchParams.afterEventId !== undefined ? Number(searchParams.afterEventId) : undefined;
|
||||
|
||||
const records = await realtimeStream.readSessionStreamRecords(
|
||||
resource.addressingKey,
|
||||
params.io,
|
||||
afterSeqNum,
|
||||
params.channel
|
||||
);
|
||||
|
||||
return json({ records });
|
||||
}
|
||||
);
|
||||
@@ -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<string>([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 };
|
||||
+15
@@ -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 (
|
||||
<RealtimeStreamViewer
|
||||
resourcePath={resourcePath}
|
||||
headerLabel={channel ? "Channel:" : "Session:"}
|
||||
displayName={displayName}
|
||||
/>
|
||||
);
|
||||
}
|
||||
case "ai-generation":
|
||||
case "ai-summary": {
|
||||
return (
|
||||
|
||||
+2
-3
@@ -1,13 +1,12 @@
|
||||
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
|
||||
import { z } from "zod";
|
||||
import { $replica } from "~/db.server";
|
||||
import { findProjectBySlug } from "~/models/project.server";
|
||||
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
|
||||
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
|
||||
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
|
||||
import {
|
||||
canonicalSessionAddressingKey,
|
||||
resolveSessionByIdOrExternalId,
|
||||
resolveSessionWithWriterFallback,
|
||||
} from "~/services/realtime/sessions.server";
|
||||
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
|
||||
import { requireUserId } from "~/services/session.server";
|
||||
@@ -45,7 +44,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
|
||||
return new Response("Environment not found", { status: 404 });
|
||||
}
|
||||
|
||||
const session = await resolveSessionByIdOrExternalId($replica, environment.id, sessionParam);
|
||||
const session = await resolveSessionWithWriterFallback(environment.id, sessionParam);
|
||||
if (!session) {
|
||||
return new Response("Session not found", { status: 404 });
|
||||
}
|
||||
|
||||
+70
@@ -0,0 +1,70 @@
|
||||
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
|
||||
import { z } from "zod";
|
||||
import { findProjectBySlug } from "~/models/project.server";
|
||||
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
|
||||
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
|
||||
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
|
||||
import { SESSION_CHANNEL_NAME_REGEX } from "~/services/realtime/sessionChannels.server";
|
||||
import {
|
||||
canonicalSessionAddressingKey,
|
||||
resolveSessionWithWriterFallback,
|
||||
} from "~/services/realtime/sessions.server";
|
||||
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
|
||||
import { requireUserId } from "~/services/session.server";
|
||||
import { EnvironmentParamSchema } from "~/utils/pathBuilder";
|
||||
|
||||
const ParamsSchema = z.object({
|
||||
sessionParam: z.string(),
|
||||
channel: z.string().regex(SESSION_CHANNEL_NAME_REGEX),
|
||||
io: z.enum(["out", "in"]),
|
||||
});
|
||||
|
||||
export async function loader({ request, params }: LoaderFunctionArgs) {
|
||||
const userId = await requireUserId(request);
|
||||
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
|
||||
const { sessionParam, channel, io } = ParamsSchema.parse(params);
|
||||
|
||||
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
|
||||
if (!project) {
|
||||
return new Response("Project not found", { status: 404 });
|
||||
}
|
||||
|
||||
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
|
||||
if (!environment) {
|
||||
return new Response("Environment not found", { status: 404 });
|
||||
}
|
||||
|
||||
const session = await resolveSessionWithWriterFallback(environment.id, sessionParam);
|
||||
if (!session) {
|
||||
return new Response("Session not found", { status: 404 });
|
||||
}
|
||||
|
||||
const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session });
|
||||
|
||||
if (!(realtimeStream instanceof S2RealtimeStreams)) {
|
||||
return new Response("Session channels require the S2 realtime backend", {
|
||||
status: 501,
|
||||
});
|
||||
}
|
||||
|
||||
const lastEventId = request.headers.get("Last-Event-ID") || undefined;
|
||||
const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds");
|
||||
let timeoutInSeconds: number | undefined;
|
||||
if (timeoutInSecondsRaw !== null) {
|
||||
timeoutInSeconds = Number(timeoutInSecondsRaw);
|
||||
if (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) {
|
||||
return new Response("Invalid timeout", { status: 400 });
|
||||
}
|
||||
}
|
||||
|
||||
const addressingKey = canonicalSessionAddressingKey(session, sessionParam);
|
||||
|
||||
return realtimeStream.streamResponseFromSessionStream(
|
||||
request,
|
||||
addressingKey,
|
||||
io,
|
||||
getRequestAbortSignal(),
|
||||
{ lastEventId, timeoutInSeconds },
|
||||
channel
|
||||
);
|
||||
}
|
||||
@@ -150,8 +150,14 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
|
||||
* the session's `friendlyId` and the I/O direction. Used by the session
|
||||
* realtime routes to route traffic to `sessions/{friendlyId}/{out|in}`.
|
||||
*/
|
||||
public toSessionStreamName(friendlyId: string, io: "out" | "in"): string {
|
||||
return `${this.streamPrefix}/sessions/${friendlyId}/${io}`;
|
||||
public toSessionStreamName(friendlyId: string, io: "out" | "in", channel?: string): string {
|
||||
return `${this.streamPrefix}${this.#sessionStreamRelativeName(friendlyId, io, channel)}`;
|
||||
}
|
||||
|
||||
#sessionStreamRelativeName(friendlyId: string, io: "out" | "in", channel?: string): string {
|
||||
return channel
|
||||
? `/sessions/${friendlyId}/channels/${channel}/${io}`
|
||||
: `/sessions/${friendlyId}/${io}`;
|
||||
}
|
||||
|
||||
async initializeStream(
|
||||
@@ -170,11 +176,12 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
|
||||
*/
|
||||
async initializeSessionStream(
|
||||
friendlyId: string,
|
||||
io: "out" | "in"
|
||||
io: "out" | "in",
|
||||
channel?: string
|
||||
): Promise<{ responseHeaders?: Record<string, string> }> {
|
||||
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<number> {
|
||||
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<number> {
|
||||
@@ -259,9 +267,62 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
|
||||
async readSessionStreamRecords(
|
||||
friendlyId: string,
|
||||
io: "out" | "in",
|
||||
afterSeqNum?: number
|
||||
afterSeqNum?: number,
|
||||
channel?: string
|
||||
): Promise<StreamRecord[]> {
|
||||
return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io), afterSeqNum);
|
||||
return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io, channel), afterSeqNum);
|
||||
}
|
||||
|
||||
async listSessionChannels(friendlyId: string): Promise<string[]> {
|
||||
const prefix = `${this.streamPrefix}/sessions/${friendlyId}/channels/`;
|
||||
const names = await this.#s2ListStreamNames(prefix);
|
||||
const channels = new Set<string>();
|
||||
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<string[]> {
|
||||
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<StreamRecord[]> {
|
||||
@@ -402,9 +463,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
|
||||
friendlyId: string,
|
||||
io: "out" | "in",
|
||||
signal: AbortSignal,
|
||||
options?: StreamResponseOptions
|
||||
options?: StreamResponseOptions,
|
||||
channel?: string
|
||||
): Promise<Response> {
|
||||
const s2Stream = this.toSessionStreamName(friendlyId, io);
|
||||
const s2Stream = this.toSessionStreamName(friendlyId, io, channel);
|
||||
|
||||
let waitSeconds = options?.timeoutInSeconds ?? this.s2WaitSeconds;
|
||||
let settled = false;
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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<string>): 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;
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
},
|
||||
});
|
||||
```
|
||||
|
||||
<Note>
|
||||
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.
|
||||
</Note>
|
||||
|
||||
## 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<typeof screenshots>("screenshots", {
|
||||
sessionId,
|
||||
accessToken,
|
||||
io: "out",
|
||||
from: "latest",
|
||||
maxRecords: 1,
|
||||
});
|
||||
|
||||
const latest = records[0]; // ScreenshotFrame | undefined
|
||||
return latest ? <img src={latest.url} alt={`frame ${latest.step}`} /> : <p>Waiting…</p>;
|
||||
}
|
||||
```
|
||||
|
||||
`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.
|
||||
|
||||
<Warning>
|
||||
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.
|
||||
</Warning>
|
||||
|
||||
## 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` } },
|
||||
});
|
||||
```
|
||||
|
||||
<Note>
|
||||
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.
|
||||
</Note>
|
||||
|
||||
## Next steps
|
||||
|
||||
<CardGroup cols={2}>
|
||||
<Card title="Sessions" icon="layer-group" href="/ai-chat/sessions">
|
||||
The durable, cross-run primitive side channels are built on.
|
||||
</Card>
|
||||
<Card title="Read a session channel in React" icon="react" href="/realtime/react-hooks/session-stream">
|
||||
The `useSessionStream` hook `useSessionStreamChannel` mirrors.
|
||||
</Card>
|
||||
</CardGroup>
|
||||
@@ -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",
|
||||
|
||||
@@ -271,3 +271,40 @@ Close an agent chat conversation. The agent exits its loop gracefully. Without t
|
||||
<Callout type="warning">
|
||||
The `start_agent_chat`, `send_agent_message`, and `close_agent_chat` tools are write operations and are not available in readonly mode.
|
||||
</Callout>
|
||||
|
||||
## 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"`
|
||||
|
||||
<Callout type="warning">
|
||||
`write_session_channel` is a write operation and is not available in readonly mode.
|
||||
</Callout>
|
||||
|
||||
@@ -107,3 +107,20 @@ const { records, lastControl } = useSessionStream<string>(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<typeof screenshots>("screenshots", {
|
||||
sessionId,
|
||||
accessToken,
|
||||
io: "out",
|
||||
from: "latest",
|
||||
maxRecords: 1,
|
||||
});
|
||||
```
|
||||
|
||||
@@ -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.",
|
||||
},
|
||||
};
|
||||
|
||||
@@ -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,
|
||||
];
|
||||
|
||||
|
||||
@@ -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.`,
|
||||
},
|
||||
],
|
||||
};
|
||||
}),
|
||||
};
|
||||
@@ -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<AsyncIterableStream<T>> {
|
||||
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(),
|
||||
|
||||
@@ -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<readonly [string, string]>
|
||||
extraHeaders?: ReadonlyArray<readonly [string, string]>,
|
||||
channel?: string
|
||||
): Promise<StreamWriteResult> {
|
||||
const stream = await getS2Stream(apiClient, sessionId, io);
|
||||
const stream = await getS2Stream(apiClient, sessionId, io, channel);
|
||||
const headers: ReadonlyArray<readonly [string, string]> = [
|
||||
[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<void> {
|
||||
const stream = await getS2Stream(apiClient, sessionId, "out");
|
||||
const stream = await getS2Stream(apiClient, sessionId, "out", channel);
|
||||
await stream.append(AppendInput.create([AppendRecord.trim(earliestSeqNum)]));
|
||||
}
|
||||
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
export type SessionChannelShape = { in?: unknown; out?: unknown };
|
||||
|
||||
/**
|
||||
* A typed declaration of a named Session channel. The channel analogue of
|
||||
* `Task<TId, TIn, TOut>`: `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<string, SessionChannelShape>;
|
||||
|
||||
/** Extract a channel's literal name, the analogue of `TaskIdentifier`. */
|
||||
export type SessionChannelName<C extends AnySessionChannel> =
|
||||
C extends SessionChannel<infer N, any> ? N : never;
|
||||
|
||||
/** Extract the `.out` record type, the analogue of `TaskOutput`. */
|
||||
export type SessionChannelOut<C extends AnySessionChannel> =
|
||||
C extends SessionChannel<any, infer S> ? (S extends { out: infer O } ? O : unknown) : never;
|
||||
|
||||
/** Extract the `.in` record type, the analogue of `TaskPayload`. */
|
||||
export type SessionChannelIn<C extends AnySessionChannel> =
|
||||
C extends SessionChannel<any, infer S> ? (S extends { in: infer I } ? I : unknown) : never;
|
||||
@@ -36,110 +36,139 @@ export class SessionStreamsAPI implements SessionStreamManager {
|
||||
public on(
|
||||
sessionId: string,
|
||||
io: SessionChannelIO,
|
||||
handler: (data: unknown) => void | boolean | Promise<void>
|
||||
handler: (data: unknown) => void | boolean | Promise<void>,
|
||||
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<void>
|
||||
handler: (record: SessionStreamRecord) => void | boolean | Promise<void>,
|
||||
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<unknown> {
|
||||
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<SessionStreamRecord> {
|
||||
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<SessionStreamRecord> {
|
||||
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 {
|
||||
|
||||
@@ -70,6 +70,87 @@ function repeatingApiClient(record: {
|
||||
} as unknown as ApiClient;
|
||||
}
|
||||
|
||||
function channelAwareApiClient(
|
||||
byChannel: Record<string, Array<{ id: string; chunk: unknown; timestamp: number }>>
|
||||
): ApiClient {
|
||||
const delivered = new Set<string>();
|
||||
return {
|
||||
async subscribeToSessionStream<T>(
|
||||
_sessionIdOrExternalId: string,
|
||||
_io: "out" | "in",
|
||||
options?: {
|
||||
onPart?: (part: SSEStreamPart<T>) => 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<T>);
|
||||
}
|
||||
}
|
||||
const signal = options?.signal;
|
||||
// eslint-disable-next-line require-yield
|
||||
return (async function* () {
|
||||
if (signal?.aborted) return;
|
||||
await new Promise<void>((resolve) => {
|
||||
signal?.addEventListener("abort", () => resolve(), { once: true });
|
||||
});
|
||||
})() as unknown as Awaited<ReturnType<ApiClient["subscribeToSessionStream"]>>;
|
||||
},
|
||||
} 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;
|
||||
|
||||
@@ -47,8 +47,8 @@ type TailState = {
|
||||
promise: Promise<void>;
|
||||
};
|
||||
|
||||
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<unknown> {
|
||||
const recordPromise = this.onceRecord(sessionId, io, options);
|
||||
const recordPromise = this.onceRecord(sessionId, io, options, channel);
|
||||
return new InputStreamOncePromise<unknown>((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<SessionStreamRecord> {
|
||||
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<SessionStreamRecord> {
|
||||
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<SessionStreamRecord> {
|
||||
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<void> {
|
||||
const key = keyFor(sessionId, io);
|
||||
async #runTail(
|
||||
sessionId: string,
|
||||
io: SessionChannelIO,
|
||||
signal: AbortSignal,
|
||||
channel?: string
|
||||
): Promise<void> {
|
||||
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<unknown>(sessionId, io, {
|
||||
signal,
|
||||
baseUrl: this.baseUrl,
|
||||
channel,
|
||||
timeoutInSeconds: 600,
|
||||
lastEventId: lastSeq !== undefined ? String(lastSeq) : undefined,
|
||||
onPart: (part) => {
|
||||
|
||||
@@ -47,7 +47,8 @@ export interface SessionStreamManager {
|
||||
on(
|
||||
sessionId: string,
|
||||
io: SessionChannelIO,
|
||||
handler: (data: unknown) => void | boolean | Promise<void>
|
||||
handler: (data: unknown) => void | boolean | Promise<void>,
|
||||
channel?: string
|
||||
): { off: () => void };
|
||||
|
||||
/**
|
||||
@@ -57,21 +58,24 @@ export interface SessionStreamManager {
|
||||
onRecord?(
|
||||
sessionId: string,
|
||||
io: SessionChannelIO,
|
||||
handler: (record: SessionStreamRecord) => void | boolean | Promise<void>
|
||||
handler: (record: SessionStreamRecord) => void | boolean | Promise<void>,
|
||||
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<unknown>;
|
||||
|
||||
/** Wait for and consume the next record, including its durable metadata. */
|
||||
onceRecord?(
|
||||
sessionId: string,
|
||||
io: SessionChannelIO,
|
||||
options?: InputStreamOnceOptions
|
||||
options?: InputStreamOnceOptions,
|
||||
channel?: string
|
||||
): InputStreamOncePromise<SessionStreamRecord>;
|
||||
|
||||
/**
|
||||
@@ -83,23 +87,28 @@ export interface SessionStreamManager {
|
||||
sessionId: string,
|
||||
io: SessionChannelIO,
|
||||
predicate: SessionStreamRecordPredicate,
|
||||
options?: InputStreamOnceOptions
|
||||
options?: InputStreamOnceOptions,
|
||||
channel?: string
|
||||
): InputStreamOncePromise<SessionStreamRecord>;
|
||||
|
||||
/** 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;
|
||||
|
||||
@@ -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<TChannel extends AnySessionChannel, S extends "in" | "out"> = S extends "out"
|
||||
? SessionChannelOut<TChannel>
|
||||
: SessionChannelIn<TChannel>;
|
||||
|
||||
export type UseSessionStreamChannelInstance<TRecord> = {
|
||||
/** The records received so far on the channel, in arrival order. */
|
||||
records: Array<TRecord>;
|
||||
/** 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<SSEStreamPart<ChannelRecord<TChannel, S>>>) => 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<TChannel>`) 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<typeof screenshotsChannel>("screenshots", {
|
||||
* sessionId,
|
||||
* accessToken,
|
||||
* io: "out",
|
||||
* from: "latest",
|
||||
* maxRecords: 1,
|
||||
* });
|
||||
* ```
|
||||
*/
|
||||
export function useSessionStreamChannel<
|
||||
TChannel extends AnySessionChannel = AnySessionChannel,
|
||||
S extends "in" | "out" = "out",
|
||||
>(
|
||||
channel: SessionChannelName<TChannel>,
|
||||
options: UseSessionStreamChannelOptions<TChannel, S>
|
||||
): UseSessionStreamChannelInstance<ChannelRecord<TChannel, S>> {
|
||||
type TRecord = ChannelRecord<TChannel, S>;
|
||||
|
||||
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<TRecord>);
|
||||
|
||||
const { data: records, mutate: mutateRecords } = useSWR<Array<TRecord>>(
|
||||
[idKey, sessionId, channelName, io, "records"],
|
||||
null,
|
||||
{ fallbackData: initialRecordsFallback }
|
||||
);
|
||||
|
||||
const recordsRef = useRef<Array<TRecord>>(records ?? ([] as Array<TRecord>));
|
||||
useEffect(() => {
|
||||
recordsRef.current = records || ([] as Array<TRecord>);
|
||||
}, [records]);
|
||||
|
||||
const { data: lastEventId = undefined, mutate: setLastEventId } = useSWR<undefined | string>(
|
||||
[idKey, sessionId, channelName, io, "lastEventId"],
|
||||
null
|
||||
);
|
||||
const lastEventIdRef = useRef<string | undefined>(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<boolean>(
|
||||
[idKey, sessionId, channelName, io, "complete"],
|
||||
null
|
||||
);
|
||||
|
||||
const { data: error = undefined, mutate: setError } = useSWR<undefined | Error>(
|
||||
[idKey, sessionId, channelName, io, "error"],
|
||||
null
|
||||
);
|
||||
|
||||
const abortControllerRef = useRef<AbortController | null>(null);
|
||||
|
||||
const stop = useCallback(() => {
|
||||
if (abortControllerRef.current) {
|
||||
abortControllerRef.current.abort();
|
||||
abortControllerRef.current = null;
|
||||
}
|
||||
}, []);
|
||||
|
||||
const onRecordsCallback = options.onRecords;
|
||||
const onRecords = useCallback(
|
||||
(recordsBatch: Array<SSEStreamPart<TRecord>>) => {
|
||||
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<TRecord>(
|
||||
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<TRecord>(
|
||||
sessionIdOrExternalId: string,
|
||||
io: "out" | "in",
|
||||
channel: string,
|
||||
apiClient: ApiClient,
|
||||
mutateRecordsData: KeyedMutator<Array<TRecord>>,
|
||||
existingRecordsRef: React.MutableRefObject<Array<TRecord>>,
|
||||
setLastEventId: KeyedMutator<undefined | string>,
|
||||
setLastControl: KeyedMutator<undefined | ControlEvent>,
|
||||
onError: (e: Error) => void,
|
||||
onRecords: (records: Array<SSEStreamPart<TRecord>>) => void,
|
||||
onControl: (event: ControlEvent) => void,
|
||||
abortControllerRef: React.MutableRefObject<AbortController | null>,
|
||||
timeoutInSeconds?: number,
|
||||
lastEventId?: string,
|
||||
throttleInMs?: number,
|
||||
from?: "beginning" | "latest",
|
||||
maxRecords?: number
|
||||
) {
|
||||
let lastSeenEventId: string | undefined;
|
||||
let publishedEventId: string | undefined;
|
||||
let partsBatch: Array<SSEStreamPart<TRecord>> = [];
|
||||
|
||||
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<TRecord>(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<TRecord>(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;
|
||||
}
|
||||
}
|
||||
@@ -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";
|
||||
|
||||
@@ -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: <C extends AnySessionChannel = AnySessionChannel>(
|
||||
channel: SessionChannelName<C> | C
|
||||
): SessionChannelHandleFor<C> => getChatSession().channel<C>(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}. */
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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<C extends AnySessionChannel = AnySessionChannel>(
|
||||
channel: SessionChannelName<C> | C
|
||||
): SessionChannelHandleFor<C> {
|
||||
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<C>;
|
||||
}
|
||||
}
|
||||
|
||||
export type SessionChannelHandleFor<C extends AnySessionChannel> = {
|
||||
readonly name: string;
|
||||
readonly out: SessionOutputChannel<SessionChannelOut<C>>;
|
||||
readonly in: SessionInputChannel<SessionChannelIn<C>>;
|
||||
};
|
||||
|
||||
export type SessionChannelHandle = SessionChannelHandleFor<AnySessionChannel>;
|
||||
|
||||
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<typeof channel>` so the record types line up
|
||||
* on every side.
|
||||
*/
|
||||
function defineChannel<
|
||||
TShape extends SessionChannelShape = SessionChannelShape,
|
||||
const TName extends string = string,
|
||||
>(name: TName): SessionChannel<TName, TShape> {
|
||||
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<PipeStreamOptions, "target">;
|
||||
* 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<TOut = unknown> {
|
||||
// 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<InitializeSessionStreamResponseLike>;
|
||||
|
||||
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<T>(value: T, options?: SessionPipeStreamOptions): Promise<void> {
|
||||
const { waitUntilComplete } = this.writer<T>({
|
||||
async append(value: TOut, options?: SessionPipeStreamOptions): Promise<void> {
|
||||
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<T>(
|
||||
pipe<T = TOut>(
|
||||
value: AsyncIterable<T> | ReadableStream<T>,
|
||||
options?: SessionPipeStreamOptions
|
||||
): PipeStreamResult<T> {
|
||||
@@ -331,7 +394,7 @@ export class SessionOutputChannel {
|
||||
* stream and await completion. Span is collapsible via `options.spanName`
|
||||
* / `options.collapsed`.
|
||||
*/
|
||||
writer<T>(options: WriterStreamOptions<T>): PipeStreamResult<T> {
|
||||
writer<T = TOut>(options: WriterStreamOptions<T>): PipeStreamResult<T> {
|
||||
let controller!: ReadableStreamDefaultController<T>;
|
||||
const ongoingStreamPromises: Promise<void>[] = [];
|
||||
|
||||
@@ -407,11 +470,12 @@ export class SessionOutputChannel {
|
||||
* shared {@link SSEStreamSubscription} plumbing used by run-scoped
|
||||
* realtime streams.
|
||||
*/
|
||||
async read<T = unknown>(options?: SessionSubscribeOptions<T>): Promise<AsyncIterableStream<T>> {
|
||||
async read<T = TOut>(options?: SessionSubscribeOptions<T>): Promise<AsyncIterableStream<T>> {
|
||||
const apiClient = apiClientManager.clientOrThrow();
|
||||
|
||||
return apiClient.subscribeToSessionStream<T>(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<readonly [string, string]>
|
||||
): Promise<StreamWriteResult> {
|
||||
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<void> {
|
||||
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<TIn = unknown> {
|
||||
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<void> {
|
||||
async send(value: TIn, requestOptions?: ApiRequestOptions): Promise<void> {
|
||||
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<T = unknown>(handler: (data: T) => void | boolean | Promise<void>): { off: () => void } {
|
||||
on<T = TIn>(handler: (data: T) => void | boolean | Promise<void>): { off: () => void } {
|
||||
return sessionStreams.on(
|
||||
this.sessionId,
|
||||
"in",
|
||||
handler as (data: unknown) => void | boolean | Promise<void>
|
||||
handler as (data: unknown) => void | boolean | Promise<void>,
|
||||
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<T = unknown>(options?: InputStreamOnceOptions): InputStreamOncePromise<T> {
|
||||
once<T = TIn>(options?: InputStreamOnceOptions): InputStreamOncePromise<T> {
|
||||
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<T>((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 = unknown>(): T | undefined {
|
||||
return sessionStreams.peek(this.sessionId, "in") as T | undefined;
|
||||
peek<T = TIn>(): 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<T = unknown>(options?: InputStreamWaitOptions): ManualWaitpointPromise<T> {
|
||||
return new ManualWaitpointPromise<T>(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<T = unknown>(
|
||||
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 =
|
||||
|
||||
Reference in New Issue
Block a user