import { json } from "@remix-run/server-runtime"; import { CreateSessionRequestBody, type CreatedSessionResponseBody, ListSessionsQueryParams, type ListSessionsResponseBody, type SessionItem, type SessionStatus, } from "@trigger.dev/core/v3"; import { SessionId } from "@trigger.dev/core/v3/isomorphic"; import type { Prisma, Session } from "@trigger.dev/database"; 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 { ensureRunForSession, type SessionTriggerConfig, } from "~/services/realtime/sessionRunManager.server"; import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server"; import { serializeSession, serializeSessionsWithFriendlyRunIds, } from "~/services/realtime/sessions.server"; import { SessionsRepository } from "~/services/sessionsRepository/sessionsRepository.server"; import { anyResource, createActionApiRoute, createLoaderApiRoute, } from "~/services/routeBuilders/apiBuilder.server"; import { ServiceValidationError } from "~/v3/services/common.server"; import { runStore } from "~/v3/runStore.server"; function asArray(value: T | T[] | undefined): T[] | undefined { if (value === undefined) return undefined; return Array.isArray(value) ? value : [value]; } export const loader = createLoaderApiRoute( { searchParams: ListSessionsQueryParams, allowJWT: true, corsStrategy: "all", authorization: { action: "read", // Multi-key resource preserves the pre-RBAC superScope semantics: // - Per-task scoping via `read:tasks:` matches a task element // - Type-level `read:sessions` (the old superScope) matches the // sessions element (collection-level — no id) // - `read:all` / `admin` bypass via the JWT ability's wildcard branches // The taskIdentifier filter accepts a string or an array; expand to // one resource per task id so any per-task-scoped JWT among them // grants access (the array gets OR semantics). resource: (_, __, searchParams) => { const taskFilter = asArray(searchParams["filter[taskIdentifier]"]) ?? []; return anyResource([ ...taskFilter.map((id) => ({ type: "tasks" as const, id })), { type: "sessions" as const }, ]); }, }, findResource: async () => 1, }, async ({ searchParams, authentication }) => { const clickhouse = await clickhouseFactory.getClickhouseForOrganization( authentication.environment.organizationId, "standard" ); const repository = new SessionsRepository({ clickhouse, prisma: $replica as PrismaClient, }); // `page[after]` is the forward cursor, `page[before]` is the backward // cursor. The repository internally keys off `{cursor, direction}`. const cursor = searchParams["page[after]"] ?? searchParams["page[before]"]; const direction = searchParams["page[before]"] ? "backward" : "forward"; const { sessions: rows, pagination } = await repository.listSessions({ organizationId: authentication.environment.organizationId, projectId: authentication.environment.projectId, environmentId: authentication.environment.id, types: asArray(searchParams["filter[type]"]), tags: asArray(searchParams["filter[tags]"]), taskIdentifiers: asArray(searchParams["filter[taskIdentifier]"]), externalId: searchParams["filter[externalId]"], statuses: asArray(searchParams["filter[status]"]) as SessionStatus[] | undefined, period: searchParams["filter[createdAt][period]"], from: searchParams["filter[createdAt][from]"], to: searchParams["filter[createdAt][to]"], page: { size: searchParams["page[size]"], cursor, direction, }, }); // Batched friendlyId translation: `currentRunId` on the wire is the // public `run_*` form, matching the single-session routes. One `IN` // lookup per page. const data = await serializeSessionsWithFriendlyRunIds( rows.map( (row) => ({ ...row, // Columns the list query doesn't select — filled so the // serializer can operate on a narrowed payload without type errors. projectId: authentication.environment.projectId, environmentType: authentication.environment.type, organizationId: authentication.environment.organizationId, }) as Session ), { projectId: authentication.environment.projectId, runtimeEnvironmentId: authentication.environment.id, } ); return json({ data, pagination: { ...(pagination.nextCursor ? { next: pagination.nextCursor } : {}), ...(pagination.previousCursor ? { previous: pagination.previousCursor } : {}), }, }); } ); const { action } = createActionApiRoute( { body: CreateSessionRequestBody, method: "POST", maxContentLength: 1024 * 32, // 32KB — metadata is the only thing that grows // Customer's server (typically wrapping // `chat.createStartSessionAction`) owns session creation so any // authorization decision (per-user/plan/quota) sits server-side // alongside whatever DB write the customer pairs with the create. // The session-scoped PAT returned in the response body is what the // browser uses thereafter against `.in/append`, `.out` SSE, // `end-and-continue`, etc. // // JWT is allowed when the caller holds an explicit `write:sessions` / // `admin` super-scope plus a `tasks:` scope — gates // server-side surfaces like the cli-v3 MCP from creating sessions on // behalf of the developer without weakening the browser model. allowJWT: true, authorization: { // Per-task scoping via `body.taskIdentifier` (action-route resource // callbacks receive the parsed body as the 4th arg — see // `apiBuilder.server.ts:710`). A JWT scoped only to `write:tasks:foo` // can only create sessions whose `taskIdentifier` is `"foo"`. // // Multi-key resource: pre-RBAC this route had a `superScopes: // ["write:sessions", "admin"]` whitelist; post-RBAC the equivalent // is the `{ type: "sessions" }` element below — a `write:sessions` // JWT (no id) matches it directly, deliberately bypassing the // per-task check exactly as before. `admin` / `write:all` bypass // via the JWT ability's wildcard branches. action: "write", resource: (_params, _searchParams, _headers, body) => anyResource([{ type: "tasks", id: body.taskIdentifier }, { type: "sessions" }]), }, corsStrategy: "all", }, async ({ authentication, body }) => { try { const { id, friendlyId } = SessionId.generate(); // Idempotent on (env, externalId): two concurrent POSTs converge // to the same row. We refresh `triggerConfig` on the cached path // so newly-deployed schema changes (e.g. an updated // `clientDataSchema` on the agent) propagate to subsequent runs // — the next `ensureRunForSession` reads back the latest config. let session: Session; let isCached = false; const triggerConfigJson = body.triggerConfig as unknown as Prisma.InputJsonValue; if (body.externalId) { session = await prisma.session.upsert({ where: { runtimeEnvironmentId_externalId: { runtimeEnvironmentId: authentication.environment.id, externalId: body.externalId, }, }, create: { id, friendlyId, externalId: body.externalId, type: body.type, taskIdentifier: body.taskIdentifier, triggerConfig: triggerConfigJson, tags: body.tags ?? [], metadata: body.metadata as Prisma.InputJsonValue | undefined, expiresAt: body.expiresAt ?? null, projectId: authentication.environment.projectId, runtimeEnvironmentId: authentication.environment.id, environmentType: authentication.environment.type, organizationId: authentication.environment.organizationId, streamBasinName: authentication.environment.organization.streamBasinName, chatSnapshotStoragePath: chatSnapshotStoragePathForSession(friendlyId), }, update: { triggerConfig: triggerConfigJson }, }); isCached = session.id !== id; } else { session = await prisma.session.create({ data: { id, friendlyId, type: body.type, taskIdentifier: body.taskIdentifier, triggerConfig: triggerConfigJson, tags: body.tags ?? [], metadata: body.metadata as Prisma.InputJsonValue | undefined, expiresAt: body.expiresAt ?? null, projectId: authentication.environment.projectId, runtimeEnvironmentId: authentication.environment.id, environmentType: authentication.environment.type, organizationId: authentication.environment.organizationId, streamBasinName: authentication.environment.organization.streamBasinName, chatSnapshotStoragePath: chatSnapshotStoragePathForSession(friendlyId), }, }); } // Reject create on a closed session. The upsert path will return // an already-closed row when the caller reuses an externalId, and // without this guard `ensureRunForSession` would trigger a fresh // run that can't receive `.in` input (the append handler 409s on // closed sessions). Force the caller to use a different externalId // — `close` is one-way. if (session.closedAt) { return json( { error: "Session is closed; use a different externalId to create a new session" }, { status: 409 } ); } // Same guard as the append / end-and-continue handlers: an expired // row must not spawn a run, because every subsequent `.in/append` // would 400 on the expiry check — a run boots but the chat can // never receive input. if (session.expiresAt && session.expiresAt.getTime() < Date.now()) { return json( { error: "Session is expired; use a different externalId to create a new session" }, { status: 409 } ); } // Session is task-bound — every session has a live run by // construction. `ensureRunForSession` is idempotent: on the // cached path it sees `currentRunId` is alive and returns it // without re-triggering. const ensureResult = await ensureRunForSession({ session, environment: authentication.environment, reason: isCached ? "continuation" : "initial", }); // Read-after-write: the run was just triggered in this request, // so go to the writer rather than $replica. Replica lag here // would null this out and turn a successful create into a 500. const run = await runStore.findRun( { id: ensureResult.runId }, { select: { friendlyId: true } }, prisma ); if (!run) { throw new Error(`Triggered run ${ensureResult.runId} not found`); } // Mint a session-scoped PAT keyed on the addressing string the // transport will use everywhere (`.in/append`, `.out` SSE, // `end-and-continue`). For sessions with an externalId, that's // the externalId; otherwise the friendlyId. Mirrors the // canonical addressing key used server-side. const addressingKey = session.externalId ?? session.friendlyId; const publicAccessToken = await mintSessionToken(authentication.environment, addressingKey); const sessionItem: SessionItem = { ...serializeSession(session), triggerConfig: session.triggerConfig as unknown as SessionTriggerConfig, currentRunId: run.friendlyId, }; const responseBody: CreatedSessionResponseBody = { ...sessionItem, runId: run.friendlyId, publicAccessToken, isCached, }; return json(responseBody, { status: isCached ? 200 : 201, }); } catch (error) { if (error instanceof ServiceValidationError) { return json({ error: error.message }, { status: 422 }); } logger.error("Failed to create session", { error }); return json({ error: "Something went wrong" }, { status: 500 }); } } ); export { action };