diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index 99094d3ff..02a32ab18 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/packages/core/src/v3/apiClient/index.ts @@ -13,8 +13,16 @@ import { BatchTriggerTaskV3RequestBody, BatchTriggerTaskV3Response, CanceledRunResponse, + CloseSessionRequestBody, CompleteWaitpointTokenRequestBody, CompleteWaitpointTokenResponseBody, + CreatedSessionResponseBody, + CreateSessionRequestBody, + ListSessionsOptions, + ListSessionsResponseBody, + ListedSessionItem, + RetrieveSessionResponseBody, + UpdateSessionRequestBody, CreateBatchRequestBody, CreateBatchResponse, CreateEnvironmentVariableRequestBody, @@ -1095,6 +1103,182 @@ export class ApiClient { ); } + // ======================================================================== + // Sessions + // ======================================================================== + + createSession(body: CreateSessionRequestBody, requestOptions?: ZodFetchOptions) { + return zodfetch( + CreatedSessionResponseBody, + `${this.baseUrl}/api/v1/sessions`, + { + method: "POST", + headers: this.#getHeaders(false), + body: JSON.stringify(body), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + + retrieveSession(sessionIdOrExternalId: string, requestOptions?: ZodFetchOptions) { + return zodfetch( + RetrieveSessionResponseBody, + `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}`, + { + method: "GET", + headers: this.#getHeaders(false), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + + updateSession( + sessionIdOrExternalId: string, + body: UpdateSessionRequestBody, + requestOptions?: ZodFetchOptions + ) { + return zodfetch( + RetrieveSessionResponseBody, + `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}`, + { + method: "PATCH", + headers: this.#getHeaders(false), + body: JSON.stringify(body), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + + closeSession( + sessionIdOrExternalId: string, + body?: CloseSessionRequestBody, + requestOptions?: ZodFetchOptions + ) { + return zodfetch( + RetrieveSessionResponseBody, + `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/close`, + { + method: "POST", + headers: this.#getHeaders(false), + body: JSON.stringify(body ?? {}), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + + listSessions( + options?: ListSessionsOptions, + requestOptions?: ZodFetchOptions + ): CursorPagePromise { + const searchParams = createSearchQueryForListSessions(options); + + return zodfetchCursorPage( + ListedSessionItem, + `${this.baseUrl}/api/v1/sessions`, + { + query: searchParams, + limit: options?.limit, + after: options?.after, + before: options?.before, + }, + { + method: "GET", + headers: this.#getHeaders(false), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + + // ======================================================================== + // Session realtime channels + // ======================================================================== + + async initializeSessionStream( + sessionIdOrExternalId: string, + io: "out" | "in", + requestOptions?: ZodFetchOptions + ) { + // 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`. + return zodfetch( + CreateStreamResponseBody, + `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}`, + { + method: "PUT", + headers: this.#getHeaders(false), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ) + .withResponse() + .then(({ data, response }) => ({ + ...data, + headers: Object.fromEntries(response.headers.entries()), + })); + } + + async appendToSessionStream( + sessionIdOrExternalId: string, + io: "out" | "in", + part: TBody, + requestOptions?: ZodFetchOptions + ) { + return zodfetch( + AppendToStreamResponseBody, + `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}/append`, + { + method: "POST", + headers: this.#getHeaders(false), + body: part, + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + + /** + * Subscribe to SSE records on a Session channel. Reuses the same + * {@link SSEStreamSubscription} plumbing as `readStream` for run-scoped + * realtime streams — auto-retry, Last-Event-ID resume, abort-on-cancel. + */ + async subscribeToSessionStream( + sessionIdOrExternalId: string, + io: "out" | "in", + options?: { + signal?: AbortSignal; + baseUrl?: string; + timeoutInSeconds?: number; + onComplete?: () => void; + onError?: (error: Error) => void; + lastEventId?: string; + onPart?: (part: SSEStreamPart) => void; + } + ): Promise> { + const url = `${options?.baseUrl ?? this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}`; + + const subscription = new SSEStreamSubscription(url, { + headers: this.getHeaders(), + signal: options?.signal, + onComplete: options?.onComplete, + onError: options?.onError, + timeoutInSeconds: options?.timeoutInSeconds, + lastEventId: options?.lastEventId, + }); + + const stream = await subscription.subscribe(); + const onPart = options?.onPart; + + return stream.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + const data = chunk.chunk as T; + onPart?.(chunk as SSEStreamPart); + controller.enqueue(data); + }, + }) + ); + } + async waitForDuration( runId: string, body: WaitForDurationRequestBody, @@ -1836,6 +2020,47 @@ function queueNameFromQueueTypeName(queue: QueueTypeName): string { return queue.name; } +function createSearchQueryForListSessions(options?: ListSessionsOptions): URLSearchParams { + const searchParams = new URLSearchParams(); + + if (!options) return searchParams; + + const appendMany = (name: string, value: string | string[] | undefined) => { + if (value === undefined) return; + searchParams.append(name, Array.isArray(value) ? value.join(",") : value); + }; + + appendMany("filter[type]", options.type); + appendMany("filter[tags]", options.tag); + appendMany("filter[taskIdentifier]", options.taskIdentifier); + + if (options.externalId) { + searchParams.append("filter[externalId]", options.externalId); + } + + appendMany("filter[status]", options.status as string | string[] | undefined); + + if (options.period) { + searchParams.append("filter[createdAt][period]", options.period); + } + + if (options.from !== undefined) { + searchParams.append( + "filter[createdAt][from]", + options.from instanceof Date ? options.from.getTime().toString() : options.from.toString() + ); + } + + if (options.to !== undefined) { + searchParams.append( + "filter[createdAt][to]", + options.to instanceof Date ? options.to.getTime().toString() : options.to.toString() + ); + } + + return searchParams; +} + function createSearchQueryForListWaitpointTokens( query?: ListWaitpointTokensQueryParams ): URLSearchParams { diff --git a/packages/trigger-sdk/src/v3/auth.ts b/packages/trigger-sdk/src/v3/auth.ts index 1f2df463b..16de798b0 100644 --- a/packages/trigger-sdk/src/v3/auth.ts +++ b/packages/trigger-sdk/src/v3/auth.ts @@ -67,6 +67,17 @@ type PublicTokenPermissionProperties = { * Grant access to send data to input streams on specific runs */ inputStreams?: string | string[]; + + /** + * Grant access to specific Sessions (the durable, typed I/O primitive that + * outlives a single run). Use the session's friendlyId (e.g. `session_abc`). + * + * `read:sessions:{id}` lets the bearer read both the `.out` and `.in` + * channels and list runs on the session. `write:sessions:{id}` lets the + * bearer append to the session's channels. `trigger:sessions:{id}` permits + * triggering new runs on the session. + */ + sessions?: string | string[]; }; export type PublicTokenPermissions = { diff --git a/packages/trigger-sdk/src/v3/index.ts b/packages/trigger-sdk/src/v3/index.ts index 2e0f883f8..66a8b1313 100644 --- a/packages/trigger-sdk/src/v3/index.ts +++ b/packages/trigger-sdk/src/v3/index.ts @@ -17,6 +17,7 @@ export * from "./otel.js"; export * from "./schemas.js"; export * from "./heartbeats.js"; export * from "./streams.js"; +export * from "./sessions.js"; export * from "./query.js"; export type { Context }; diff --git a/packages/trigger-sdk/src/v3/sessions.ts b/packages/trigger-sdk/src/v3/sessions.ts new file mode 100644 index 000000000..cfb4fc991 --- /dev/null +++ b/packages/trigger-sdk/src/v3/sessions.ts @@ -0,0 +1,318 @@ +import type { + ApiPromise, + ApiRequestOptions, + AsyncIterableStream, + CloseSessionRequestBody, + CreatedSessionResponseBody, + CreateSessionRequestBody, + ListSessionsOptions, + ListedSessionItem, + RetrieveSessionResponseBody, + UpdateSessionRequestBody, +} from "@trigger.dev/core/v3"; +import { + CursorPagePromise, + accessoryAttributes, + apiClientManager, + mergeRequestOptions, +} from "@trigger.dev/core/v3"; +import { tracer } from "./tracer.js"; + +export type { + CreatedSessionResponseBody, + CreateSessionRequestBody, + CloseSessionRequestBody, + ListSessionsOptions, + ListedSessionItem, + RetrieveSessionResponseBody, + UpdateSessionRequestBody, +}; + +export const sessions = { + create: createSession, + retrieve: retrieveSession, + update: updateSession, + close: closeSession, + list: listSessions, + open, +}; + +/** + * Create a {@link Session} — a durable, typed, bidirectional I/O primitive + * that outlives a single run. Idempotent via `externalId`. + */ +function createSession( + body: CreateSessionRequestBody, + requestOptions?: ApiRequestOptions +): ApiPromise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "sessions.create()", + icon: "sessions", + attributes: sessionAttributes(body.externalId ?? body.type, { + type: body.type, + ...(body.externalId ? { externalId: body.externalId } : {}), + }), + }, + requestOptions + ); + + return apiClient.createSession(body, $requestOptions); +} + +/** + * Retrieve a Session by `friendlyId` (`session_*`) or user-supplied + * `externalId`. The server disambiguates via the `session_` prefix. + */ +function retrieveSession( + sessionIdOrExternalId: string, + requestOptions?: ApiRequestOptions +): ApiPromise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "sessions.retrieve()", + icon: "sessions", + attributes: sessionAttributes(sessionIdOrExternalId), + }, + requestOptions + ); + + return apiClient.retrieveSession(sessionIdOrExternalId, $requestOptions); +} + +/** Update mutable fields on a Session (tags, metadata, externalId). */ +function updateSession( + sessionIdOrExternalId: string, + body: UpdateSessionRequestBody, + requestOptions?: ApiRequestOptions +): ApiPromise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "sessions.update()", + icon: "sessions", + attributes: sessionAttributes(sessionIdOrExternalId), + }, + requestOptions + ); + + return apiClient.updateSession(sessionIdOrExternalId, body, $requestOptions); +} + +/** Mark a Session as closed (terminal, idempotent). */ +function closeSession( + sessionIdOrExternalId: string, + body?: CloseSessionRequestBody, + requestOptions?: ApiRequestOptions +): ApiPromise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "sessions.close()", + icon: "sessions", + attributes: sessionAttributes(sessionIdOrExternalId, { + ...(body?.reason ? { reason: body.reason } : {}), + }), + }, + requestOptions + ); + + return apiClient.closeSession(sessionIdOrExternalId, body, $requestOptions); +} + +/** + * List Sessions in the current environment with filters + cursor pagination. + * Returns a {@link CursorPagePromise} so callers can iterate pages with + * `for await`. + */ +function listSessions( + options?: ListSessionsOptions, + requestOptions?: ApiRequestOptions +): CursorPagePromise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "sessions.list()", + icon: "sessions", + attributes: { + ...(options?.type ? { type: toAttr(options.type) } : {}), + ...(options?.tag ? { tag: toAttr(options.tag) } : {}), + ...(options?.status ? { status: toAttr(options.status) } : {}), + ...(options?.externalId ? { externalId: options.externalId } : {}), + }, + }, + requestOptions + ); + + return apiClient.listSessions(options, $requestOptions); +} + +/** + * Open a lightweight handle to a Session's realtime channels. Does not + * perform a network call on its own — each channel method hits the + * corresponding realtime endpoint. + */ +function open(sessionIdOrExternalId: string): SessionHandle { + return new SessionHandle(sessionIdOrExternalId); +} + +export class SessionHandle { + public readonly out: SessionChannel; + public readonly in: SessionChannel; + + constructor(public readonly id: string) { + this.out = new SessionChannel(id, "out"); + this.in = new SessionChannel(id, "in"); + } +} + +export type SessionChannelCredentials = { + accessToken: string; + basin: string; + streamName: string; + endpoint?: string; + flushIntervalMs?: number; + maxRetries?: number; +}; + +/** One direction of a Session's bidirectional channel pair. */ +export class SessionChannel { + constructor( + public readonly sessionId: string, + public readonly io: "out" | "in" + ) {} + + /** + * Append a single record to this channel via the server-side append + * endpoint. For high-throughput writes use {@link initialize} to get S2 + * credentials and write directly to S2. + */ + async append(part: string | unknown, requestOptions?: ApiRequestOptions): Promise { + const apiClient = apiClientManager.clientOrThrow(); + const body = typeof part === "string" ? part : JSON.stringify(part); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: `sessions.open(${this.sessionId}).${this.io}.append()`, + icon: "sessions", + attributes: sessionAttributes(this.sessionId, { io: this.io }), + }, + requestOptions + ); + + await apiClient.appendToSessionStream(this.sessionId, this.io, body, $requestOptions); + } + + /** + * Friendly alias for `channel.append(value)` used on the `.in` channel by + * clients producing messages for the task runtime. + */ + send(value: unknown, requestOptions?: ApiRequestOptions): Promise { + return this.append(value, requestOptions); + } + + /** + * Subscribe to SSE records on this channel. Delegates to the shared + * {@link SSEStreamSubscription} plumbing (auto-retry, Last-Event-ID + * resume, abort propagation) used by run-scoped realtime streams — + * session subscribers get the same guarantees. + */ + async subscribe( + options?: SessionSubscribeOptions + ): Promise> { + const apiClient = apiClientManager.clientOrThrow(); + + return apiClient.subscribeToSessionStream(this.sessionId, this.io, { + signal: options?.signal, + timeoutInSeconds: options?.timeoutInSeconds, + lastEventId: + options?.lastEventId != null ? String(options.lastEventId) : undefined, + onPart: options?.onPart, + onComplete: options?.onComplete, + onError: options?.onError, + }); + } + + /** + * Fetch S2 credentials for direct-to-S2 writes. Returns the same header + * bag the server hands to {@link StreamsWriterV2}. + */ + async initialize(requestOptions?: ApiRequestOptions): Promise { + const apiClient = apiClientManager.clientOrThrow(); + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: `sessions.open(${this.sessionId}).${this.io}.initialize()`, + icon: "sessions", + attributes: sessionAttributes(this.sessionId, { io: this.io }), + }, + requestOptions + ); + + const response = await apiClient.initializeSessionStream( + this.sessionId, + this.io, + $requestOptions + ); + + return { + accessToken: response.headers["x-s2-access-token"] ?? "", + basin: response.headers["x-s2-basin"] ?? "", + streamName: response.headers["x-s2-stream-name"] ?? "", + endpoint: response.headers["x-s2-endpoint"], + flushIntervalMs: numHeader(response.headers, "x-s2-flush-interval-ms"), + maxRetries: numHeader(response.headers, "x-s2-max-retries"), + }; + } +} + +export type SessionSubscribeOptions = { + signal?: AbortSignal; + lastEventId?: string | number; + /** Timeout in seconds for the underlying long-poll (max 600). */ + timeoutInSeconds?: number; + /** Called for each SSE event with the full event metadata (id, timestamp). */ + onPart?: (part: { id: string; chunk: T; timestamp: number }) => void; + /** Called when the server signals end-of-stream. */ + onComplete?: () => void; + /** Called on unrecoverable errors after the retry budget is exhausted. */ + onError?: (error: Error) => void; +}; + +// ─── helpers ──────────────────────────────────────────────────────── + +function sessionAttributes(id: string, extra?: Record) { + return { + session: id, + ...(extra ?? {}), + ...accessoryAttributes({ + items: [{ text: id, variant: "normal" }], + style: "codepath", + }), + }; +} + +function toAttr(value: string | string[]): string { + return Array.isArray(value) ? value.join(",") : value; +} + +function numHeader(headers: Record, name: string): number | undefined { + const raw = headers[name]; + if (raw == null) return undefined; + const parsed = Number(raw); + return Number.isFinite(parsed) ? parsed : undefined; +} diff --git a/references/hello-world/src/trigger/sessionsSmoke.ts b/references/hello-world/src/trigger/sessionsSmoke.ts new file mode 100644 index 000000000..a3ce1f574 --- /dev/null +++ b/references/hello-world/src/trigger/sessionsSmoke.ts @@ -0,0 +1,159 @@ +import { logger, sessions, task } from "@trigger.dev/sdk"; + +/** + * End-to-end smoke test for the Session SDK and server routes. + * + * Exercises every code path: + * - control-plane CRUD (create / retrieve / update / close) + * - polymorphic lookup via friendlyId and externalId + * - list with tag / type / externalId filters + * - cursor pagination (page 1 -> page 2) + * - realtime `.out` initialize + append + subscribe (SSE round-trip) + * - realtime `.in` send + * - idempotent close + * + * Trigger from the dashboard or via the MCP `trigger_task` tool: + * + * mcp__trigger__trigger_task(taskId: "sessions-smoke", payload: {}) + * + * Inside a run, the SDK picks up the ambient environment credentials, so + * no `configure()` call is needed. + */ +export const sessionsSmoke = task({ + id: "sessions-smoke", + run: async () => { + const runId = Date.now(); + const results: Record = {}; + + logger.info("sessions.create"); + const created = await sessions.create({ + type: "chat.agent", + externalId: `smoke-${runId}`, + tags: ["smoketest", "sdk"], + metadata: { purpose: "session-smoketest", runId }, + }); + results.created = { id: created.id, isCached: created.isCached }; + + logger.info("sessions.retrieve by friendlyId"); + const byId = await sessions.retrieve(created.id); + results.retrievedByFriendlyId = byId.externalId; + + logger.info("sessions.retrieve by externalId (polymorphic)"); + const byExt = await sessions.retrieve(created.externalId!); + results.retrievedByExternalId = byExt.id; + + logger.info("sessions.update tags + metadata"); + const updated = await sessions.update(created.id, { + tags: ["smoketest", "sdk", "updated"], + metadata: { purpose: "session-smoketest", runId, touched: true }, + }); + results.updated = { + tags: updated.tags, + touched: (updated.metadata as Record | null)?.touched, + }; + + const handle = sessions.open(created.externalId!); + + logger.info("sessions.open(...).out.initialize (S2 creds)"); + const outCreds = await handle.out.initialize(); + results.outInitialize = { basin: outCreds.basin, streamName: outCreds.streamName }; + + logger.info("sessions.open(...).out.append x2 + .in.send x1"); + await handle.out.append({ chunk: "first", ts: Date.now() }); + await handle.out.append({ chunk: "second", ts: Date.now() }); + await handle.in.send({ role: "user", content: "hello from smoketest" }); + + logger.info("sessions.open(...).out.subscribe (SSE round-trip)"); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), 4000); + const received: Array<{ id: string; chunk: unknown }> = []; + try { + const stream = await handle.out.subscribe({ + signal: controller.signal, + timeoutInSeconds: 3, + onPart: (part) => { + received.push({ id: part.id, chunk: part.chunk }); + }, + }); + let count = 0; + for await (const _chunk of stream) { + count += 1; + if (count >= 2) break; + } + } catch (err) { + if ((err as Error).name !== "AbortError") throw err; + } finally { + clearTimeout(timer); + } + results.subscribed = received; + + // Seed a couple extra sessions so list queries have multiple hits. + await sessions.create({ + type: "chat.agent", + externalId: `smoke-${runId}-b`, + tags: ["smoketest"], + }); + await sessions.create({ + type: "run.output", + externalId: `smoke-${runId}-c`, + tags: ["smoketest"], + }); + + // Let ClickHouse replication catch up. + await new Promise((resolve) => setTimeout(resolve, 1500)); + + logger.info("sessions.list by tag"); + const listedAll = await sessions.list({ tag: "smoketest", limit: 50 }); + results.listByTag = listedAll.data.length; + + logger.info("sessions.list type + tag"); + const listedChat = await sessions.list({ + type: "chat.agent", + tag: "smoketest", + limit: 50, + }); + results.listByTypeAndTag = { + count: listedChat.data.length, + types: [...new Set(listedChat.data.map((session) => session.type))], + }; + + logger.info("sessions.list by externalId"); + const listedOne = await sessions.list({ externalId: `smoke-${runId}` }); + results.listByExternalId = { + count: listedOne.data.length, + match: listedOne.data[0]?.id === created.id, + }; + + logger.info("sessions.list pagination"); + const page1 = await sessions.list({ tag: "smoketest", limit: 2 }); + let page2Ids: string[] = []; + if (page1.pagination.next) { + const page2 = await sessions.list({ + tag: "smoketest", + limit: 2, + after: page1.pagination.next, + }); + page2Ids = page2.data.map((s) => s.id); + } + results.pagination = { + page1Ids: page1.data.map((s) => s.id), + next: page1.pagination.next, + page2Ids, + }; + + logger.info("sessions.close"); + const closed = await sessions.close(created.externalId!, { reason: "smoketest-done" }); + results.closed = { closedAt: closed.closedAt, reason: closed.closedReason }; + + logger.info("sessions.close (idempotent)"); + const reclosed = await sessions.close(created.externalId!, { + reason: "should-not-clobber", + }); + results.idempotentClose = { + closedAtUnchanged: reclosed.closedAt?.toString() === closed.closedAt?.toString(), + reasonUnchanged: reclosed.closedReason === closed.closedReason, + }; + + return { ok: true, runId, results }; + }, +});