feat: Sessions - bidirectional durable agent streams (#3417)

> ⚠️ **Not released yet.** This PR is the server-side foundation only.
The SDK changes that customers will actually use (`chat.agent`
migration, `chat.createStartSessionAction`, `useTriggerChatTransport`
updates) live on a separate branch and ship together in an upcoming
`@trigger.dev/sdk` prerelease. Until that prerelease is published, this
surface is reachable only via direct HTTP.

## What this gives Trigger.dev users

A new first-class primitive, **Session**, for durable, task-bound,
bidirectional I/O that outlives any single run. Sessions are the run
manager for `chat.agent` going forward, and they unblock anything else
that needs "one identifier, many runs over time" with a stable channel
pair the client can write to and subscribe to.

### Use cases unblocked

- **Chat agents that persist across many runs.** One session per chat
(keyed on your own `chatId` via `externalId`), turns 1..N attach to the
same Session, the UI subscribes once and keeps receiving output as new
runs take over.
- **Approval loops and long-running tasks with user feedback.** The task
waits on `.in`, the client writes to `.in`, the server enforces
no-writes-after-close.
- **Workflow progress streams that live past the run.** Subscribe to
`.out` after the task finishes to replay history.
- **Resume-next-day flows.** A session is a durable row, not a transient
stream. Send a message a day later and the server triggers a fresh run
on the same session.

### How it works (Session-as-run-manager)

A Session row is task-bound (`taskIdentifier` + `triggerConfig` are
required) and owns its current run via `currentRunId` +
`currentRunVersion` for optimistic claim. Three trigger paths:

1. **Session create** — `POST /api/v1/sessions` creates the row and
triggers the first run synchronously.
2. **Append-time probe** — `POST
/realtime/v1/sessions/:session/in/append` checks if the current run is
alive; if it has terminated (idle exit, crash, etc.), the server
triggers a new run before processing the append.
3. **End-and-continue handoff** — `POST
/api/v1/sessions/:session/end-and-continue`, called by the running
agent, triggers a fresh run and atomically swaps `currentRunId`. Used by
`chat.requestUpgrade()` for version handoffs.

Every triggered run is recorded in the `SessionRun` audit table with a
reason (`initial`, `continuation`, `upgrade`, `manual`).

## Public API surface

### Control plane

- `POST /api/v1/sessions` — create. Idempotent on `(env, externalId)`.
Triggers the first run, returns the session and a session-scoped public
access token. Returns 409 if the upserted row is already closed.
- `GET /api/v1/sessions/:session` — retrieve by friendlyId
(`session_abc...`) or by your own externalId (server disambiguates by
prefix).
- `GET /api/v1/sessions` — list with filters (`type`, `tag`,
`taskIdentifier`, `externalId`, derived `status` ACTIVE/CLOSED/EXPIRED,
created-at range) and cursor pagination. Backed by ClickHouse.
- `PATCH /api/v1/sessions/:session` — update tags / metadata /
externalId.
- `POST /api/v1/sessions/:session/close` — terminate. Idempotent,
hard-blocks new server-brokered writes.
- `POST /api/v1/sessions/:session/end-and-continue` — agent-only handoff
to a fresh run.

### Realtime

- `PUT /realtime/v1/sessions/:session/:io` — initialize a channel.
Returns S2 credentials in headers so high-throughput clients can write
direct to S2.
- `GET /realtime/v1/sessions/:session/:io` — SSE subscribe. Supports
Last-Event-ID resume and an opt-in `X-Peek-Settled: 1` header that
fast-closes the stream when the upstream is already settled
(`trigger:turn-complete`), eliminating long-poll wait on
reconnect-on-reload paths.
- `POST /realtime/v1/sessions/:session/:io/append` — server-side
appends.
- `POST /api/v1/runs/:runFriendlyId/session-streams/wait` — runs wait on
a session stream as a waitpoint, with a race-check to avoid suspending
if data already landed.

### Auth scopes

`sessions` is a new resource type. `read:sessions:{id}`,
`write:sessions:{id}`, `admin:sessions:{id}` flow through the existing
JWT validator. Session-scoped public access tokens minted by the server
replace browser-held trigger-task tokens for chat-style flows — the
browser never sees a run identifier or a run-scoped token in steady
state.

## What's coming after this PR

- **SDK + chat.agent migration**: separate branch, separate PR, ships in
the next `@trigger.dev/sdk` prerelease alongside this server deploy.
Customers using the prerelease `chat.agent` will follow the [upgrade
guide](https://github.com/triggerdotdev/trigger.dev/blob/docs/tri-7532-ai-sdk-chat-transport-and-chat-task-system/docs/ai-chat/upgrade-guide.mdx).
- **Dashboard surfaces**: dedicated agent list, agent playground, agent
view on the run dashboard. Tracking separately.

## Implementation notes

- **Postgres `Session` table**: scalar scoping columns (`projectId`,
`runtimeEnvironmentId`, `environmentType`, `organizationId`) without
FKs, matching the January TaskRun FK-removal decision. Point-lookup
indexes only — list queries go to ClickHouse. Terminal markers
(`closedAt`, `expiresAt`) are write-once.
- **ClickHouse `sessions_v1`**: ReplacingMergeTree, partitioned by
month, ordered by `(org_id, project_id, environment_id, created_at,
session_id)`. Tags indexed via `tokenbf_v1` skip index.
- **`SessionsReplicationService`**: mirrors `RunsReplicationService`
exactly — leader-locked logical replication consumer,
`ConcurrentFlushScheduler`, retry with exponential backoff + jitter,
identical metric shape. Dedicated slot + publication so the two consume
independently.
- **S2 keys**: `sessions/{addressingKey}/{out|in}`. The existing
`runs/{runId}/{streamId}` key format for run-scoped streams is
untouched.
- **Optimistic claim**: `ensureRunForSession` triggers a run upfront
(cheap to cancel if it loses the race), then attempts an `updateMany`
keyed on `currentRunVersion`. Loser cancels its triggered run and reuses
the winner's. No DB lock held across the trigger.

### What did NOT change

Run-scoped `streams.pipe` / `streams.input` and the existing
`/realtime/v1/streams/{runId}/...` routes are unchanged. Sessions are
net-new — not a reshaping of the current streams API.

## Deploy notes

- Set `SESSION_REPLICATION_CLICKHOUSE_URL` and
`SESSION_REPLICATION_ENABLED=1` to enable the replication consumer.
- The `Session` table needs `REPLICA IDENTITY FULL` set on the prod
source DB before the publication is created (same one-time DDL we did
for `TaskRun`). Required for delete events to carry full column values.
- Cross-form authorization on the `GET /api/v1/sessions/:session` loader
(a JWT minted for either form authorizes both URL forms). Action routes
are URL-form-specific, matching how the SDK mints PATs.

## Verification

- Webapp typecheck clean (10/10).
- `apps/webapp/test/sessionsReplicationService.test.ts` — round-trip
tests for insert/update/delete through Postgres logical replication into
ClickHouse via testcontainers.
- Live end-to-end against local dev: create + retrieve (both forms) +
update + close, `.out.initialize` + `.out.append` x2 + `.in.send` +
`.out.subscribe` over SSE, list with all filter combinations +
pagination, `end-and-continue` swap, `X-Peek-Settled` fast-close
(verified in browser via reconnect-on-reload and via curl). Replicated
row lands in ClickHouse within ~1s.
- Multi-round Devin + CodeRabbit review feedback addressed
(read-after-write paths use `prisma` writer, info-leak on auth-routes
masked as 403, peek-settled discriminator parsing fix, etc.).

## Test plan

- [ ] `pnpm run typecheck --filter webapp`
- [ ] `pnpm run test --filter webapp
./test/sessionsReplicationService.test.ts --run`
- [ ] Start the webapp with `SESSION_REPLICATION_CLICKHOUSE_URL` and
`SESSION_REPLICATION_ENABLED=1`. Confirm the slot and publication
auto-create on boot.
- [ ] `POST /api/v1/sessions` and verify the row replicates to
`trigger_dev.sessions_v1` within a couple of seconds.
- [ ] `POST /api/v1/sessions/:id/close`, then confirm `POST
/realtime/v1/sessions/:id/out/append` returns 400.
- [ ] Reuse a closed session's `externalId` on `POST /api/v1/sessions`
and confirm 409.
- [ ] `GET /realtime/v1/sessions/:id/out` with `X-Peek-Settled: 1` after
a turn completes and confirm `X-Session-Settled: true` response header +
immediate close.
This commit is contained in:
Eric Allam
2026-04-28 12:35:55 +01:00
committed by GitHub
parent e134da7306
commit c69e939c34
35 changed files with 4284 additions and 14 deletions
+5
View File
@@ -0,0 +1,5 @@
---
"@trigger.dev/core": patch
---
Add `SessionId` friendly ID generator and schemas for the new durable Session primitive. Exported from `@trigger.dev/core/v3/isomorphic` alongside `RunId`, `BatchId`, etc. Ships the `CreateSessionStreamWaitpoint` request/response schemas alongside the main Session CRUD.
+6
View File
@@ -0,0 +1,6 @@
---
area: webapp
type: feature
---
Add the `Session` primitive — a durable, task-bound, bidirectional I/O channel that outlives a single run and acts as the run manager for `chat.agent`. Ships the Postgres `Session` + `SessionRun` tables, ClickHouse `sessions_v1` + replication service, the `sessions` JWT scope, and the public CRUD + realtime routes (`/api/v1/sessions`, `/realtime/v1/sessions/:session/:io`) including `end-and-continue` for server-orchestrated run handoffs and session-stream waitpoints.
+38
View File
@@ -23,6 +23,44 @@ import {
registerRunEngineEventBusHandlers,
setupBatchQueueCallbacks,
} from "./v3/runEngineHandlers.server";
import { sessionsReplicationInstance } from "./services/sessionsReplicationInstance.server";
import { signalsEmitter } from "./services/signals.server";
// Start the sessions replication service (subscribes to the logical replication
// slot, runs leader election, flushes to ClickHouse). Done at entry level so it
// runs deterministically on webapp boot rather than lazily via a singleton
// reference elsewhere in the module graph.
if (sessionsReplicationInstance && env.SESSION_REPLICATION_ENABLED === "1") {
// Capture a non-nullable reference so the shutdown closure below
// doesn't need to re-null-check (TS narrowing doesn't follow through
// an inner function scope).
const replicator = sessionsReplicationInstance;
replicator
.start()
.then(() => {
console.log("🗃️ Sessions replication service started");
})
.catch((error) => {
console.error("🗃️ Sessions replication service failed to start", {
error,
});
});
// Wrap the async shutdown in a sync handler that catches rejections —
// SIGTERM/SIGINT fire during process teardown, and an unhandled
// promise rejection from `_replicationClient.stop()` there would
// bubble up past the process exit. Matches the pattern in
// dynamicFlushScheduler.server.ts.
const shutdownSessionsReplication = () => {
replicator.shutdown().catch((error) => {
console.error("🗃️ Sessions replication service shutdown error", {
error,
});
});
};
signalsEmitter.on("SIGTERM", shutdownSessionsReplication);
signalsEmitter.on("SIGINT", shutdownSessionsReplication);
}
const ABORT_DELAY = 30000;
+32
View File
@@ -1237,6 +1237,38 @@ const EnvironmentSchema = z
RUN_REPLICATION_DISABLE_PAYLOAD_INSERT: z.string().default("0"),
RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING: z.string().default("0"),
// Session replication (Postgres → ClickHouse sessions_v1). Shares Redis
// with the runs replicator for leader locking but has its own slot and
// publication so the two consume independently.
SESSION_REPLICATION_CLICKHOUSE_URL: z.string().optional(),
SESSION_REPLICATION_ENABLED: z.string().default("0"),
SESSION_REPLICATION_SLOT_NAME: z.string().default("sessions_to_clickhouse_v1"),
SESSION_REPLICATION_PUBLICATION_NAME: z
.string()
.default("sessions_to_clickhouse_v1_publication"),
SESSION_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(1),
SESSION_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
SESSION_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
SESSION_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
SESSION_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
SESSION_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce.number().int().default(10_000),
SESSION_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce.number().int().default(500),
SESSION_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
SESSION_REPLICATION_LOG_LEVEL: z
.enum(["log", "error", "warn", "info", "debug"])
.default("info"),
SESSION_REPLICATION_CLICKHOUSE_LOG_LEVEL: z
.enum(["log", "error", "warn", "info", "debug"])
.default("info"),
SESSION_REPLICATION_WAIT_FOR_ASYNC_INSERT: z.string().default("0"),
SESSION_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("0"),
SESSION_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
SESSION_REPLICATION_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
SESSION_REPLICATION_INSERT_STRATEGY: z.enum(["insert", "insert_async"]).default("insert"),
SESSION_REPLICATION_INSERT_MAX_RETRIES: z.coerce.number().int().default(3),
SESSION_REPLICATION_INSERT_BASE_DELAY_MS: z.coerce.number().int().default(100),
SESSION_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),
// Clickhouse
CLICKHOUSE_URL: z.string(),
CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
@@ -0,0 +1,188 @@
import { json } from "@remix-run/server-runtime";
import {
CreateSessionStreamWaitpointRequestBody,
type CreateSessionStreamWaitpointResponseBody,
} from "@trigger.dev/core/v3";
import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
import { z } from "zod";
import { $replica } from "~/db.server";
import { createWaitpointTag, MAX_TAGS_PER_WAITPOINT } from "~/models/waitpointTag.server";
import {
canonicalSessionAddressingKey,
isSessionFriendlyIdForm,
resolveSessionByIdOrExternalId,
} from "~/services/realtime/sessions.server";
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import {
addSessionStreamWaitpoint,
removeSessionStreamWaitpoint,
} from "~/services/sessionStreamWaitpointCache.server";
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
import { logger } from "~/services/logger.server";
import { parseDelay } from "~/utils/delays";
import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
import { engine } from "~/v3/runEngine.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
const ParamsSchema = z.object({
runFriendlyId: z.string(),
});
const { action, loader } = createActionApiRoute(
{
params: ParamsSchema,
body: CreateSessionStreamWaitpointRequestBody,
maxContentLength: 1024 * 10, // 10KB
method: "POST",
},
async ({ authentication, body, params }) => {
try {
const run = await $replica.taskRun.findFirst({
where: {
friendlyId: params.runFriendlyId,
runtimeEnvironmentId: authentication.environment.id,
},
select: {
id: true,
friendlyId: true,
realtimeStreamsVersion: true,
},
});
if (!run) {
return json({ error: "Run not found" }, { status: 404 });
}
// Row-optional addressing — see the .out / .in.append handlers.
// The waitpoint cache + S2 stream key derive from the row's
// canonical identity (externalId if set, else friendlyId), so
// the agent's wait registration and the append-side drain
// converge regardless of which URL form each side used.
const maybeSession = await resolveSessionByIdOrExternalId(
$replica,
authentication.environment.id,
body.session
);
if (!maybeSession && isSessionFriendlyIdForm(body.session)) {
return json({ error: "Session not found" }, { status: 404 });
}
const addressingKey = canonicalSessionAddressingKey(maybeSession, body.session);
const idempotencyKeyExpiresAt = body.idempotencyKeyTTL
? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
: undefined;
const timeout = await parseDelay(body.timeout);
const bodyTags = typeof body.tags === "string" ? [body.tags] : body.tags;
if (bodyTags && bodyTags.length > MAX_TAGS_PER_WAITPOINT) {
throw new ServiceValidationError(
`Waitpoints can only have ${MAX_TAGS_PER_WAITPOINT} tags, you're trying to set ${bodyTags.length}.`
);
}
if (bodyTags && bodyTags.length > 0) {
for (const tag of bodyTags) {
await createWaitpointTag({
tag,
environmentId: authentication.environment.id,
projectId: authentication.environment.projectId,
});
}
}
// Step 1: Create the waitpoint.
const result = await engine.createManualWaitpoint({
environmentId: authentication.environment.id,
projectId: authentication.environment.projectId,
idempotencyKey: body.idempotencyKey,
idempotencyKeyExpiresAt,
timeout,
tags: bodyTags,
});
// Step 2: Register the waitpoint on the session channel so the next
// append fires it. Keyed by (addressingKey, io) — the canonical
// string for the row. The append handler drains by the same
// canonical key, so writers and readers converge regardless of
// which URL form the agent vs. the appending caller used.
const ttlMs = timeout ? timeout.getTime() - Date.now() : undefined;
await addSessionStreamWaitpoint(
addressingKey,
body.io,
result.waitpoint.id,
ttlMs && ttlMs > 0 ? ttlMs : undefined
);
// Step 3: Race-check. If a record landed on the channel before this
// .wait() call, complete the waitpoint synchronously with that data
// and remove the pending registration.
if (!result.isCached) {
try {
// Session streams are always v2 (S2) — the writer in
// `appendPartToSessionStream` and the SSE subscribe both
// hardcode "v2", so the race-check reader has to match.
// Don't fall through to the run's own `realtimeStreamsVersion`,
// which only describes the run's run-scoped streams.
const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
if (realtimeStream instanceof S2RealtimeStreams) {
const records = await realtimeStream.readSessionStreamRecords(
addressingKey,
body.io,
body.lastSeqNum
);
if (records.length > 0) {
const record = records[0]!;
await engine.completeWaitpoint({
id: result.waitpoint.id,
output: {
value: record.data,
type: "application/json",
isError: false,
},
});
await removeSessionStreamWaitpoint(
addressingKey,
body.io,
result.waitpoint.id
);
}
}
} catch (error) {
// Non-fatal: pending registration stays in Redis; the next append
// will complete the waitpoint via the append handler path. Log so
// a broken race-check doesn't silently degrade to timeout-only.
logger.warn("session-stream wait race-check failed", {
addressingKey,
io: body.io,
waitpointId: WaitpointId.toFriendlyId(result.waitpoint.id),
error,
});
}
}
return json<CreateSessionStreamWaitpointResponseBody>({
waitpointId: WaitpointId.toFriendlyId(result.waitpoint.id),
isCached: result.isCached,
});
} catch (error) {
if (error instanceof ServiceValidationError) {
return json({ error: error.message }, { status: 422 });
}
// Don't forward raw internal error messages (could leak Prisma/engine
// details). Log server-side and return a generic 500.
logger.error("Failed to create session-stream waitpoint", { error });
return json({ error: "Something went wrong" }, { status: 500 });
}
}
);
export { action, loader };
@@ -0,0 +1,79 @@
import { json } from "@remix-run/server-runtime";
import {
CloseSessionRequestBody,
type RetrieveSessionResponseBody,
} from "@trigger.dev/core/v3";
import { z } from "zod";
import { $replica, prisma } from "~/db.server";
import {
resolveSessionByIdOrExternalId,
serializeSessionWithFriendlyRunId,
} from "~/services/realtime/sessions.server";
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
const ParamsSchema = z.object({
session: z.string(),
});
const { action, loader } = createActionApiRoute(
{
params: ParamsSchema,
body: CloseSessionRequestBody,
maxContentLength: 1024,
method: "POST",
allowJWT: true,
corsStrategy: "all",
authorization: {
action: "admin",
resource: (params) => ({ sessions: params.session }),
superScopes: ["admin:sessions", "admin:all", "admin"],
},
},
async ({ authentication, params, body }) => {
const existing = await resolveSessionByIdOrExternalId(
$replica,
authentication.environment.id,
params.session
);
if (!existing) {
return json({ error: "Session not found" }, { status: 404 });
}
// Idempotent: if already closed, return the current row without clobbering
// the original closedAt / closedReason.
if (existing.closedAt) {
return json<RetrieveSessionResponseBody>(
await serializeSessionWithFriendlyRunId(existing)
);
}
// `closedAt: null` on the where clause makes the update conditional at
// the DB level. Two concurrent closes race through the earlier read,
// but only one can win this update — the loser hits `count === 0` and
// falls back to reading the winning row. Closedness is write-once.
const { count } = await prisma.session.updateMany({
where: { id: existing.id, closedAt: null },
data: {
closedAt: new Date(),
closedReason: body.reason ?? null,
},
});
if (count === 0) {
const final = await prisma.session.findFirst({ where: { id: existing.id } });
if (!final) return json({ error: "Session not found" }, { status: 404 });
return json<RetrieveSessionResponseBody>(
await serializeSessionWithFriendlyRunId(final)
);
}
const updated = await prisma.session.findFirst({ where: { id: existing.id } });
if (!updated) return json({ error: "Session not found" }, { status: 404 });
return json<RetrieveSessionResponseBody>(
await serializeSessionWithFriendlyRunId(updated)
);
}
);
export { action, loader };
@@ -0,0 +1,135 @@
import { json } from "@remix-run/server-runtime";
import {
EndAndContinueSessionRequestBody,
type EndAndContinueSessionResponseBody,
} from "@trigger.dev/core/v3";
import { z } from "zod";
import { $replica, prisma } from "~/db.server";
import { logger } from "~/services/logger.server";
import { swapSessionRun } from "~/services/realtime/sessionRunManager.server";
import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
const ParamsSchema = z.object({
session: z.string(),
});
// POST /api/v1/sessions/:session/end-and-continue
//
// Generic "the running run is exiting; please trigger a fresh one for
// this session and swap `currentRunId` to it" endpoint. The agent calls
// this from `chat.requestUpgrade` and other planned-handoff paths. The
// transport's `.out` SSE keeps streaming across the swap because S2 is
// keyed on the session, not the run — v1's last chunks land, v2's new
// chunks land on the same stream.
//
// Auth: `write:sessions:{ext}` — the running agent's internal API key
// (PRIVATE) bypasses authorization; a browser holding the session PAT
// can also reach this endpoint, which is fine: if you have the session
// PAT, you own the chat.
const { action, loader } = createActionApiRoute(
{
params: ParamsSchema,
body: EndAndContinueSessionRequestBody,
method: "POST",
maxContentLength: 1024,
allowJWT: true,
corsStrategy: "all",
// Resolved before authorization so the auth scope can expand to both
// addressing forms (friendlyId + externalId). Handler reads the row
// from `resource` instead of re-fetching.
findResource: async (params, auth) =>
resolveSessionByIdOrExternalId($replica, auth.environment.id, params.session),
authorization: {
action: "write",
resource: (params, _, __, ___, session) => {
const ids = new Set<string>([params.session]);
if (session) {
ids.add(session.friendlyId);
if (session.externalId) ids.add(session.externalId);
}
return { sessions: [...ids] };
},
superScopes: ["write:sessions", "write:all", "admin"],
},
},
async ({ authentication, params, body, resource: session }) => {
if (!session) {
// Unreachable — `findResource` 404s before this runs. Type narrow.
return json({ error: "Session not found" }, { status: 404 });
}
if (session.closedAt) {
return json(
{ error: "Cannot end-and-continue a closed session" },
{ status: 400 }
);
}
if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
return json(
{ error: "Cannot end-and-continue an expired session" },
{ status: 400 }
);
}
// The wire `callingRunId` is a friendlyId (that's what the agent
// SDK exposes via `ctx.run.id`). Internally `Session.currentRunId`
// stores the TaskRun.id cuid, so resolve before handing to the
// optimistic-claim service.
const callingRun = await $replica.taskRun.findFirst({
where: {
friendlyId: body.callingRunId,
runtimeEnvironmentId: authentication.environment.id,
},
select: { id: true },
});
if (!callingRun) {
return json({ error: "callingRunId not found in this environment" }, { status: 404 });
}
try {
// Body's `reason` is free-form for forward-compat (audit metadata
// only); narrow into the closed `EnsureRunReason` set, defaulting
// to `"manual"` for unknown labels.
const reason: "initial" | "continuation" | "upgrade" | "manual" =
body.reason === "upgrade" ||
body.reason === "continuation" ||
body.reason === "initial" ||
body.reason === "manual"
? body.reason
: "manual";
const result = await swapSessionRun({
session,
callingRunId: callingRun.id,
environment: authentication.environment,
reason,
});
// Read-after-write: the swap just triggered (or claimed) the
// run on the writer, so read it from `prisma` rather than
// `$replica`. A replica miss here would silently fall back to
// returning the internal cuid, which the public API contract
// says is a friendlyId.
const run = await prisma.taskRun.findFirst({
where: { id: result.runId },
select: { friendlyId: true },
});
const responseBody: EndAndContinueSessionResponseBody = {
runId: run?.friendlyId ?? result.runId,
swapped: result.swapped,
};
return json<EndAndContinueSessionResponseBody>(responseBody);
} catch (error) {
logger.error("Failed end-and-continue", {
sessionId: session.id,
error,
});
return json({ error: "Failed to swap session run" }, { status: 500 });
}
}
);
export { action, loader };
@@ -0,0 +1,108 @@
import { json } from "@remix-run/server-runtime";
import {
type RetrieveSessionResponseBody,
UpdateSessionRequestBody,
} from "@trigger.dev/core/v3";
import { Prisma } from "@trigger.dev/database";
import { z } from "zod";
import { $replica, prisma } from "~/db.server";
import {
resolveSessionByIdOrExternalId,
serializeSessionWithFriendlyRunId,
} from "~/services/realtime/sessions.server";
import {
createActionApiRoute,
createLoaderApiRoute,
} from "~/services/routeBuilders/apiBuilder.server";
const ParamsSchema = z.object({
session: z.string(),
});
export const loader = createLoaderApiRoute(
{
params: ParamsSchema,
allowJWT: true,
corsStrategy: "all",
findResource: async (params, auth) => {
return resolveSessionByIdOrExternalId($replica, auth.environment.id, params.session);
},
authorization: {
action: "read",
resource: (session) => ({ sessions: [session.friendlyId, session.externalId ?? ""] }),
superScopes: ["read:sessions", "read:all", "admin"],
},
},
async ({ resource: session }) => {
return json<RetrieveSessionResponseBody>(
await serializeSessionWithFriendlyRunId(session)
);
}
);
const { action } = createActionApiRoute(
{
params: ParamsSchema,
body: UpdateSessionRequestBody,
maxContentLength: 1024 * 32,
method: "PATCH",
allowJWT: true,
corsStrategy: "all",
authorization: {
action: "admin",
resource: (params) => ({ sessions: params.session }),
superScopes: ["admin:sessions", "admin:all", "admin"],
},
},
async ({ authentication, params, body }) => {
const existing = await resolveSessionByIdOrExternalId(
$replica,
authentication.environment.id,
params.session
);
if (!existing) {
return json({ error: "Session not found" }, { status: 404 });
}
try {
const updated = await prisma.session.update({
where: { id: existing.id },
data: {
...(body.tags !== undefined ? { tags: body.tags } : {}),
...(body.metadata !== undefined
? {
metadata:
body.metadata === null
? Prisma.JsonNull
: (body.metadata as Prisma.InputJsonValue),
}
: {}),
...(body.externalId !== undefined ? { externalId: body.externalId } : {}),
},
});
return json<RetrieveSessionResponseBody>(
await serializeSessionWithFriendlyRunId(updated)
);
} catch (error) {
// A duplicate externalId in the same environment violates the
// `(runtimeEnvironmentId, externalId)` unique constraint. Surface that
// as a 409 rather than a generic 500.
if (
error instanceof Prisma.PrismaClientKnownRequestError &&
error.code === "P2002" &&
Array.isArray((error.meta as { target?: string[] })?.target) &&
((error.meta as { target?: string[] }).target ?? []).includes("externalId")
) {
return json(
{ error: "A session with this externalId already exists in this environment" },
{ status: 409 }
);
}
throw error;
}
}
);
export { action };
+238
View File
@@ -0,0 +1,238 @@
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 { clickhouseClient } from "~/services/clickhouseInstance.server";
import { logger } from "~/services/logger.server";
import { mintSessionToken } from "~/services/realtime/mintSessionToken.server";
import {
ensureRunForSession,
type SessionTriggerConfig,
} from "~/services/realtime/sessionRunManager.server";
import { serializeSession } from "~/services/realtime/sessions.server";
import { SessionsRepository } from "~/services/sessionsRepository/sessionsRepository.server";
import {
createActionApiRoute,
createLoaderApiRoute,
} from "~/services/routeBuilders/apiBuilder.server";
import { ServiceValidationError } from "~/v3/services/common.server";
function asArray<T>(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",
resource: (_, __, searchParams) => ({ tasks: searchParams["filter[taskIdentifier]"] }),
superScopes: ["read:sessions", "read:all", "admin"],
},
findResource: async () => 1,
},
async ({ searchParams, authentication }) => {
const repository = new SessionsRepository({
clickhouse: clickhouseClient,
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,
},
});
return json<ListSessionsResponseBody>({
data: rows.map((row) =>
serializeSession({
...row,
// Columns the list query doesn't select — filled so `serializeSession`
// can operate on a narrowed payload without type errors.
projectId: authentication.environment.projectId,
environmentType: authentication.environment.type,
organizationId: authentication.environment.organizationId,
} as Session)
),
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
// Secret-key only. 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.
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,
},
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,
},
});
}
// 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 }
);
}
// 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 prisma.taskRun.findFirst({
where: { id: ensureResult.runId },
select: { friendlyId: true },
});
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<CreatedSessionResponseBody>(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 };
@@ -0,0 +1,183 @@
import { json } from "@remix-run/server-runtime";
import { tryCatch } from "@trigger.dev/core/utils";
import { nanoid } from "nanoid";
import { z } from "zod";
import { $replica } from "~/db.server";
import { logger } from "~/services/logger.server";
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
import { ensureRunForSession } from "~/services/realtime/sessionRunManager.server";
import {
canonicalSessionAddressingKey,
resolveSessionByIdOrExternalId,
} from "~/services/realtime/sessions.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { drainSessionStreamWaitpoints } from "~/services/sessionStreamWaitpointCache.server";
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
import { engine } from "~/v3/runEngine.server";
import { ServiceValidationError } from "~/v3/services/common.server";
const ParamsSchema = z.object({
session: z.string(),
io: z.enum(["out", "in"]),
});
// POST: server-side append of a single record to a session channel. Mirrors
// the existing /realtime/v1/streams/:runId/:target/:streamId/append route,
// scoped to a Session primitive.
// S2 enforces a 1 MiB per-record limit (metered as
// `8 + 2*H + Σ(header name+value) + body`). We cap the raw HTTP body at
// 512 KiB so the JSON wrapper (`{"data":"...","id":"..."}`), string
// escaping, and any future per-record header additions all stay comfortably
// below S2's ceiling. See https://s2.dev/docs/limits.
const MAX_APPEND_BODY_BYTES = 1024 * 512;
const { action, loader } = createActionApiRoute(
{
params: ParamsSchema,
method: "POST",
maxContentLength: MAX_APPEND_BODY_BYTES,
allowJWT: true,
corsStrategy: "all",
// Sessions are task-bound (created by `POST /api/v1/sessions` which
// also triggers the first run). The row exists before any caller
// can reach `.in/append` — no row, no append. Resolved here so the
// authorization scope can expand to both addressing forms (friendlyId
// + externalId) and the handler can skip its own lookup.
findResource: async (params, auth) =>
resolveSessionByIdOrExternalId($replica, auth.environment.id, params.session),
authorization: {
action: "write",
// Authorize against the union of the URL form, friendlyId, and
// externalId so a JWT scoped to any form authorizes any URL.
resource: (params, _, __, ___, session) => {
const ids = new Set<string>([params.session]);
if (session) {
ids.add(session.friendlyId);
if (session.externalId) ids.add(session.externalId);
}
return { sessions: [...ids] };
},
superScopes: ["write:sessions", "write:all", "admin"],
},
},
async ({ request, params, authentication, resource: session }) => {
if (!session) {
// Unreachable — `findResource` short-circuits to 404 before this
// handler runs. Type-narrow the rest of the body.
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 }
);
}
const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
if (!(realtimeStream instanceof S2RealtimeStreams)) {
return json(
{ ok: false, error: "Session channels require the S2 realtime backend" },
{ status: 501 }
);
}
// Probe + ensure a live run before appending. The append itself is
// run-independent (S2 stream is durable, keyed on the session) but
// the message is useless if no run is alive to consume it. The
// probe is a single Prisma read; ensureRunForSession is no-op when
// currentRunId is alive, so the steady-state cost is one extra
// read in the hot path.
//
// Best-effort: if ensureRunForSession throws (e.g. the trigger
// call fails transiently), still append to S2 — the record is
// durable and the next append will retry the ensure step. Don't
// surface the error to the caller; the SSE tail just won't deliver
// it until a run boots.
const [ensureError] = await tryCatch(
ensureRunForSession({
session,
environment: authentication.environment,
reason: "continuation",
})
);
if (ensureError) {
logger.error("Failed to ensureRunForSession on .in/append", {
sessionId: session.id,
externalId: session.externalId,
error: ensureError,
});
}
const addressingKey = canonicalSessionAddressingKey(session, params.session);
const part = await request.text();
const partId = request.headers.get("X-Part-Id") ?? nanoid(7);
const [appendError] = await tryCatch(
realtimeStream.appendPartToSessionStream(part, partId, addressingKey, params.io)
);
if (appendError) {
if (appendError instanceof ServiceValidationError) {
return json(
{ ok: false, error: appendError.message },
{ status: appendError.status ?? 422 }
);
}
return json({ ok: false, error: appendError.message }, { status: 500 });
}
// Fire any run-scoped waitpoints registered against this channel. Best
// effort — a failure here must not fail the append (the record is
// durable in S2; the SSE tail will still deliver it). Waitpoints are
// keyed on the canonical addressing key the agent registered with via
// `sessions.open(...).in.wait()`, so writers and readers converge
// regardless of which URL form they used.
const [drainError, waitpointIds] = await tryCatch(
drainSessionStreamWaitpoints(addressingKey, params.io)
);
if (drainError) {
logger.error("Failed to drain session stream waitpoints", {
addressingKey,
io: params.io,
error: drainError,
});
} else if (waitpointIds && waitpointIds.length > 0) {
await Promise.all(
waitpointIds.map(async (waitpointId) => {
const [completeError] = await tryCatch(
engine.completeWaitpoint({
id: waitpointId,
output: {
value: part,
type: "application/json",
isError: false,
},
})
);
if (completeError) {
logger.error("Failed to complete session stream waitpoint", {
addressingKey,
io: params.io,
waitpointId,
error: completeError,
});
}
})
);
}
return json({ ok: true }, { status: 200 });
}
);
export { action, loader };
@@ -0,0 +1,181 @@
import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
import {
canonicalSessionAddressingKey,
isSessionFriendlyIdForm,
resolveSessionByIdOrExternalId,
} from "~/services/realtime/sessions.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import {
createActionApiRoute,
createLoaderApiRoute,
} from "~/services/routeBuilders/apiBuilder.server";
const ParamsSchema = z.object({
session: z.string(),
io: z.enum(["out", "in"]),
});
// PUT: initialize the S2 channel for this (session, io) pair — returns S2
// credentials in response headers so the caller can write/read directly
// against S2. GET is handled by the loader below.
const { action } = createActionApiRoute(
{
params: ParamsSchema,
method: "PUT",
allowJWT: true,
corsStrategy: "all",
authorization: {
action: "write",
resource: (params) => ({ sessions: params.session }),
superScopes: ["write:sessions", "write:all", "admin"],
},
},
async ({ params, authentication }) => {
// Row-optional addressing. The agent calls PUT initialize as part
// of `session.out.writer()`, by which time it has already created
// the row at bind, so a missing row here is an unusual case
// (manual init from outside chat.agent). Require a real row only
// for opaque friendlyIds, and treat closedAt as a soft reject only
// when a row exists. The S2 stream key is built from the row's
// canonical key (externalId if set, else friendlyId) so writers
// and readers converge regardless of URL form.
const maybeSession = await resolveSessionByIdOrExternalId(
$replica,
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,
});
}
const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
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
);
return json({ version: "v2" }, { status: 202, headers: responseHeaders });
}
);
// GET: SSE subscribe to a session channel. HEAD returns the last chunk index
// for resume semantics, mirroring the existing run-stream route.
//
// Subscribes are row-optional: the chat.agent transport opens the SSE on
// `chatId` (externalId) before the agent has booted and upserted the
// Session row. The S2 stream is keyed on the row's *canonical* identity
// (externalId if set, else friendlyId) so two callers addressing the
// same row via different URL forms converge on the same stream. We
// short-circuit to 404 only for opaque `session_*` friendlyIds (those
// must come from a real mint).
const loader = createLoaderApiRoute(
{
params: ParamsSchema,
allowJWT: true,
corsStrategy: "all",
findResource: async (params, auth) => {
const row = await resolveSessionByIdOrExternalId(
$replica,
auth.environment.id,
params.session
);
if (!row && isSessionFriendlyIdForm(params.session)) {
return undefined; // 404 — opaque friendlyId must reference a real row
}
// Non-null wrapper so missing row doesn't 404 for externalId form.
return {
row,
addressingKey: canonicalSessionAddressingKey(row, params.session),
};
},
authorization: {
action: "read",
resource: ({ row, addressingKey }) => {
const ids = new Set<string>([addressingKey]);
if (row) {
ids.add(row.friendlyId);
if (row.externalId) ids.add(row.externalId);
}
return { sessions: [...ids] };
},
superScopes: ["read:sessions", "read:all", "admin"],
},
},
async ({ params, request, authentication, resource }) => {
const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
if (!(realtimeStream instanceof S2RealtimeStreams)) {
return new Response("Session channels require the S2 realtime backend", {
status: 501,
});
}
if (request.method === "HEAD") {
// No last-chunk-index on the S2 backend (clients resume via Last-Event-ID
// on the SSE stream directly). Return 200 with a zero index for
// compatibility with the run-stream shape.
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) {
// `Number()` rejects `"10abc"` as NaN; `parseInt` would silently accept
// the trailing garbage and bypass the bounds checks below.
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;
}
// Opt-in: only consider the settled-peek shortcut when the client
// asks for it via `X-Peek-Settled: 1`. Reconnect-on-reload paths
// (`TriggerChatTransport.reconnectToStream`) set this; the active
// send-a-message path (`sendMessages → subscribeToSessionStream`)
// does not — otherwise the peek races with the newly-triggered
// turn's first chunk and the SSE closes before records land.
const peekSettled = request.headers.get("X-Peek-Settled") === "1";
return realtimeStream.streamResponseFromSessionStream(
request,
resource.addressingKey,
params.io,
getRequestAbortSignal(),
{ lastEventId, timeoutInSeconds, peekSettled }
);
}
);
export { action, loader };
@@ -13,9 +13,17 @@ const ParamsSchema = z.object({
streamId: z.string(),
});
// S2 enforces a 1 MiB per-record limit (metered as
// `8 + 2*H + Σ(header name+value) + body`). Cap the raw HTTP body at
// 512 KiB so the JSON wrapper, string escaping, and any future per-record
// header additions all stay well under S2's ceiling.
// See https://s2.dev/docs/limits.
const MAX_APPEND_BODY_BYTES = 1024 * 512;
const { action } = createActionApiRoute(
{
params: ParamsSchema,
maxContentLength: MAX_APPEND_BODY_BYTES,
},
async ({ request, params, authentication }) => {
const run = await $replica.taskRun.findFirst({
@@ -1,6 +1,6 @@
export type AuthorizationAction = "read" | "write" | string; // Add more actions as needed
const ResourceTypes = ["tasks", "tags", "runs", "batch", "waitpoints", "deployments", "inputStreams", "query", "prompts"] as const;
const ResourceTypes = ["tasks", "tags", "runs", "batch", "waitpoints", "deployments", "inputStreams", "query", "prompts", "sessions"] as const;
export type AuthorizationResources = {
[key in (typeof ResourceTypes)[number]]?: string | string[];
@@ -0,0 +1,40 @@
import { generateJWT as internal_generateJWT } from "@trigger.dev/core/v3";
import { extractJwtSigningSecretKey } from "./jwtAuth.server";
type Environment = Parameters<typeof extractJwtSigningSecretKey>[0];
export type MintSessionTokenOptions = {
/** Token expiration. Defaults to "1h". */
expirationTime?: string;
};
/**
* Mint a session-scoped public access token (JWT) covering both `.in`
* append and `.out` subscribe for a session's realtime channels.
*
* Returned by `POST /api/v1/sessions` so the browser holds a single
* long-lived token that survives across runs (sessions outlive any
* single run). Includes both read and write scopes since the transport
* needs both: read for SSE subscribe on `.out`, write for `.in` appends
* (`stop`, follow-up messages, action chunks).
*/
export async function mintSessionToken(
environment: Environment,
sessionAddressingKey: string,
options: MintSessionTokenOptions = {}
): Promise<string> {
const scopes = [
`read:sessions:${sessionAddressingKey}`,
`write:sessions:${sessionAddressingKey}`,
];
return internal_generateJWT({
secretKey: extractJwtSigningSecretKey(environment),
payload: {
sub: environment.id,
pub: true,
scopes,
},
expirationTime: options.expirationTime ?? "1h",
});
}
@@ -88,9 +88,42 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
return `${this.streamPrefix}/runs/${runId}/${streamId}`;
}
/**
* Build an S2 stream name for a `Session`-primitive channel, addressed by
* 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}`;
}
async initializeStream(
runId: string,
streamId: string
): Promise<{ responseHeaders?: Record<string, string> }> {
return this.#initializeStreamByName(
this.toStreamName(runId, streamId),
`/runs/${runId}/${streamId}`
);
}
/**
* Initialize an S2 stream by `(sessionFriendlyId, io)` — mirrors
* {@link initializeStream} but addresses the new `sessions/*` key format.
*/
async initializeSessionStream(
friendlyId: string,
io: "out" | "in"
): Promise<{ responseHeaders?: Record<string, string> }> {
return this.#initializeStreamByName(
this.toSessionStreamName(friendlyId, io),
`/sessions/${friendlyId}/${io}`
);
}
async #initializeStreamByName(
prefixedName: string,
relativeName: string
): Promise<{ responseHeaders?: Record<string, string> }> {
const accessToken = this.skipAccessTokens
? this.token
@@ -99,9 +132,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
return {
responseHeaders: {
"X-S2-Access-Token": accessToken,
"X-S2-Stream-Name": this.skipAccessTokens
? this.toStreamName(runId, streamId)
: `/runs/${runId}/${streamId}`,
"X-S2-Stream-Name": this.skipAccessTokens ? prefixedName : relativeName,
"X-S2-Basin": this.basin,
"X-S2-Flush-Interval-Ms": this.flushIntervalMs.toString(),
"X-S2-Max-Retries": this.maxRetries.toString(),
@@ -121,8 +152,22 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
}
async appendPart(part: string, partId: string, runId: string, streamId: string): Promise<void> {
const s2Stream = this.toStreamName(runId, streamId);
return this.#appendPartByName(part, partId, this.toStreamName(runId, streamId));
}
/**
* Append a single record to a `Session`-primitive channel.
*/
async appendPartToSessionStream(
part: string,
partId: string,
friendlyId: string,
io: "out" | "in"
): Promise<void> {
return this.#appendPartByName(part, partId, this.toSessionStreamName(friendlyId, io));
}
async #appendPartByName(part: string, partId: string, s2Stream: string): Promise<void> {
this.logger.debug(`S2 appending to stream`, { part, stream: s2Stream });
const result = await this.s2Append(s2Stream, {
@@ -141,7 +186,22 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
streamId: string,
afterSeqNum?: number
): Promise<StreamRecord[]> {
const s2Stream = this.toStreamName(runId, streamId);
return this.#readRecordsByName(this.toStreamName(runId, streamId), afterSeqNum);
}
/**
* Read records from a `Session`-primitive channel starting after the
* given sequence number. Used by the `.wait()` race-check path.
*/
async readSessionStreamRecords(
friendlyId: string,
io: "out" | "in",
afterSeqNum?: number
): Promise<StreamRecord[]> {
return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io), afterSeqNum);
}
async #readRecordsByName(s2Stream: string, afterSeqNum?: number): Promise<StreamRecord[]> {
const startSeq = afterSeqNum != null ? afterSeqNum + 1 : 0;
const qs = new URLSearchParams();
@@ -227,7 +287,142 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
signal: AbortSignal,
options?: StreamResponseOptions
): Promise<Response> {
const s2Stream = this.toStreamName(runId, streamId);
return this.#streamResponseByName(this.toStreamName(runId, streamId), signal, options);
}
/**
* Serve SSE from a `Session`-primitive channel addressed by
* `(friendlyId, io)`.
*
* For `io=out`, peek the tail record first. If it's
* `trigger:turn-complete`, the agent has finished a turn and is
* either idle-waiting on `.in` or has exited — either way, no more
* chunks will arrive without further user action. We switch the
* downstream S2 read to `wait=0` (drain whatever's left, close fast)
* and set `X-Session-Settled: true` so the client knows this SSE
* close is terminal instead of the normal 60s long-poll cycle.
*
* Mid-turn tail (streaming UIMessageChunk) falls through to the
* long-poll path; a crashed-mid-turn stream is indistinguishable
* here and behaves like today (client sees wait=60 close, retries).
*/
async streamResponseFromSessionStream(
request: Request,
friendlyId: string,
io: "out" | "in",
signal: AbortSignal,
options?: StreamResponseOptions
): Promise<Response> {
const s2Stream = this.toSessionStreamName(friendlyId, io);
let waitSeconds = options?.timeoutInSeconds ?? this.s2WaitSeconds;
let settled = false;
// Only peek + settle when the client opts in via `options.peekSettled`.
// Reconnect-on-reload paths (`TriggerChatTransport.reconnectToStream`)
// set it; active send-a-message paths don't — otherwise the peek
// races the newly-triggered turn's first chunk and the SSE closes
// before records land.
if (io === "out" && options?.peekSettled) {
const lastChunk = await this.#peekLastChunkBody(s2Stream);
const lastChunkType =
lastChunk != null && typeof lastChunk === "object"
? (lastChunk as { type?: unknown }).type
: null;
if (lastChunkType === "trigger:turn-complete") {
settled = true;
waitSeconds = 0;
}
}
const s2Response = await this.#streamResponseByName(s2Stream, signal, {
...options,
timeoutInSeconds: waitSeconds,
});
if (!settled) return s2Response;
const headers = new Headers(s2Response.headers);
headers.set("X-Session-Settled", "true");
return new Response(s2Response.body, {
status: s2Response.status,
statusText: s2Response.statusText,
headers,
});
}
async #peekLastChunkBody(s2Stream: string): Promise<unknown | null> {
const qs = new URLSearchParams();
// `tail_offset=1` reads one record before the next seq — i.e. the
// most recently appended record. `count=1` caps it to just that
// record. `wait=0` returns immediately with no long-poll.
qs.set("tail_offset", "1");
qs.set("count", "1");
qs.set("wait", "0");
let res: Response;
try {
res = await fetch(
`${this.baseUrl}/streams/${encodeURIComponent(s2Stream)}/records?${qs}`,
{
method: "GET",
headers: {
Authorization: `Bearer ${this.token}`,
Accept: "application/json",
"S2-Format": "raw",
"S2-Basin": this.basin,
},
}
);
} catch (err) {
this.logger.warn("S2 peek last record: fetch failed", { err, stream: s2Stream });
return null;
}
if (!res.ok) {
// 404: stream has never been written to. 416: range not
// satisfiable (empty stream). Both mean "nothing to peek."
if (res.status === 404 || res.status === 416) return null;
const text = await res.text().catch(() => "");
this.logger.warn("S2 peek last record failed", {
status: res.status,
statusText: res.statusText,
text,
stream: s2Stream,
});
return null;
}
try {
const json = (await res.json()) as {
records?: Array<{ body: string; seq_num: number; timestamp: number }>;
};
const record = json.records?.[0];
if (!record) return null;
// The record body is a JSON string `{data: <chunkAsString>, id: partId}`.
// The agent-side writer (`StreamsWriterV2`) hands `appendPart` an
// already-JSON-stringified chunk, so `data` round-trips as a string,
// not an object. Parse it once more to surface the chunk shape.
const envelope = JSON.parse(record.body) as { data: unknown; id: string };
if (typeof envelope.data === "string") {
try {
return JSON.parse(envelope.data);
} catch {
return envelope.data;
}
}
return envelope.data;
} catch (err) {
this.logger.warn("S2 peek last record: parse failed", { err, stream: s2Stream });
return null;
}
}
async #streamResponseByName(
s2Stream: string,
signal: AbortSignal,
options?: StreamResponseOptions
): Promise<Response> {
const startSeq = this.parseLastEventId(options?.lastEventId);
this.logger.info(`S2 streaming records from stream`, { stream: s2Stream, startSeq });
@@ -0,0 +1,375 @@
import type { Session, TaskRunStatus } from "@trigger.dev/database";
import { SessionTriggerConfig as SessionTriggerConfigZod } from "@trigger.dev/core/v3";
import { z } from "zod";
import { prisma, $replica } from "~/db.server";
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server";
import { TriggerTaskService } from "~/v3/services/triggerTask.server";
import { isFinalRunStatus } from "~/v3/taskStatus";
/**
* Schema for `Session.triggerConfig` (stored as JSONB). The wire-format
* source of truth lives in `@trigger.dev/core/v3` as `SessionTriggerConfig`;
* we re-export it here for the trigger machinery to validate on read.
*
* `basePayload` carries the customer's wire payload (for chat.agent:
* `{ chatId, ...clientData, idleTimeoutInSeconds? }`). Runtime fields
* specific to a particular trigger (e.g. `trigger: "trigger" | "preload"`,
* an `isContinuation` flag) come in via the `payloadOverrides` argument
* to `ensureRunForSession` and shallow-merge on top of `basePayload`.
*/
export const SessionTriggerConfigSchema = SessionTriggerConfigZod;
export type SessionTriggerConfig = z.infer<typeof SessionTriggerConfigSchema>;
export type EnsureRunReason = "initial" | "continuation" | "upgrade" | "manual";
/**
* Hard cap on how many times `ensureRunForSession` will recurse on the
* pathological "we lost the claim race AND the winner's run was already
* terminal" path. In practice progress through the run engine bounds
* this, but a misconfigured task that crashes before it can be dequeued
* could otherwise loop without limit. After this many attempts we
* surface `SessionRunManagerError` so the caller can 5xx instead of
* blowing the stack.
*/
const ENSURE_RUN_FOR_SESSION_MAX_ATTEMPTS = 3;
type EnsureRunForSessionParams = {
/**
* Session row to operate on. Caller is responsible for the env match —
* we don't re-check `runtimeEnvironmentId` against `environment.id`.
*/
session: Pick<
Session,
"id" | "taskIdentifier" | "triggerConfig" | "currentRunId" | "currentRunVersion"
>;
environment: AuthenticatedEnvironment;
reason: EnsureRunReason;
/**
* Shallow-merged on top of `triggerConfig.basePayload`. Runtime fields
* only — caller-controlled data that varies per trigger (`trigger:
* "preload"` vs `"trigger"`, etc).
*/
payloadOverrides?: Record<string, unknown>;
/**
* @internal Recursion-guard counter for the lost-claim-race retry path.
* Public callers should leave this unset; the function recurses with
* an incremented value on the pathological "winner's run was already
* terminal" branch and throws once it exceeds
* {@link ENSURE_RUN_FOR_SESSION_MAX_ATTEMPTS}.
*/
_attempt?: number;
};
export type EnsureRunResult = {
runId: string;
/** True if this call triggered a fresh run; false if it reused an alive existing one. */
triggered: boolean;
};
/**
* Idempotently make sure the session has a live run.
*
* Algorithm:
* 1. If `currentRunId` is set, probe its status. Alive → return as-is.
* 2. Trigger a new run upfront (cheap to cancel if we lose the race).
* 3. Atomic claim via `updateMany` keyed on `currentRunVersion`.
* - Won: return new runId, record SessionRun audit row.
* - Lost: cancel our triggered run, re-read session, reuse winner's
* run if alive. If pathological (winner's run already terminal),
* recurse.
*
* No DB lock is held across the trigger call. Wasted-trigger window is
* the rare multi-tab race on a dead run; cancel cost is negligible and
* the run-engine handles it gracefully.
*/
export async function ensureRunForSession(
params: EnsureRunForSessionParams
): Promise<EnsureRunResult> {
const { session, environment, reason, payloadOverrides, _attempt = 1 } = params;
if (_attempt > ENSURE_RUN_FOR_SESSION_MAX_ATTEMPTS) {
throw new SessionRunManagerError(
`ensureRunForSession exceeded ${ENSURE_RUN_FOR_SESSION_MAX_ATTEMPTS} attempts for session ${session.id} — every triggered run reached a terminal state before claim could resolve`
);
}
// 1. Probe currentRunId.
if (session.currentRunId) {
const status = await getRunStatus(session.currentRunId);
if (status && !isFinalRunStatus(status)) {
return { runId: session.currentRunId, triggered: false };
}
}
// 2. Validate config + trigger upfront.
const config = SessionTriggerConfigSchema.parse(session.triggerConfig);
const triggered = await triggerSessionRun({
session,
config,
environment,
payloadOverrides,
});
// 3. Try to claim the slot atomically.
const claim = await prisma.session.updateMany({
where: {
id: session.id,
currentRunVersion: session.currentRunVersion,
},
data: {
currentRunId: triggered.id,
currentRunVersion: { increment: 1 },
},
});
if (claim.count === 1) {
// Won. Audit the SessionRun. Best-effort — failure here doesn't
// invalidate the live run, just leaves a missing audit row.
prisma.sessionRun
.create({
data: { sessionId: session.id, runId: triggered.id, reason },
})
.catch((error) => {
logger.warn("Failed to record SessionRun audit row", {
sessionId: session.id,
runId: triggered.id,
reason,
error,
});
});
return { runId: triggered.id, triggered: true };
}
// 4. Lost the race. Cancel our triggered run; reuse the winner's.
cancelLostRaceRun(triggered.id, environment).catch((error) => {
logger.warn("Failed to cancel lost-race session run", {
sessionId: session.id,
runId: triggered.id,
error,
});
});
// Read-after-write: the winner just wrote `currentRunId` /
// `currentRunVersion` on the writer. Reading from `$replica` could
// return pre-race state and cause us to recurse with the same stale
// version, losing the next claim, until we exhaust max attempts.
const fresh = await prisma.session.findFirst({
where: { id: session.id },
select: {
id: true,
taskIdentifier: true,
triggerConfig: true,
currentRunId: true,
currentRunVersion: true,
},
});
if (!fresh) {
// Session vanished mid-flight. Surface as an error — caller decides
// whether to 404 or retry.
throw new SessionRunManagerError(`Session ${session.id} not found after lost claim race`);
}
if (fresh.currentRunId) {
const status = await getRunStatus(fresh.currentRunId);
if (status && !isFinalRunStatus(status)) {
return { runId: fresh.currentRunId, triggered: false };
}
}
// Pathological: winner's run already terminal. Recurse with the fresh
// version. Bounded by `ENSURE_RUN_FOR_SESSION_MAX_ATTEMPTS` so a task
// that always crashes before being dequeued surfaces as an error
// instead of a stack overflow.
return ensureRunForSession({
session: fresh,
environment,
reason,
payloadOverrides,
_attempt: _attempt + 1,
});
}
/**
* Trigger a single run for a session. Builds `TriggerTaskRequestBody`
* by shallow-merging `payloadOverrides` over `config.basePayload` and
* threading `config`'s machine/queue/tags through the trigger options.
*/
async function triggerSessionRun(params: {
session: Pick<Session, "id" | "taskIdentifier">;
config: SessionTriggerConfig;
environment: AuthenticatedEnvironment;
payloadOverrides?: Record<string, unknown>;
}): Promise<{ id: string; friendlyId: string }> {
const { session, config, environment, payloadOverrides } = params;
const payload = {
...config.basePayload,
...(config.idleTimeoutInSeconds !== undefined
? { idleTimeoutInSeconds: config.idleTimeoutInSeconds }
: {}),
...(payloadOverrides ?? {}),
};
const body = {
payload,
context: {},
options: {
...(config.machine ? { machine: config.machine as never } : {}),
...(config.queue ? { queue: { name: config.queue } } : {}),
...(config.tags ? { tags: config.tags } : {}),
...(config.maxAttempts !== undefined ? { maxAttempts: config.maxAttempts } : {}),
},
};
const service = new TriggerTaskService();
const result = await service.call(session.taskIdentifier, environment, body, {
triggerSource: "session",
triggerAction: "trigger",
});
if (!result) {
throw new SessionRunManagerError(
`TriggerTaskService returned no result for taskIdentifier=${session.taskIdentifier}`
);
}
return { id: result.run.id, friendlyId: result.run.friendlyId };
}
type SwapSessionRunParams = {
session: Pick<
Session,
"id" | "taskIdentifier" | "triggerConfig" | "currentRunId" | "currentRunVersion"
>;
/**
* The run requesting the swap. Optimistic claim requires
* `Session.currentRunId === callingRunId` so the swap can't clobber
* a run triggered out-of-band (e.g. a parallel `.in/append` probe
* that already replaced the dead run).
*/
callingRunId: string;
environment: AuthenticatedEnvironment;
reason: EnsureRunReason;
payloadOverrides?: Record<string, unknown>;
};
export type SwapSessionRunResult = {
/** runId of the newly-triggered run that has taken over the session. */
runId: string;
/**
* False when the swap was preempted (currentRunId is no longer the
* calling run). The caller should treat this as "someone else
* already moved on" — exit cleanly without expecting to drive the
* next run.
*/
swapped: boolean;
};
/**
* Force-swap the session to a freshly-triggered run, regardless of
* whether the current run is alive. Called by `end-and-continue` when
* the running agent wants a clean handoff (typically version upgrade).
*
* Differs from `ensureRunForSession`: never reuses the current run.
* The optimistic claim is keyed on `currentRunId === callingRunId`, so
* a parallel append-time probe that already swapped to a different
* run wins the race and `swapped: false` is surfaced.
*/
export async function swapSessionRun(
params: SwapSessionRunParams
): Promise<SwapSessionRunResult> {
const { session, callingRunId, environment, reason, payloadOverrides } = params;
const config = SessionTriggerConfigSchema.parse(session.triggerConfig);
const triggered = await triggerSessionRun({
session,
config,
environment,
payloadOverrides,
});
const claim = await prisma.session.updateMany({
where: {
id: session.id,
currentRunId: callingRunId,
currentRunVersion: session.currentRunVersion,
},
data: {
currentRunId: triggered.id,
currentRunVersion: { increment: 1 },
},
});
if (claim.count === 1) {
prisma.sessionRun
.create({
data: { sessionId: session.id, runId: triggered.id, reason },
})
.catch((error) => {
logger.warn("Failed to record SessionRun audit row", {
sessionId: session.id,
runId: triggered.id,
reason,
error,
});
});
return { runId: triggered.id, swapped: true };
}
// Lost the race — someone else already swapped to a new run. Cancel
// ours, surface the existing winner.
cancelLostRaceRun(triggered.id, environment).catch((error) => {
logger.warn("Failed to cancel preempted swap run", {
sessionId: session.id,
runId: triggered.id,
error,
});
});
// Read-after-write: the winner's swap was just committed on the
// writer. A replica read could return the pre-swap `currentRunId`
// (often `callingRunId` itself), which would tell the caller it is
// still the canonical run when in fact a different run has taken
// over.
const fresh = await prisma.session.findFirst({
where: { id: session.id },
select: { currentRunId: true },
});
return {
runId: fresh?.currentRunId ?? callingRunId,
swapped: false,
};
}
async function getRunStatus(runId: string): Promise<TaskRunStatus | null> {
// Use the read replica — this is a hot-path probe and stale-by-ms is
// fine. The append handler re-checks if it ends up reusing the runId.
const row = await $replica.taskRun.findFirst({
where: { id: runId },
select: { status: true },
});
return row?.status ?? null;
}
async function cancelLostRaceRun(
runId: string,
environment: AuthenticatedEnvironment
): Promise<void> {
const service = new CancelTaskRunService();
// Read-after-write: the run was just triggered on the writer, so go
// through `prisma`. A `$replica` miss here would silently no-op the
// cancel and leak an orphan run that no session is going to claim.
const run = await prisma.taskRun.findFirst({ where: { id: runId } });
if (!run) return;
await service.call(run, { reason: "Lost session-run claim race" });
}
export class SessionRunManagerError extends Error {
readonly name = "SessionRunManagerError";
}
@@ -0,0 +1,127 @@
import type { PrismaClient, Session } from "@trigger.dev/database";
import type { SessionItem } from "@trigger.dev/core/v3";
import { $replica } from "~/db.server";
/**
* Prefix that {@link SessionId.generate} attaches to every Session friendlyId.
* Used to distinguish friendlyId lookups (`session_abc...`) from externalId
* lookups on the public `GET /api/v1/sessions/:session` route.
*/
const SESSION_FRIENDLY_ID_PREFIX = "session_";
/**
* Resolve a session from a URL path parameter that may contain either a
* friendlyId (`session_abc...`) or a user-supplied externalId.
*
* Disambiguated by prefix: values starting with `session_` are treated as
* friendlyIds, anything else is looked up against `externalId` scoped to
* the caller's environment.
*/
export async function resolveSessionByIdOrExternalId(
prisma: Pick<PrismaClient, "session">,
runtimeEnvironmentId: string,
idOrExternalId: string
): Promise<Session | null> {
if (isSessionFriendlyIdForm(idOrExternalId)) {
return prisma.session.findFirst({
where: { friendlyId: idOrExternalId, runtimeEnvironmentId },
});
}
// `findFirst` rather than `findUnique` per the repo rule — `findUnique`'s
// implicit DataLoader has open correctness bugs in Prisma 6.x that bite
// hot-path lookups exactly like this one.
return prisma.session.findFirst({
where: { runtimeEnvironmentId, externalId: idOrExternalId },
});
}
/** True for `session_*` friendlyId form, false for everything else. */
export function isSessionFriendlyIdForm(value: string): boolean {
return value.startsWith(SESSION_FRIENDLY_ID_PREFIX);
}
/**
* Canonicalise the addressing key used for everything stream-level: the
* S2 stream path and the run-engine waitpoint cache key. `chat.agent`
* and the rest of the operational surface always pass `externalId`, but
* a public-API caller may legitimately address by `friendlyId` — and a
* session created without an `externalId` only has a friendlyId at all.
*
* Rule:
* - If we have a Session row, the canonical key is `externalId` if
* set, else `friendlyId`. This way two callers addressing the same
* row via different forms always converge to the same S2 stream.
* - If we have no row (yet — chat.agent's transport may subscribe
* before the agent's bind-time upsert lands), the canonical key is
* whatever the URL had. Operationally that's always an externalId.
* Friendlyid-form callers without a matching row are rejected by
* the route handler before this is reached.
*/
export function canonicalSessionAddressingKey(
row: Session | null,
paramSession: string
): string {
if (row) {
return row.externalId ?? row.friendlyId;
}
return paramSession;
}
/**
* Convert a Prisma `Session` row to the public {@link SessionItem} wire format.
* Strips internal columns (project/environment/organization ids) and narrows
* the `metadata` JSON to a record.
*
* Note: `currentRunId` is left as-is — Prisma stores the internal run id
* (cuid), but `SessionItem.currentRunId` is the *friendly* form. Routes
* that emit a single `SessionItem` should use
* {@link serializeSessionWithFriendlyRunId} instead, which resolves the
* friendlyId via a TaskRun lookup. List endpoints stay on this raw form
* to avoid N+1 lookups when paginating.
*/
export function serializeSession(session: Session): SessionItem {
return {
id: session.friendlyId,
externalId: session.externalId,
type: session.type,
taskIdentifier: session.taskIdentifier,
triggerConfig: session.triggerConfig as SessionItem["triggerConfig"],
currentRunId: session.currentRunId,
tags: session.tags,
metadata: (session.metadata ?? null) as SessionItem["metadata"],
closedAt: session.closedAt,
closedReason: session.closedReason,
expiresAt: session.expiresAt,
createdAt: session.createdAt,
updatedAt: session.updatedAt,
};
}
/**
* Same as {@link serializeSession} but resolves `currentRunId` from the
* internal cuid to the public `run_*` friendlyId via a TaskRun lookup.
* Single-row endpoints (`POST/GET/PATCH/close /api/v1/sessions/:s`) use
* this so the wire-side `currentRunId` is consistent with the rest of
* the public API (which only accepts friendlyIds for run lookups).
*
* Skips the lookup when `currentRunId` is null. The read goes through
* `$replica` — a TaskRun's `friendlyId` is immutable so replica lag is
* harmless, and serializing on the writer would just add hot-path load.
*/
export async function serializeSessionWithFriendlyRunId(
session: Session
): Promise<SessionItem> {
const base = serializeSession(session);
if (!session.currentRunId) return base;
const run = await $replica.taskRun.findFirst({
where: { id: session.currentRunId },
select: { friendlyId: true },
});
return {
...base,
currentRunId: run?.friendlyId ?? null,
};
}
@@ -33,6 +33,17 @@ export interface StreamIngestor {
export type StreamResponseOptions = {
timeoutInSeconds?: number;
lastEventId?: string;
/**
* Session-stream-only. When `true`, the responder MAY peek the tail
* of `.out` and short-circuit to `wait=0` + `X-Session-Settled: true`
* if the last chunk is a terminal marker (e.g. `trigger:turn-complete`).
* Used by `TriggerChatTransport.reconnectToStream` on page reload.
*
* When absent/false, the responder keeps the unconditional long-poll
* behavior — required on the active send-a-message path where the
* peek would race the newly-triggered turn's first chunk.
*/
peekSettled?: boolean;
};
// Interface for stream response
@@ -469,7 +469,13 @@ type ApiKeyActionRouteBuilderOptions<
: undefined,
body: TBodySchema extends z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>
? z.infer<TBodySchema>
: undefined
: undefined,
// The resolved resource from `findResource`. `undefined` when the route
// doesn't declare `findResource`. Routes that need to expand the auth
// scope to alternate identifiers of the same row (e.g. friendlyId +
// externalId for sessions) read it here so a JWT minted for either form
// authorizes both URL forms.
resource: TResource | undefined
) => AuthorizationResources;
superScopes?: string[];
};
@@ -667,9 +673,33 @@ export function createActionApiRoute<
parsedBody = body.data;
}
// Resolve the resource before authorization so the auth scope check
// can expand to alternate identifiers of the same row (e.g. a Session
// is addressable by both `friendlyId` and `externalId` and a JWT minted
// for either form should authorize both URL forms). Mirrors the
// ordering in `createLoaderApiRoute`.
const resource = options.findResource
? await options.findResource(parsedParams, authenticationResult, parsedSearchParams)
: undefined;
// Run authorization first — but with the resolved resource available
// as the 5th arg so the auth scope check can expand to alternate
// identifiers of the same row (e.g. a Session is addressable by both
// `friendlyId` and `externalId`). Resource-null is checked AFTER auth
// so:
// - underscoped JWT + missing resource → 403 (no info leak)
// - underscoped JWT + existing resource → 403 (existing behavior)
// - PRIVATE key + missing resource → auth passes → 404 (correct)
// - PRIVATE key + existing resource → auth passes → handler runs
if (authorization) {
const { action, resource, superScopes } = authorization;
const $resource = resource(parsedParams, parsedSearchParams, parsedHeaders, parsedBody);
const { action, resource: authResource, superScopes } = authorization;
const $resource = authResource(
parsedParams,
parsedSearchParams,
parsedHeaders,
parsedBody,
resource
);
logger.debug("Checking authorization", {
action,
@@ -702,10 +732,6 @@ export function createActionApiRoute<
}
}
const resource = options.findResource
? await options.findResource(parsedParams, authenticationResult, parsedSearchParams)
: undefined;
if (options.findResource && !resource) {
return await wrapResponse(
request,
@@ -0,0 +1,147 @@
import { Redis } from "ioredis";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
import { logger } from "./logger.server";
// "ssw" — session-stream-waitpoint. Parallel to the input-stream variant
// (`isw:{runFriendlyId}:{streamId}`). Keyed purely on `{sessionId, io}` so
// a send() lands on the channel regardless of which run is waiting, and
// multiple concurrent waiters (e.g. two agents on one chat) all wake.
const KEY_PREFIX = "ssw:";
const DEFAULT_TTL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days
function buildKey(sessionFriendlyId: string, io: "out" | "in"): string {
return `${KEY_PREFIX}${sessionFriendlyId}:${io}`;
}
function initializeRedis(): Redis | undefined {
const host = env.CACHE_REDIS_HOST;
if (!host) {
return undefined;
}
return new Redis({
connectionName: "sessionStreamWaitpointCache",
host,
port: env.CACHE_REDIS_PORT,
username: env.CACHE_REDIS_USERNAME,
password: env.CACHE_REDIS_PASSWORD,
keyPrefix: "tr:",
enableAutoPipelining: true,
...(env.CACHE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
});
}
const redis = singleton("sessionStreamWaitpointCache", initializeRedis);
// Atomic SADD + PEXPIRE that only ever extends the key's TTL.
//
// Two concerns rolled into one script:
// 1. SADD + PEXPIRE as separate commands can leave the key with no TTL
// if the second call fails (or the process crashes in between).
// 2. Each waitpoint registers with its own `ttlMs` (derived from the
// waitpoint's timeout). Calling PEXPIRE unconditionally would let a
// short-TTL registration shrink the key's TTL below a longer-TTL
// sibling — evicting the sibling early and degrading the append-path
// fast drain to engine-timeout-only.
//
// The script: SADD the member, then set PEXPIRE only if the new TTL is
// greater than the current PTTL (or the key has no TTL yet). Engine-
// level timeouts still fire per-waitpoint; this keeps the Redis key
// alive for the longest-lived member.
const ADD_WAITPOINT_SCRIPT = `
redis.call("SADD", KEYS[1], ARGV[1])
local newTtl = tonumber(ARGV[2])
local currentTtl = redis.call("PTTL", KEYS[1])
if currentTtl < 0 or newTtl > currentTtl then
redis.call("PEXPIRE", KEYS[1], newTtl)
end
return 1
`;
/**
* Register a waitpoint as pending on the given session channel. Called
* from the `.wait()` create-waitpoint route. Multiple waiters on the same
* channel are allowed (stored as a Redis set).
*/
export async function addSessionStreamWaitpoint(
sessionFriendlyId: string,
io: "out" | "in",
waitpointId: string,
ttlMs?: number
): Promise<void> {
if (!redis) return;
try {
const key = buildKey(sessionFriendlyId, io);
await redis.eval(
ADD_WAITPOINT_SCRIPT,
1,
key,
waitpointId,
String(ttlMs ?? DEFAULT_TTL_MS)
);
} catch (error) {
logger.error("Failed to set session stream waitpoint cache", {
sessionFriendlyId,
io,
error,
});
}
}
/**
* Atomically read + clear all waitpoints registered on the given session
* channel. Called from the append handler so the next append sees an
* empty set even if two appends race.
*/
export async function drainSessionStreamWaitpoints(
sessionFriendlyId: string,
io: "out" | "in"
): Promise<string[]> {
if (!redis) return [];
try {
const key = buildKey(sessionFriendlyId, io);
const pipeline = redis.multi();
pipeline.smembers(key);
pipeline.del(key);
const results = await pipeline.exec();
if (!results) return [];
const [smembersResult] = results;
if (!smembersResult) return [];
const [err, members] = smembersResult;
if (err) return [];
return Array.isArray(members) ? (members as string[]) : [];
} catch (error) {
logger.error("Failed to drain session stream waitpoint cache", {
sessionFriendlyId,
io,
error,
});
return [];
}
}
/**
* Remove a single waitpoint from the pending set. Called after a race
* where `.wait()` completed the waitpoint from pre-arrived data.
*/
export async function removeSessionStreamWaitpoint(
sessionFriendlyId: string,
io: "out" | "in",
waitpointId: string
): Promise<void> {
if (!redis) return;
try {
const key = buildKey(sessionFriendlyId, io);
await redis.srem(key, waitpointId);
} catch (error) {
logger.error("Failed to remove session stream waitpoint cache entry", {
sessionFriendlyId,
io,
error,
});
}
}
@@ -0,0 +1,72 @@
import { ClickHouse } from "@internal/clickhouse";
import invariant from "tiny-invariant";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
import { meter, provider } from "~/v3/tracer.server";
import { SessionsReplicationService } from "./sessionsReplicationService.server";
export const sessionsReplicationInstance = singleton(
"sessionsReplicationInstance",
initializeSessionsReplicationInstance
);
function initializeSessionsReplicationInstance() {
const { DATABASE_URL } = process.env;
invariant(typeof DATABASE_URL === "string", "DATABASE_URL env var not set");
if (!env.SESSION_REPLICATION_CLICKHOUSE_URL) {
console.log("🗃️ Sessions replication service not enabled");
return;
}
console.log("🗃️ Sessions replication service enabled");
const clickhouse = new ClickHouse({
url: env.SESSION_REPLICATION_CLICKHOUSE_URL,
name: "sessions-replication",
keepAlive: {
enabled: env.SESSION_REPLICATION_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.SESSION_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.SESSION_REPLICATION_CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.SESSION_REPLICATION_MAX_OPEN_CONNECTIONS,
});
const service = new SessionsReplicationService({
clickhouse: clickhouse,
pgConnectionUrl: DATABASE_URL,
serviceName: "sessions-replication",
slotName: env.SESSION_REPLICATION_SLOT_NAME,
publicationName: env.SESSION_REPLICATION_PUBLICATION_NAME,
redisOptions: {
keyPrefix: "sessions-replication:",
port: env.RUN_REPLICATION_REDIS_PORT ?? undefined,
host: env.RUN_REPLICATION_REDIS_HOST ?? undefined,
username: env.RUN_REPLICATION_REDIS_USERNAME ?? undefined,
password: env.RUN_REPLICATION_REDIS_PASSWORD ?? undefined,
enableAutoPipelining: true,
...(env.RUN_REPLICATION_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
},
maxFlushConcurrency: env.SESSION_REPLICATION_MAX_FLUSH_CONCURRENCY,
flushIntervalMs: env.SESSION_REPLICATION_FLUSH_INTERVAL_MS,
flushBatchSize: env.SESSION_REPLICATION_FLUSH_BATCH_SIZE,
leaderLockTimeoutMs: env.SESSION_REPLICATION_LEADER_LOCK_TIMEOUT_MS,
leaderLockExtendIntervalMs: env.SESSION_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS,
leaderLockAcquireAdditionalTimeMs: env.SESSION_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS,
leaderLockRetryIntervalMs: env.SESSION_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS,
ackIntervalSeconds: env.SESSION_REPLICATION_ACK_INTERVAL_SECONDS,
logLevel: env.SESSION_REPLICATION_LOG_LEVEL,
waitForAsyncInsert: env.SESSION_REPLICATION_WAIT_FOR_ASYNC_INSERT === "1",
tracer: provider.getTracer("sessions-replication-service"),
meter,
insertMaxRetries: env.SESSION_REPLICATION_INSERT_MAX_RETRIES,
insertBaseDelayMs: env.SESSION_REPLICATION_INSERT_BASE_DELAY_MS,
insertMaxDelayMs: env.SESSION_REPLICATION_INSERT_MAX_DELAY_MS,
insertStrategy: env.SESSION_REPLICATION_INSERT_STRATEGY,
});
return service;
}
@@ -0,0 +1,763 @@
import type { ClickHouse, SessionInsertArray } from "@internal/clickhouse";
import { getSessionField } from "@internal/clickhouse";
import { type RedisOptions } from "@internal/redis";
import {
LogicalReplicationClient,
type MessageDelete,
type MessageInsert,
type MessageUpdate,
type PgoutputMessage,
} from "@internal/replication";
import {
getMeter,
recordSpanError,
startSpan,
trace,
type Counter,
type Histogram,
type Meter,
type Tracer,
} from "@internal/tracing";
import { Logger, type LogLevel } from "@trigger.dev/core/logger";
import { tryCatch } from "@trigger.dev/core/utils";
import { type Session } from "@trigger.dev/database";
import EventEmitter from "node:events";
import { ConcurrentFlushScheduler } from "./runsReplicationService.server";
interface TransactionEvent<T = any> {
tag: "insert" | "update" | "delete";
data: T;
raw: MessageInsert | MessageUpdate | MessageDelete;
}
interface Transaction<T = any> {
beginStartTimestamp: number;
commitLsn: string | null;
commitEndLsn: string | null;
xid: number;
events: TransactionEvent<T>[];
replicationLagMs: number;
}
export type SessionsReplicationServiceOptions = {
clickhouse: ClickHouse;
pgConnectionUrl: string;
serviceName: string;
slotName: string;
publicationName: string;
redisOptions: RedisOptions;
maxFlushConcurrency?: number;
flushIntervalMs?: number;
flushBatchSize?: number;
leaderLockTimeoutMs?: number;
leaderLockExtendIntervalMs?: number;
leaderLockAcquireAdditionalTimeMs?: number;
leaderLockRetryIntervalMs?: number;
ackIntervalSeconds?: number;
acknowledgeTimeoutMs?: number;
logger?: Logger;
logLevel?: LogLevel;
tracer?: Tracer;
meter?: Meter;
waitForAsyncInsert?: boolean;
insertStrategy?: "insert" | "insert_async";
// Retry configuration for insert operations
insertMaxRetries?: number;
insertBaseDelayMs?: number;
insertMaxDelayMs?: number;
};
type SessionInsert = {
_version: bigint;
session: Session;
event: "insert" | "update" | "delete";
};
export type SessionsReplicationServiceEvents = {
message: [{ lsn: string; message: PgoutputMessage; service: SessionsReplicationService }];
batchFlushed: [{ flushId: string; sessionInserts: SessionInsertArray[] }];
};
export class SessionsReplicationService {
private _isSubscribed = false;
private _currentTransaction:
| (Omit<Transaction<Session>, "commitEndLsn" | "replicationLagMs"> & {
commitEndLsn?: string | null;
replicationLagMs?: number;
})
| null = null;
private _replicationClient: LogicalReplicationClient;
private _concurrentFlushScheduler: ConcurrentFlushScheduler<SessionInsert>;
private logger: Logger;
private _isShuttingDown = false;
private _isShutDownComplete = false;
private _tracer: Tracer;
private _meter: Meter;
private _currentParseDurationMs: number | null = null;
private _lastAcknowledgedAt: number | null = null;
private _acknowledgeTimeoutMs: number;
private _latestCommitEndLsn: string | null = null;
private _lastAcknowledgedLsn: string | null = null;
private _acknowledgeInterval: NodeJS.Timeout | null = null;
// Retry configuration
private _insertMaxRetries: number;
private _insertBaseDelayMs: number;
private _insertMaxDelayMs: number;
private _insertStrategy: "insert" | "insert_async";
// Metrics
private _replicationLagHistogram: Histogram;
private _batchesFlushedCounter: Counter;
private _batchSizeHistogram: Histogram;
private _sessionsInsertedCounter: Counter;
private _insertRetriesCounter: Counter;
private _eventsProcessedCounter: Counter;
private _flushDurationHistogram: Histogram;
public readonly events: EventEmitter<SessionsReplicationServiceEvents>;
constructor(private readonly options: SessionsReplicationServiceOptions) {
this.logger =
options.logger ?? new Logger("SessionsReplicationService", options.logLevel ?? "info");
this.events = new EventEmitter();
this._tracer = options.tracer ?? trace.getTracer("sessions-replication-service");
this._meter = options.meter ?? getMeter("sessions-replication");
// Initialize metrics
this._replicationLagHistogram = this._meter.createHistogram(
"sessions_replication.replication_lag_ms",
{
description: "Replication lag from Postgres commit to processing",
unit: "ms",
}
);
this._batchesFlushedCounter = this._meter.createCounter(
"sessions_replication.batches_flushed",
{
description: "Total batches flushed to ClickHouse",
}
);
this._batchSizeHistogram = this._meter.createHistogram("sessions_replication.batch_size", {
description: "Number of items per batch flush",
unit: "items",
});
this._sessionsInsertedCounter = this._meter.createCounter(
"sessions_replication.sessions_inserted",
{
description: "Session inserts to ClickHouse",
unit: "inserts",
}
);
this._insertRetriesCounter = this._meter.createCounter("sessions_replication.insert_retries", {
description: "Insert retry attempts",
});
this._eventsProcessedCounter = this._meter.createCounter(
"sessions_replication.events_processed",
{
description: "Replication events processed (inserts, updates, deletes)",
}
);
this._flushDurationHistogram = this._meter.createHistogram(
"sessions_replication.flush_duration_ms",
{
description: "Duration of batch flush operations",
unit: "ms",
}
);
this._acknowledgeTimeoutMs = options.acknowledgeTimeoutMs ?? 1_000;
this._insertStrategy = options.insertStrategy ?? "insert";
this._replicationClient = new LogicalReplicationClient({
pgConfig: {
connectionString: options.pgConnectionUrl,
},
name: options.serviceName,
slotName: options.slotName,
publicationName: options.publicationName,
table: "Session",
redisOptions: options.redisOptions,
autoAcknowledge: false,
publicationActions: ["insert", "update", "delete"],
logger: options.logger ?? new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
leaderLockTimeoutMs: options.leaderLockTimeoutMs ?? 30_000,
leaderLockExtendIntervalMs: options.leaderLockExtendIntervalMs ?? 10_000,
ackIntervalSeconds: options.ackIntervalSeconds ?? 10,
leaderLockAcquireAdditionalTimeMs: options.leaderLockAcquireAdditionalTimeMs ?? 10_000,
leaderLockRetryIntervalMs: options.leaderLockRetryIntervalMs ?? 500,
tracer: options.tracer,
});
this._concurrentFlushScheduler = new ConcurrentFlushScheduler<SessionInsert>({
batchSize: options.flushBatchSize ?? 50,
flushInterval: options.flushIntervalMs ?? 100,
maxConcurrency: options.maxFlushConcurrency ?? 100,
callback: this.#flushBatch.bind(this),
// Key-based deduplication to reduce duplicates sent to ClickHouse
getKey: (item) => {
if (!item?.session?.id) {
this.logger.warn("Skipping replication event with null session", { event: item });
return null;
}
return `${item.event}_${item.session.id}`;
},
// Keep the session with the higher version (latest)
// and take the last occurrence for that version.
// Items originating from the same DB transaction have the same version.
shouldReplace: (existing, incoming) => incoming._version >= existing._version,
logger: new Logger("ConcurrentFlushScheduler", options.logLevel ?? "info"),
tracer: options.tracer,
});
this._replicationClient.events.on("data", async ({ lsn, log, parseDuration }) => {
this.#handleData(lsn, log, parseDuration);
});
this._replicationClient.events.on("heartbeat", async ({ lsn, shouldRespond }) => {
if (this._isShuttingDown) return;
if (this._isShutDownComplete) return;
if (shouldRespond) {
this._lastAcknowledgedLsn = lsn;
await this._replicationClient.acknowledge(lsn);
}
});
this._replicationClient.events.on("error", (error) => {
this.logger.error("Replication client error", {
error,
});
});
this._replicationClient.events.on("start", () => {
this.logger.info("Replication client started");
});
this._replicationClient.events.on("acknowledge", ({ lsn }) => {
this.logger.debug("Acknowledged", { lsn });
});
this._replicationClient.events.on("leaderElection", (isLeader) => {
this.logger.info("Leader election", { isLeader });
});
// Initialize retry configuration
this._insertMaxRetries = options.insertMaxRetries ?? 3;
this._insertBaseDelayMs = options.insertBaseDelayMs ?? 100;
this._insertMaxDelayMs = options.insertMaxDelayMs ?? 2000;
}
public async shutdown() {
if (this._isShuttingDown) return;
this._isShuttingDown = true;
this.logger.info("Initiating shutdown of sessions replication service");
if (!this._currentTransaction) {
this.logger.info("No transaction to commit, shutting down immediately");
await this._replicationClient.stop();
this._isSubscribed = false;
this._isShutDownComplete = true;
return;
}
this._concurrentFlushScheduler.shutdown();
}
async start() {
if (this._isSubscribed) {
this.logger.debug("Replication client already started, skipping start");
return;
}
this.logger.info("Starting replication client", {
lastLsn: this._latestCommitEndLsn,
});
await this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined);
this._acknowledgeInterval = setInterval(this.#acknowledgeLatestTransaction.bind(this), 1000);
this._concurrentFlushScheduler.start();
this._isSubscribed = true;
}
async stop() {
this.logger.info("Stopping replication client");
await this._replicationClient.stop();
if (this._acknowledgeInterval) {
clearInterval(this._acknowledgeInterval);
this._acknowledgeInterval = null;
}
this._isSubscribed = false;
}
async teardown() {
this.logger.info("Teardown replication client");
await this._replicationClient.teardown();
if (this._acknowledgeInterval) {
clearInterval(this._acknowledgeInterval);
this._acknowledgeInterval = null;
}
this._isSubscribed = false;
}
#handleData(lsn: string, message: PgoutputMessage, parseDuration: bigint) {
this.logger.debug("Handling data", {
lsn,
tag: message.tag,
parseDuration,
});
this.events.emit("message", { lsn, message, service: this });
switch (message.tag) {
case "begin": {
if (this._isShuttingDown || this._isShutDownComplete) {
return;
}
this._currentTransaction = {
beginStartTimestamp: Date.now(),
commitLsn: message.commitLsn,
xid: message.xid,
events: [],
};
this._currentParseDurationMs = Number(parseDuration) / 1_000_000;
break;
}
case "insert": {
if (!this._currentTransaction) {
return;
}
if (this._currentParseDurationMs) {
this._currentParseDurationMs =
this._currentParseDurationMs + Number(parseDuration) / 1_000_000;
}
this._currentTransaction.events.push({
tag: message.tag,
data: message.new as Session,
raw: message,
});
break;
}
case "update": {
if (!this._currentTransaction) {
return;
}
if (this._currentParseDurationMs) {
this._currentParseDurationMs =
this._currentParseDurationMs + Number(parseDuration) / 1_000_000;
}
this._currentTransaction.events.push({
tag: message.tag,
data: message.new as Session,
raw: message,
});
break;
}
case "delete": {
if (!this._currentTransaction) {
return;
}
if (this._currentParseDurationMs) {
this._currentParseDurationMs =
this._currentParseDurationMs + Number(parseDuration) / 1_000_000;
}
this._currentTransaction.events.push({
tag: message.tag,
data: message.old as Session,
raw: message,
});
break;
}
case "commit": {
if (!this._currentTransaction) {
return;
}
if (this._currentParseDurationMs) {
this._currentParseDurationMs =
this._currentParseDurationMs + Number(parseDuration) / 1_000_000;
}
const replicationLagMs = Date.now() - Number(message.commitTime / 1000n);
this._currentTransaction.commitEndLsn = message.commitEndLsn;
this._currentTransaction.replicationLagMs = replicationLagMs;
const transaction = this._currentTransaction as Transaction<Session>;
this._currentTransaction = null;
if (transaction.commitEndLsn) {
this._latestCommitEndLsn = transaction.commitEndLsn;
}
this.#handleTransaction(transaction);
break;
}
default: {
this.logger.debug("Unknown message tag", {
pgMessage: message,
});
}
}
}
#handleTransaction(transaction: Transaction<Session>) {
if (this._isShutDownComplete) return;
if (this._isShuttingDown) {
this._replicationClient.stop().finally(() => {
this._isSubscribed = false;
this._isShutDownComplete = true;
});
}
// If there are no events, do nothing
if (transaction.events.length === 0) {
return;
}
if (!transaction.commitEndLsn) {
this.logger.error("Transaction has no commit end lsn", {
transaction,
});
return;
}
const lsnToUInt64Start = process.hrtime.bigint();
// If there are events, we need to handle them
const _version = lsnToUInt64(transaction.commitEndLsn);
const lsnToUInt64DurationMs = Number(process.hrtime.bigint() - lsnToUInt64Start) / 1_000_000;
this._concurrentFlushScheduler.addToBatch(
transaction.events.map((event) => ({
_version,
session: event.data,
event: event.tag,
}))
);
// Record metrics
this._replicationLagHistogram.record(transaction.replicationLagMs);
// Count events by type
for (const event of transaction.events) {
this._eventsProcessedCounter.add(1, { event_type: event.tag });
}
this.logger.debug("handle_transaction", {
transaction: {
xid: transaction.xid,
commitLsn: transaction.commitLsn,
commitEndLsn: transaction.commitEndLsn,
events: transaction.events.length,
parseDurationMs: this._currentParseDurationMs,
lsnToUInt64DurationMs,
version: _version.toString(),
},
});
}
async #acknowledgeLatestTransaction() {
if (!this._latestCommitEndLsn) {
return;
}
if (this._lastAcknowledgedLsn === this._latestCommitEndLsn) {
return;
}
const now = Date.now();
if (this._lastAcknowledgedAt) {
const timeSinceLastAcknowledged = now - this._lastAcknowledgedAt;
// If we've already acknowledged within the last second, don't acknowledge again
if (timeSinceLastAcknowledged < this._acknowledgeTimeoutMs) {
return;
}
}
this._lastAcknowledgedAt = now;
this._lastAcknowledgedLsn = this._latestCommitEndLsn;
this.logger.debug("acknowledge_latest_transaction", {
commitEndLsn: this._latestCommitEndLsn,
lastAcknowledgedAt: this._lastAcknowledgedAt,
});
const [ackError] = await tryCatch(
this._replicationClient.acknowledge(this._latestCommitEndLsn)
);
if (ackError) {
this.logger.error("Error acknowledging transaction", { ackError });
}
if (this._isShutDownComplete && this._acknowledgeInterval) {
clearInterval(this._acknowledgeInterval);
}
}
async #flushBatch(flushId: string, batch: Array<SessionInsert>) {
if (batch.length === 0) {
return;
}
this.logger.debug("Flushing batch", {
flushId,
batchSize: batch.length,
});
const flushStartTime = performance.now();
await startSpan(this._tracer, "flushBatch", async (span) => {
const sessionInserts = batch
.map((item) => toSessionInsertArray(item.session, item._version, item.event === "delete"))
// batch inserts in clickhouse are more performant if the items
// are pre-sorted by the primary key
.sort((a, b) => {
const aOrgId = getSessionField(a, "organization_id");
const bOrgId = getSessionField(b, "organization_id");
if (aOrgId !== bOrgId) {
return aOrgId < bOrgId ? -1 : 1;
}
const aProjId = getSessionField(a, "project_id");
const bProjId = getSessionField(b, "project_id");
if (aProjId !== bProjId) {
return aProjId < bProjId ? -1 : 1;
}
const aEnvId = getSessionField(a, "environment_id");
const bEnvId = getSessionField(b, "environment_id");
if (aEnvId !== bEnvId) {
return aEnvId < bEnvId ? -1 : 1;
}
const aCreatedAt = getSessionField(a, "created_at");
const bCreatedAt = getSessionField(b, "created_at");
if (aCreatedAt !== bCreatedAt) {
return aCreatedAt - bCreatedAt;
}
const aSessionId = getSessionField(a, "session_id");
const bSessionId = getSessionField(b, "session_id");
if (aSessionId === bSessionId) return 0;
return aSessionId < bSessionId ? -1 : 1;
});
span.setAttribute("session_inserts", sessionInserts.length);
this.logger.debug("Flushing inserts", {
flushId,
sessionInserts: sessionInserts.length,
});
const [sessionError, sessionResult] = await this.#insertWithRetry(
(attempt) => this.#insertSessionInserts(sessionInserts, attempt),
"session inserts",
flushId
);
if (sessionError) {
this.logger.error("Error inserting session inserts", {
error: sessionError,
flushId,
});
recordSpanError(span, sessionError);
}
this.logger.debug("Flushed inserts", {
flushId,
sessionInserts: sessionInserts.length,
});
this.events.emit("batchFlushed", { flushId, sessionInserts });
// Record metrics
const flushDurationMs = performance.now() - flushStartTime;
const hasErrors = sessionError !== null;
this._batchSizeHistogram.record(batch.length);
this._flushDurationHistogram.record(flushDurationMs);
this._batchesFlushedCounter.add(1, { success: !hasErrors });
if (!sessionError) {
this._sessionsInsertedCounter.add(sessionInserts.length);
}
});
}
// New method to handle inserts with retry logic for connection errors
async #insertWithRetry<T>(
insertFn: (attempt: number) => Promise<T>,
operationName: string,
flushId: string
): Promise<[Error | null, T | null]> {
let lastError: Error | null = null;
for (let attempt = 1; attempt <= this._insertMaxRetries; attempt++) {
try {
const result = await insertFn(attempt);
return [null, result];
} catch (error) {
lastError = error instanceof Error ? error : new Error(String(error));
// Check if this is a retryable error
if (this.#isRetryableError(lastError)) {
const delay = this.#calculateRetryDelay(attempt);
this.logger.warn(`Retrying SessionsReplication insert due to error`, {
operationName,
flushId,
attempt,
maxRetries: this._insertMaxRetries,
error: lastError.message,
delay,
});
// Record retry metric
this._insertRetriesCounter.add(1, { operation: "sessions" });
await new Promise((resolve) => setTimeout(resolve, delay));
continue;
}
break;
}
}
return [lastError, null];
}
// Retry all errors except known permanent ones
#isRetryableError(error: Error): boolean {
const errorMessage = error.message.toLowerCase();
// Permanent errors that should NOT be retried
const permanentErrorPatterns = [
"authentication failed",
"permission denied",
"invalid credentials",
"table not found",
"database not found",
"column not found",
"schema mismatch",
"invalid query",
"syntax error",
"type error",
"constraint violation",
"duplicate key",
"foreign key violation",
];
// If it's a known permanent error, don't retry
if (permanentErrorPatterns.some((pattern) => errorMessage.includes(pattern))) {
return false;
}
// Retry everything else
return true;
}
#calculateRetryDelay(attempt: number): number {
// Exponential backoff: baseDelay, baseDelay*2, baseDelay*4, etc.
const delay = Math.min(
this._insertBaseDelayMs * Math.pow(2, attempt - 1),
this._insertMaxDelayMs
);
// Add some jitter to prevent thundering herd
const jitter = Math.random() * 100;
return delay + jitter;
}
#getClickhouseInsertSettings() {
if (this._insertStrategy === "insert") {
return {};
}
return {
async_insert: 1 as const,
async_insert_max_data_size: "1000000",
async_insert_busy_timeout_ms: 1000,
wait_for_async_insert: this.options.waitForAsyncInsert ? (1 as const) : (0 as const),
};
}
async #insertSessionInserts(sessionInserts: SessionInsertArray[], attempt: number) {
return await startSpan(this._tracer, "insertSessionInserts", async (span) => {
const [insertError, insertResult] =
await this.options.clickhouse.sessions.insertCompactArrays(sessionInserts, {
params: {
clickhouse_settings: this.#getClickhouseInsertSettings(),
},
});
if (insertError) {
this.logger.error("Error inserting session inserts attempt", {
error: insertError,
attempt,
});
recordSpanError(span, insertError);
throw insertError;
}
return insertResult;
});
}
}
function toSessionInsertArray(
session: Session,
version: bigint,
isDeleted: boolean
): SessionInsertArray {
return [
session.runtimeEnvironmentId,
session.organizationId,
session.projectId,
session.id,
session.environmentType,
session.friendlyId,
session.externalId ?? "",
session.type,
session.taskIdentifier ?? "",
session.tags ?? [],
{ data: session.metadata ?? null },
session.closedAt ? session.closedAt.getTime() : null,
session.closedReason ?? "",
session.expiresAt ? session.expiresAt.getTime() : null,
session.createdAt.getTime(),
session.updatedAt.getTime(),
version.toString(),
isDeleted ? 1 : 0,
];
}
function lsnToUInt64(lsn: string): bigint {
const [seg, off] = lsn.split("/");
return (BigInt("0x" + seg) << 32n) | BigInt("0x" + off);
}
@@ -0,0 +1,254 @@
import { type ClickhouseQueryBuilder } from "@internal/clickhouse";
import parseDuration from "parse-duration";
import {
convertSessionListInputOptionsToFilterOptions,
type FilterSessionsOptions,
type ISessionsRepository,
type ListSessionsOptions,
type SessionListInputOptions,
type SessionTagListOptions,
type SessionsRepositoryOptions,
} from "./sessionsRepository.server";
export class ClickHouseSessionsRepository implements ISessionsRepository {
constructor(private readonly options: SessionsRepositoryOptions) {}
get name() {
return "clickhouse";
}
async listSessionIds(options: ListSessionsOptions): Promise<string[]> {
const queryBuilder = this.options.clickhouse.sessions.queryBuilder();
applySessionFiltersToQueryBuilder(
queryBuilder,
convertSessionListInputOptionsToFilterOptions(options)
);
if (options.page.cursor) {
if (options.page.direction === "forward" || !options.page.direction) {
queryBuilder
.where("session_id < {sessionId: String}", { sessionId: options.page.cursor })
.orderBy("created_at DESC, session_id DESC")
.limit(options.page.size + 1);
} else {
queryBuilder
.where("session_id > {sessionId: String}", { sessionId: options.page.cursor })
.orderBy("created_at ASC, session_id ASC")
.limit(options.page.size + 1);
}
} else {
queryBuilder.orderBy("created_at DESC, session_id DESC").limit(options.page.size + 1);
}
const [queryError, result] = await queryBuilder.execute();
if (queryError) throw queryError;
return result.map((row) => row.session_id);
}
async listSessions(options: ListSessionsOptions) {
const sessionIds = await this.listSessionIds(options);
const hasMore = sessionIds.length > options.page.size;
let nextCursor: string | null = null;
let previousCursor: string | null = null;
const direction = options.page.direction ?? "forward";
switch (direction) {
case "forward": {
previousCursor = options.page.cursor ? sessionIds.at(0) ?? null : null;
if (hasMore) {
nextCursor = sessionIds[options.page.size - 1];
}
break;
}
case "backward": {
const reversed = [...sessionIds].reverse();
if (hasMore) {
previousCursor = reversed.at(1) ?? null;
nextCursor = reversed.at(options.page.size) ?? null;
} else {
nextCursor = reversed.at(options.page.size - 1) ?? null;
}
break;
}
}
// Both directions slice the first `size` IDs: the `size+1`th item is
// the sentinel proving another page exists (hasMore), not part of the
// page content. Backward queries sort ASC (items closest to the cursor
// first), so `[0..size)` is still the legitimate window and the last
// element is the sentinel — identical to the forward case.
const idsToReturn = sessionIds.slice(0, options.page.size);
let sessions = await this.options.prisma.session.findMany({
where: {
id: { in: idsToReturn },
runtimeEnvironmentId: options.environmentId,
},
orderBy: { createdAt: "desc" },
select: {
id: true,
friendlyId: true,
externalId: true,
type: true,
taskIdentifier: true,
tags: true,
metadata: true,
closedAt: true,
closedReason: true,
expiresAt: true,
createdAt: true,
updatedAt: true,
runtimeEnvironmentId: true,
},
});
// ClickHouse is slightly delayed; narrow by derived status in-memory to
// catch recent Postgres writes that haven't replicated yet.
if (options.statuses && options.statuses.length > 0) {
const wanted = new Set(options.statuses);
const now = Date.now();
sessions = sessions.filter((s) => {
const status =
s.closedAt != null
? "CLOSED"
: s.expiresAt != null && s.expiresAt.getTime() < now
? "EXPIRED"
: "ACTIVE";
return wanted.has(status);
});
}
return {
sessions,
pagination: { nextCursor, previousCursor },
};
}
async countSessions(options: SessionListInputOptions): Promise<number> {
const queryBuilder = this.options.clickhouse.sessions.countQueryBuilder();
applySessionFiltersToQueryBuilder(
queryBuilder,
convertSessionListInputOptionsToFilterOptions(options)
);
const [queryError, result] = await queryBuilder.execute();
if (queryError) throw queryError;
if (result.length === 0) {
throw new Error("No count rows returned");
}
return result[0].count;
}
async listTags(options: SessionTagListOptions) {
const queryBuilder = this.options.clickhouse.sessions
.tagQueryBuilder()
.where("organization_id = {organizationId: String}", {
organizationId: options.organizationId,
})
.where("project_id = {projectId: String}", { projectId: options.projectId })
.where("environment_id = {environmentId: String}", {
environmentId: options.environmentId,
});
const periodMs = options.period ? parseDuration(options.period) ?? undefined : undefined;
if (periodMs) {
queryBuilder.where("created_at >= fromUnixTimestamp64Milli({period: Int64})", {
period: new Date(Date.now() - periodMs).getTime(),
});
}
if (options.from) {
queryBuilder.where("created_at >= fromUnixTimestamp64Milli({from: Int64})", {
from: options.from,
});
}
if (options.to) {
queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", {
to: options.to,
});
}
if (options.query && options.query.trim().length > 0) {
queryBuilder.where("positionCaseInsensitiveUTF8(tag, {query: String}) > 0", {
query: options.query,
});
}
queryBuilder.orderBy("tag ASC").limit(options.limit);
const [queryError, result] = await queryBuilder.execute();
if (queryError) throw queryError;
return { tags: result.map((row) => row.tag) };
}
}
function applySessionFiltersToQueryBuilder<T>(
queryBuilder: ClickhouseQueryBuilder<T>,
options: FilterSessionsOptions
) {
queryBuilder
.where("organization_id = {organizationId: String}", {
organizationId: options.organizationId,
})
.where("project_id = {projectId: String}", { projectId: options.projectId })
.where("environment_id = {environmentId: String}", { environmentId: options.environmentId });
if (options.types && options.types.length > 0) {
queryBuilder.where("type IN {types: Array(String)}", { types: options.types });
}
if (options.tags && options.tags.length > 0) {
queryBuilder.where("hasAny(tags, {tags: Array(String)})", { tags: options.tags });
}
if (options.taskIdentifiers && options.taskIdentifiers.length > 0) {
queryBuilder.where("task_identifier IN {taskIdentifiers: Array(String)}", {
taskIdentifiers: options.taskIdentifiers,
});
}
if (options.externalId) {
queryBuilder.where("external_id = {externalId: String}", { externalId: options.externalId });
}
if (options.statuses && options.statuses.length > 0) {
const conditions: string[] = [];
if (options.statuses.includes("ACTIVE")) {
conditions.push(
"(closed_at IS NULL AND (expires_at IS NULL OR expires_at > now64(3)))"
);
}
if (options.statuses.includes("CLOSED")) {
conditions.push("closed_at IS NOT NULL");
}
if (options.statuses.includes("EXPIRED")) {
conditions.push("(closed_at IS NULL AND expires_at IS NOT NULL AND expires_at <= now64(3))");
}
if (conditions.length > 0) {
queryBuilder.where(`(${conditions.join(" OR ")})`);
}
}
if (options.period) {
queryBuilder.where("created_at >= fromUnixTimestamp64Milli({period: Int64})", {
period: new Date(Date.now() - options.period).getTime(),
});
}
if (options.from) {
queryBuilder.where("created_at >= fromUnixTimestamp64Milli({from: Int64})", {
from: options.from,
});
}
if (options.to) {
queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", {
to: options.to,
});
}
}
@@ -0,0 +1,198 @@
import { type ClickHouse } from "@internal/clickhouse";
import { type Tracer } from "@internal/tracing";
import { type Logger, type LogLevel } from "@trigger.dev/core/logger";
import { type Prisma } from "@trigger.dev/database";
import parseDuration from "parse-duration";
import { z } from "zod";
import { type PrismaClientOrTransaction } from "~/db.server";
import { startActiveSpan } from "~/v3/tracer.server";
import { ClickHouseSessionsRepository } from "./clickhouseSessionsRepository.server";
export type SessionsRepositoryOptions = {
clickhouse: ClickHouse;
prisma: PrismaClientOrTransaction;
logger?: Logger;
logLevel?: LogLevel;
tracer?: Tracer;
};
/**
* Derived status values — `Session` rows don't have a stored status column.
* `ACTIVE` is the base state; `CLOSED` means `closedAt` is set; `EXPIRED`
* means `expiresAt` has passed.
*/
export const SessionStatus = z.enum(["ACTIVE", "CLOSED", "EXPIRED"]);
export type SessionStatus = z.infer<typeof SessionStatus>;
const SessionListInputOptionsSchema = z.object({
organizationId: z.string(),
projectId: z.string(),
environmentId: z.string(),
// filters
types: z.array(z.string()).optional(),
tags: z.array(z.string()).optional(),
taskIdentifiers: z.array(z.string()).optional(),
externalId: z.string().optional(),
statuses: z.array(SessionStatus).optional(),
period: z.string().optional(),
from: z.number().optional(),
to: z.number().optional(),
});
export type SessionListInputOptions = z.infer<typeof SessionListInputOptionsSchema>;
export type SessionListInputFilters = Omit<
SessionListInputOptions,
"organizationId" | "projectId" | "environmentId"
>;
export type FilterSessionsOptions = Omit<SessionListInputOptions, "period"> & {
/** period converted to milliseconds duration */
period: number | undefined;
};
type Pagination = {
page: {
size: number;
cursor?: string;
direction?: "forward" | "backward";
};
};
export type ListSessionsOptions = SessionListInputOptions & Pagination;
type OffsetPagination = {
offset: number;
limit: number;
};
export type SessionTagListOptions = {
organizationId: string;
projectId: string;
environmentId: string;
period?: string;
from?: number;
to?: number;
/** Case-insensitive substring match on the tag name */
query?: string;
} & OffsetPagination;
export type SessionTagList = {
tags: string[];
};
export type ListedSession = Prisma.SessionGetPayload<{
select: {
id: true;
friendlyId: true;
externalId: true;
type: true;
taskIdentifier: true;
tags: true;
metadata: true;
closedAt: true;
closedReason: true;
expiresAt: true;
createdAt: true;
updatedAt: true;
runtimeEnvironmentId: true;
};
}>;
export type ISessionsRepository = {
name: string;
listSessionIds(options: ListSessionsOptions): Promise<string[]>;
listSessions(options: ListSessionsOptions): Promise<{
sessions: ListedSession[];
pagination: {
nextCursor: string | null;
previousCursor: string | null;
};
}>;
countSessions(options: SessionListInputOptions): Promise<number>;
listTags(options: SessionTagListOptions): Promise<SessionTagList>;
};
export class SessionsRepository implements ISessionsRepository {
private readonly clickHouseSessionsRepository: ClickHouseSessionsRepository;
constructor(private readonly options: SessionsRepositoryOptions) {
this.clickHouseSessionsRepository = new ClickHouseSessionsRepository(options);
}
get name() {
return "sessionsRepository";
}
async listSessionIds(options: ListSessionsOptions): Promise<string[]> {
return startActiveSpan(
"sessionsRepository.listSessionIds",
async () => this.clickHouseSessionsRepository.listSessionIds(options),
{
attributes: {
"repository.name": "clickhouse",
organizationId: options.organizationId,
projectId: options.projectId,
environmentId: options.environmentId,
},
}
);
}
async listSessions(options: ListSessionsOptions) {
return startActiveSpan(
"sessionsRepository.listSessions",
async () => this.clickHouseSessionsRepository.listSessions(options),
{
attributes: {
"repository.name": "clickhouse",
organizationId: options.organizationId,
projectId: options.projectId,
environmentId: options.environmentId,
},
}
);
}
async countSessions(options: SessionListInputOptions) {
return startActiveSpan(
"sessionsRepository.countSessions",
async () => this.clickHouseSessionsRepository.countSessions(options),
{
attributes: {
"repository.name": "clickhouse",
organizationId: options.organizationId,
projectId: options.projectId,
environmentId: options.environmentId,
},
}
);
}
async listTags(options: SessionTagListOptions) {
return startActiveSpan(
"sessionsRepository.listTags",
async () => this.clickHouseSessionsRepository.listTags(options),
{
attributes: {
"repository.name": "clickhouse",
organizationId: options.organizationId,
projectId: options.projectId,
environmentId: options.environmentId,
},
}
);
}
}
export function parseSessionListInputOptions(data: unknown): SessionListInputOptions {
return SessionListInputOptionsSchema.parse(data);
}
export function convertSessionListInputOptionsToFilterOptions(
options: SessionListInputOptions
): FilterSessionsOptions {
return {
...options,
period: options.period ? parseDuration(options.period) ?? undefined : undefined,
};
}
@@ -4,6 +4,12 @@ import { z } from "zod";
import { env } from "~/env.server";
import { logger } from "~/services/logger.server";
import { runsReplicationInstance } from "~/services/runsReplicationInstance.server";
// Reference-hold the sessions-replication singleton so module evaluation runs
// its initializer (creates the ClickHouse client, subscribes to the logical
// replication slot, wires signal handlers) when the webapp boots. A bare
// side-effect import gets tree-shaken by the bundler.
import { sessionsReplicationInstance } from "~/services/sessionsReplicationInstance.server";
void sessionsReplicationInstance;
import { singleton } from "~/utils/singleton";
import { tracer } from "../tracer.server";
import { $replica } from "~/db.server";
@@ -0,0 +1,212 @@
import { ClickHouse } from "@internal/clickhouse";
import { containerTest } from "@internal/testcontainers";
import { setTimeout } from "node:timers/promises";
import { z } from "zod";
import { SessionsReplicationService } from "~/services/sessionsReplicationService.server";
vi.setConfig({ testTimeout: 60_000 });
describe("SessionsReplicationService", () => {
containerTest(
"replicates an insert from Postgres Session → ClickHouse sessions_v1",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
// Logical replication needs full-row images for DELETE events.
await prisma.$executeRawUnsafe(`ALTER TABLE public."Session" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "sessions-replication",
compression: { request: true },
logLevel: "warn",
});
const service = new SessionsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "sessions-replication",
slotName: "sessions_to_clickhouse_v1",
publicationName: "sessions_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logLevel: "warn",
});
await service.start();
const organization = await prisma.organization.create({
data: { title: "test", slug: "test" },
});
const project = await prisma.project.create({
data: {
name: "test",
slug: "test",
organizationId: organization.id,
externalRef: "test",
},
});
const environment = await prisma.runtimeEnvironment.create({
data: {
slug: "test",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test",
pkApiKey: "test",
shortcode: "test",
},
});
const session = await prisma.session.create({
data: {
id: "session_test_insert_1",
friendlyId: "session_abc123",
externalId: "my-test-session",
type: "chat.agent",
projectId: project.id,
runtimeEnvironmentId: environment.id,
environmentType: "DEVELOPMENT",
organizationId: organization.id,
taskIdentifier: "my-agent",
triggerConfig: {
basePayload: { messages: [], trigger: "preload" },
},
tags: ["user:42", "plan:pro"],
metadata: { plan: "pro", seats: 3 },
},
});
// Allow the replication pipeline to flush
await setTimeout(2000);
const querySessions = clickhouse.reader.query({
name: "read-sessions",
query: "SELECT * FROM trigger_dev.sessions_v1 FINAL",
schema: z.any(),
});
const [queryError, result] = await querySessions({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1);
expect(result?.[0]).toEqual(
expect.objectContaining({
session_id: session.id,
friendly_id: session.friendlyId,
external_id: "my-test-session",
type: "chat.agent",
project_id: project.id,
environment_id: environment.id,
organization_id: organization.id,
environment_type: "DEVELOPMENT",
task_identifier: "my-agent",
tags: ["user:42", "plan:pro"],
_is_deleted: 0,
})
);
await service.stop();
}
);
containerTest(
"replicates an update (close) from Postgres → ClickHouse",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."Session" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "sessions-replication",
compression: { request: true },
logLevel: "warn",
});
const service = new SessionsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "sessions-replication",
slotName: "sessions_to_clickhouse_v1",
publicationName: "sessions_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
logLevel: "warn",
});
await service.start();
const organization = await prisma.organization.create({
data: { title: "test", slug: "test" },
});
const project = await prisma.project.create({
data: {
name: "test",
slug: "test",
organizationId: organization.id,
externalRef: "test",
},
});
const environment = await prisma.runtimeEnvironment.create({
data: {
slug: "test",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test",
pkApiKey: "test",
shortcode: "test",
},
});
const created = await prisma.session.create({
data: {
id: "session_test_update_1",
friendlyId: "session_update1",
type: "chat.agent",
projectId: project.id,
runtimeEnvironmentId: environment.id,
environmentType: "DEVELOPMENT",
organizationId: organization.id,
taskIdentifier: "my-agent",
triggerConfig: {
basePayload: { messages: [], trigger: "preload" },
},
},
});
await setTimeout(1000);
await prisma.session.update({
where: { id: created.id },
data: { closedAt: new Date(), closedReason: "test-close" },
});
await setTimeout(2000);
const querySessions = clickhouse.reader.query({
name: "read-sessions-closed",
query: "SELECT closed_reason, closed_at FROM trigger_dev.sessions_v1 FINAL",
schema: z.any(),
});
const [queryError, result] = await querySessions({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1);
expect(result?.[0].closed_reason).toBe("test-close");
expect(result?.[0].closed_at).toBeDefined();
await service.stop();
}
);
});
@@ -0,0 +1,42 @@
-- +goose Up
CREATE TABLE trigger_dev.sessions_v1
(
/* ─── identity ─────────────────────────────────────────────── */
environment_id String,
organization_id String,
project_id String,
session_id String,
environment_type LowCardinality(String),
friendly_id String,
external_id String DEFAULT '',
/* ─── type discriminator ──────────────────────────────────── */
type LowCardinality(String),
task_identifier String DEFAULT '',
/* ─── filtering / free-form ──────────────────────────────── */
tags Array(String) CODEC(ZSTD(1)),
metadata JSON(max_dynamic_paths = 256),
/* ─── terminal markers ────────────────────────────────────── */
closed_at Nullable(DateTime64(3)),
closed_reason String DEFAULT '',
expires_at Nullable(DateTime64(3)),
/* ─── timing ─────────────────────────────────────────────── */
created_at DateTime64(3),
updated_at DateTime64(3),
/* ─── commit lsn ────────────────────────────────────────── */
_version UInt64,
_is_deleted UInt8 DEFAULT 0
)
ENGINE = ReplacingMergeTree(_version, _is_deleted)
PARTITION BY toYYYYMM(created_at)
ORDER BY (organization_id, project_id, environment_id, created_at, session_id)
SETTINGS enable_json_type = 1;
-- +goose Down
DROP TABLE IF EXISTS trigger_dev.sessions_v1;
+18
View File
@@ -28,6 +28,12 @@ import {
} from "./taskEvents.js";
import { insertMetrics } from "./metrics.js";
import { insertLlmMetrics } from "./llmMetrics.js";
import {
getSessionTagsQueryBuilder,
getSessionsCountQueryBuilder,
getSessionsQueryBuilder,
insertSessionsCompactArrays,
} from "./sessions.js";
import {
getGlobalModelMetrics,
getGlobalModelComparison,
@@ -57,6 +63,7 @@ export type * from "./metrics.js";
export type * from "./llmMetrics.js";
export type * from "./llmModelAggregates.js";
export type * from "./errors.js";
export type * from "./sessions.js";
export type * from "./client/queryBuilder.js";
// Re-export column constants, indices, and type-safe accessors
@@ -69,6 +76,8 @@ export {
getPayloadField,
} from "./taskRuns.js";
export { SESSION_COLUMNS, SESSION_INDEX, getSessionField } from "./sessions.js";
// TSQL query execution
export {
executeTSQL,
@@ -251,6 +260,15 @@ export class ClickHouse {
};
}
get sessions() {
return {
insertCompactArrays: insertSessionsCompactArrays(this.writer),
queryBuilder: getSessionsQueryBuilder(this.reader),
countQueryBuilder: getSessionsCountQueryBuilder(this.reader),
tagQueryBuilder: getSessionTagsQueryBuilder(this.reader),
};
}
get taskEventsV2() {
return {
insert: insertTaskEventsV2(this.writer),
@@ -0,0 +1,184 @@
import { ClickHouseSettings } from "@clickhouse/client";
import { z } from "zod";
import { ClickhouseReader, ClickhouseWriter } from "./client/types.js";
export const SessionV1 = z.object({
environment_id: z.string(),
organization_id: z.string(),
project_id: z.string(),
session_id: z.string(),
environment_type: z.string(),
friendly_id: z.string(),
external_id: z.string().default(""),
type: z.string(),
task_identifier: z.string().default(""),
tags: z.array(z.string()).default([]),
metadata: z.unknown(),
closed_at: z.number().int().nullish(),
closed_reason: z.string().default(""),
expires_at: z.number().int().nullish(),
created_at: z.number().int(),
updated_at: z.number().int(),
_version: z.string(),
_is_deleted: z.number().int().default(0),
});
export type SessionV1 = z.input<typeof SessionV1>;
// Column order for compact format - must match ClickHouse table schema
export const SESSION_COLUMNS = [
"environment_id",
"organization_id",
"project_id",
"session_id",
"environment_type",
"friendly_id",
"external_id",
"type",
"task_identifier",
"tags",
"metadata",
"closed_at",
"closed_reason",
"expires_at",
"created_at",
"updated_at",
"_version",
"_is_deleted",
] as const;
export type SessionColumnName = (typeof SESSION_COLUMNS)[number];
export const SESSION_INDEX = Object.fromEntries(SESSION_COLUMNS.map((col, idx) => [col, idx])) as {
readonly [K in SessionColumnName]: number;
};
export type SessionFieldTypes = {
environment_id: string;
organization_id: string;
project_id: string;
session_id: string;
environment_type: string;
friendly_id: string;
external_id: string;
type: string;
task_identifier: string;
tags: string[];
metadata: { data: unknown };
closed_at: number | null;
closed_reason: string;
expires_at: number | null;
created_at: number;
updated_at: number;
_version: string;
_is_deleted: number;
};
/**
* Type-safe tuple representing a Session insert array.
* Order matches {@link SESSION_COLUMNS} exactly.
*/
export type SessionInsertArray = [
environment_id: string,
organization_id: string,
project_id: string,
session_id: string,
environment_type: string,
friendly_id: string,
external_id: string,
type: string,
task_identifier: string,
tags: string[],
metadata: { data: unknown },
closed_at: number | null,
closed_reason: string,
expires_at: number | null,
created_at: number,
updated_at: number,
_version: string,
_is_deleted: number,
];
export function getSessionField<K extends SessionColumnName>(
session: SessionInsertArray,
field: K
): SessionFieldTypes[K] {
return session[SESSION_INDEX[field]] as SessionFieldTypes[K];
}
export function insertSessionsCompactArrays(ch: ClickhouseWriter, settings?: ClickHouseSettings) {
return ch.insertCompactRaw({
name: "insertSessionsCompactArrays",
table: "trigger_dev.sessions_v1",
columns: SESSION_COLUMNS,
settings: {
enable_json_type: 1,
type_json_skip_duplicated_paths: 1,
...settings,
},
});
}
export function insertSessions(ch: ClickhouseWriter, settings?: ClickHouseSettings) {
return ch.insert({
name: "insertSessions",
table: "trigger_dev.sessions_v1",
schema: SessionV1,
settings: {
enable_json_type: 1,
type_json_skip_duplicated_paths: 1,
...settings,
},
});
}
// ─── read path ───────────────────────────────────────────────────
export const SessionV1QueryResult = z.object({
session_id: z.string(),
});
export type SessionV1QueryResult = z.infer<typeof SessionV1QueryResult>;
/**
* Base query builder for listing Sessions. Filters + pagination are composed
* on top of this; callers can chain `.where(...).orderBy(...).limit(...)`.
*/
export function getSessionsQueryBuilder(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.queryBuilder({
name: "getSessions",
baseQuery: "SELECT session_id FROM trigger_dev.sessions_v1 FINAL",
schema: SessionV1QueryResult,
settings,
});
}
export function getSessionsCountQueryBuilder(
ch: ClickhouseReader,
settings?: ClickHouseSettings
) {
return ch.queryBuilder({
name: "getSessionsCount",
baseQuery: "SELECT count() as count FROM trigger_dev.sessions_v1 FINAL",
schema: z.object({ count: z.number().int() }),
settings,
});
}
export const SessionTagsQueryResult = z.object({
tag: z.string(),
});
export type SessionTagsQueryResult = z.infer<typeof SessionTagsQueryResult>;
export function getSessionTagsQueryBuilder(
ch: ClickhouseReader,
settings?: ClickHouseSettings
) {
return ch.queryBuilder({
name: "getSessionTags",
baseQuery: "SELECT DISTINCT arrayJoin(tags) as tag FROM trigger_dev.sessions_v1",
schema: SessionTagsQueryResult,
settings,
});
}
@@ -0,0 +1,33 @@
-- CreateTable
CREATE TABLE "Session" (
"id" TEXT NOT NULL,
"friendlyId" TEXT NOT NULL,
"externalId" TEXT,
"type" TEXT NOT NULL,
"projectId" TEXT NOT NULL,
"runtimeEnvironmentId" TEXT NOT NULL,
"environmentType" "RuntimeEnvironmentType" NOT NULL,
"organizationId" TEXT NOT NULL,
"taskIdentifier" TEXT,
"tags" TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[],
"metadata" JSONB,
"closedAt" TIMESTAMP(3),
"closedReason" TEXT,
"expiresAt" TIMESTAMP(3),
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" TIMESTAMP(3) NOT NULL,
CONSTRAINT "Session_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "Session_friendlyId_key"
ON "Session"("friendlyId");
-- CreateIndex
CREATE UNIQUE INDEX "Session_runtimeEnvironmentId_externalId_key"
ON "Session"("runtimeEnvironmentId", "externalId");
-- CreateIndex
CREATE INDEX "Session_expiresAt_idx"
ON "Session"("expiresAt");
@@ -0,0 +1,31 @@
-- AlterTable
ALTER TABLE "Session"
ADD COLUMN "currentRunId" TEXT,
ADD COLUMN "currentRunVersion" INTEGER NOT NULL DEFAULT 0,
ADD COLUMN "triggerConfig" JSONB NOT NULL,
ALTER COLUMN "taskIdentifier" SET NOT NULL;
-- CreateTable
CREATE TABLE "SessionRun" (
"id" TEXT NOT NULL,
"sessionId" TEXT NOT NULL,
"runId" TEXT NOT NULL,
"reason" TEXT NOT NULL,
"triggeredAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "SessionRun_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "SessionRun_runId_key"
ON "SessionRun"("runId");
-- CreateIndex
CREATE INDEX "SessionRun_sessionId_idx"
ON "SessionRun"("sessionId");
-- AddForeignKey
ALTER TABLE "SessionRun"
ADD CONSTRAINT "SessionRun_sessionId_fkey"
FOREIGN KEY ("sessionId") REFERENCES "Session"("id")
ON DELETE CASCADE ON UPDATE CASCADE;
@@ -0,0 +1,3 @@
-- CreateIndex
CREATE INDEX CONCURRENTLY IF NOT EXISTS "Session_currentRunId_idx"
ON "Session"("currentRunId");
@@ -686,6 +686,92 @@ enum TaskTriggerSource {
SCHEDULED
}
/// Durable, typed, bidirectional I/O primitive. Owns two S2 streams (.out / .in).
/// The row is essentially static — no status, no counters, no pointers. No
/// foreign keys: project/runtimeEnvironment/organization ids are plain
/// scalar columns (matches TaskRun pattern). List-style queries are served
/// from ClickHouse, not Postgres, so only point-lookup indexes live here.
model Session {
id String @id @default(cuid())
friendlyId String @unique
/// User-supplied identifier scoped to the environment. Used for
/// idempotent upsert and for resolving sessions via the public API.
externalId String?
/// Plain string — intentionally not an enum.
type String
/// Denormalized scoping columns — no FK relations.
projectId String
runtimeEnvironmentId String
environmentType RuntimeEnvironmentType
organizationId String
/// Task this session triggers runs against. Required — Sessions are
/// task-bound: creating a session also triggers its first run, and
/// every subsequent re-trigger uses this same identifier.
taskIdentifier String
/// Trigger config used for every run this session schedules. Shape
/// (validated at the route layer, opaque to the DB):
/// { basePayload: object, machine?: string, queue?: string,
/// tags?: string[], maxAttempts?: number,
/// idleTimeoutInSeconds?: number }
/// `basePayload` carries the customer's client-data; runtime fields
/// (chatId, messages, trigger) are merged at trigger time.
triggerConfig Json
tags String[] @default([])
metadata Json?
/// Live run pointer — non-FK so run deletion never cascades. Can lag
/// reality; the `.in/append` handler re-checks the snapshot status
/// before reusing it.
currentRunId String?
/// Monotonic counter used for optimistic locking on `currentRunId`
/// swaps. Bumped atomically alongside any update that changes
/// `currentRunId`.
currentRunVersion Int @default(0)
/// Terminal markers — written once, never flipped back.
closedAt DateTime?
closedReason String?
expiresAt DateTime?
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
runs SessionRun[]
/// Idempotency: `(env, externalId)` uniquely identifies a session.
/// PostgreSQL treats NULLs as distinct, so `externalId=NULL` rows never collide.
@@unique([runtimeEnvironmentId, externalId])
@@index([expiresAt])
@@index([currentRunId])
}
/// Historical record of every run a Session has owned. Append-only —
/// rows are inserted on each `ensureRunForSession` claim, never updated.
/// Lets us reconstruct the run timeline of a chat for debugging /
/// dashboard surfaces. The relation cascades on Session delete (tied to
/// the session lifecycle) but `runId` is a plain string column with no
/// FK to TaskRun so run pruning is independent.
model SessionRun {
id String @id @default(cuid())
sessionId String
/// TaskRun.id (no FK — runs may be archived independently of session history)
runId String @unique
/// One of: "initial" | "continuation" | "upgrade" | "manual".
/// Plain string for forward-compat with future trigger reasons.
reason String
triggeredAt DateTime @default(now())
session Session @relation(fields: [sessionId], references: [id], onDelete: Cascade)
@@index([sessionId])
}
model TaskRun {
id String @id @default(cuid())
@@ -97,6 +97,7 @@ export const BatchId = new IdUtil("batch");
export const BulkActionId = new IdUtil("bulk");
export const AttemptId = new IdUtil("attempt");
export const ErrorId = new IdUtil("error");
export const SessionId = new IdUtil("session");
export class IdGenerator {
private alphabet: string;
+245
View File
@@ -1411,6 +1411,38 @@ export type CreateInputStreamWaitpointResponseBody = z.infer<
typeof CreateInputStreamWaitpointResponseBody
>;
/**
* Create a run-scoped waitpoint that completes when the next record lands on
* a Session channel (`.in` or `.out`). Mirrors `CreateInputStreamWaitpointRequestBody`
* but keyed by `{sessionId, io}` instead of `{runId, streamId}`. The run is
* still the thing being suspended — Session only supplies the trigger source.
*/
export const CreateSessionStreamWaitpointRequestBody = z.object({
/** Session friendlyId (`session_*`) or user-supplied externalId. */
session: z.string(),
io: z.enum(["out", "in"]),
timeout: z.string().optional(),
idempotencyKey: z.string().optional(),
idempotencyKeyTTL: z.string().optional(),
tags: z.union([z.string(), z.array(z.string())]).optional(),
/**
* Last S2 sequence number the client has seen on this session channel.
* Used to catch data that arrived before `.wait()` was called.
*/
lastSeqNum: z.number().optional(),
});
export type CreateSessionStreamWaitpointRequestBody = z.infer<
typeof CreateSessionStreamWaitpointRequestBody
>;
export const CreateSessionStreamWaitpointResponseBody = z.object({
waitpointId: z.string(),
isCached: z.boolean(),
});
export type CreateSessionStreamWaitpointResponseBody = z.infer<
typeof CreateSessionStreamWaitpointResponseBody
>;
export const waitpointTokenStatuses = ["WAITING", "COMPLETED", "TIMED_OUT"] as const;
export const WaitpointTokenStatus = z.enum(waitpointTokenStatuses);
export type WaitpointTokenStatus = z.infer<typeof WaitpointTokenStatus>;
@@ -1449,6 +1481,219 @@ export const CompleteWaitpointTokenRequestBody = z.object({
});
export type CompleteWaitpointTokenRequestBody = z.infer<typeof CompleteWaitpointTokenRequestBody>;
/**
* Trigger config persisted on a Session. Drives every run the session
* schedules — `basePayload` is the customer's wire payload (for
* chat.agent: `{ chatId, ...clientData }`), runtime fields like
* `trigger: "preload" | "trigger"` are merged on top per-call by the
* server's trigger machinery.
*/
export const SessionTriggerConfig = z.object({
basePayload: z.record(z.unknown()),
machine: MachinePresetName.optional(),
queue: z.string().max(128).optional(),
tags: z.array(z.string().max(128)).max(5).optional(),
maxAttempts: z.number().int().positive().max(10).optional(),
/** Convenience field surfaced to chat.agent via the wire payload. */
idleTimeoutInSeconds: z.number().int().positive().max(3600).optional(),
});
export type SessionTriggerConfig = z.infer<typeof SessionTriggerConfig>;
/**
* Request body for `POST /api/v1/sessions`. Creates a Session and
* triggers its first run. Sessions are task-bound: `taskIdentifier` and
* `triggerConfig` are required, and re-runs scheduled by the server
* (after run termination, after `end-and-continue`) reuse the same
* config.
*/
export const CreateSessionRequestBody = z.object({
/** Plain string discriminator — e.g. `"chat.agent"`. Not validated against an enum on the server. */
type: z.string().min(1).max(64),
/** User-supplied idempotency key. Unique per environment. Empty strings are rejected. */
externalId: z
.string()
.trim()
.min(1)
.max(256)
.refine((v) => !v.startsWith("session_"), {
message: "externalId cannot start with 'session_' (reserved prefix for internal friendlyIds)",
})
.optional(),
/** Task this session triggers runs against. Required. */
taskIdentifier: z.string().min(1).max(128),
/** Trigger config used for every run scheduled by this session. */
triggerConfig: SessionTriggerConfig,
/** Up to 10 tags for dashboard filtering. */
tags: z.array(z.string().max(128)).max(10).optional(),
/** Arbitrary JSON metadata. */
metadata: z.record(z.unknown()).optional(),
/** Absolute expiry timestamp for retention. */
expiresAt: z.coerce.date().optional(),
});
export type CreateSessionRequestBody = z.infer<typeof CreateSessionRequestBody>;
export const SessionItem = z.object({
id: z.string(),
externalId: z.string().nullable(),
type: z.string(),
taskIdentifier: z.string(),
/**
* Optional on the wire because some surfaces (the list endpoint backed
* by ClickHouse, list-page rendering) don't carry triggerConfig.
* Always populated on `POST /sessions` and `GET /sessions/:id`.
*/
triggerConfig: SessionTriggerConfig.optional(),
/**
* Friendly id of the live run for this session, if any. Optional on
* the wire — list surfaces may not include it. Routes that emit
* `SessionItem` are responsible for resolving the friendly form
* from the underlying cuid before returning.
*/
currentRunId: z.string().nullable().optional(),
tags: z.array(z.string()),
metadata: z.record(z.unknown()).nullable(),
closedAt: z.coerce.date().nullable(),
closedReason: z.string().nullable(),
expiresAt: z.coerce.date().nullable(),
createdAt: z.coerce.date(),
updatedAt: z.coerce.date(),
});
export type SessionItem = z.infer<typeof SessionItem>;
export const CreatedSessionResponseBody = SessionItem.extend({
/** Friendly id of the first run triggered alongside session create. */
runId: z.string(),
/** Session-scoped public access token: `read:sessions:{ext} + write:sessions:{ext}`. */
publicAccessToken: z.string(),
/** True if the session existed already (idempotent upsert), false if newly created. */
isCached: z.boolean(),
});
export type CreatedSessionResponseBody = z.infer<typeof CreatedSessionResponseBody>;
export const RetrieveSessionResponseBody = SessionItem;
export type RetrieveSessionResponseBody = z.infer<typeof RetrieveSessionResponseBody>;
/**
* Body for `POST /api/v1/sessions/:session/end-and-continue`. Used by the
* running agent to request a clean handoff to a fresh run on the latest
* deployed version (typical use case: `chat.requestUpgrade`). The
* server triggers a new run, atomically swaps `currentRunId`, and the
* caller exits.
*/
export const EndAndContinueSessionRequestBody = z.object({
/** The friendlyId of the run requesting the handoff. */
callingRunId: z.string(),
/** Free-form label for the SessionRun audit row. e.g. `"upgrade"`. */
reason: z.string().max(64),
});
export type EndAndContinueSessionRequestBody = z.infer<typeof EndAndContinueSessionRequestBody>;
export const EndAndContinueSessionResponseBody = z.object({
/** friendlyId of the run that has taken over the session. */
runId: z.string(),
/**
* False when the swap was preempted (a different run was already
* running by the time we tried to claim). The caller should treat
* this as "someone else moved on" — exit cleanly without expecting
* to drive the next run.
*/
swapped: z.boolean(),
});
export type EndAndContinueSessionResponseBody = z.infer<
typeof EndAndContinueSessionResponseBody
>;
export const UpdateSessionRequestBody = z.object({
tags: z.array(z.string().max(128)).max(10).optional(),
metadata: z.record(z.unknown()).nullable().optional(),
// Null explicitly clears the externalId; non-null values must be non-empty.
externalId: z
.union([
z.literal(null),
z
.string()
.trim()
.min(1)
.max(256)
.refine((v) => !v.startsWith("session_"), {
message:
"externalId cannot start with 'session_' (reserved prefix for internal friendlyIds)",
}),
])
.optional(),
});
export type UpdateSessionRequestBody = z.infer<typeof UpdateSessionRequestBody>;
export const CloseSessionRequestBody = z.object({
reason: z.string().max(256).optional(),
});
export type CloseSessionRequestBody = z.infer<typeof CloseSessionRequestBody>;
export const SessionStatus = z.enum(["ACTIVE", "CLOSED", "EXPIRED"]);
export type SessionStatus = z.infer<typeof SessionStatus>;
/**
* Server-side validation schema for `GET /api/v1/sessions`. Follows the same
* cursor-pagination convention as runs/waitpoints (`page[size]`,
* `page[after]`, `page[before]`) and uses the `filter[*]` prefix for
* narrowing fields — both produced automatically by `zodfetchCursorPage`
* and the matching client-side search-query helper.
*/
export const ListSessionsQueryParams = z
.object({
"page[size]": z.coerce.number().int().min(1).max(100).default(20),
"page[after]": z.string().optional(),
"page[before]": z.string().optional(),
"filter[type]": z.union([z.string(), z.array(z.string())]).optional(),
"filter[tags]": z.union([z.string(), z.array(z.string())]).optional(),
"filter[taskIdentifier]": z.union([z.string(), z.array(z.string())]).optional(),
"filter[externalId]": z.string().optional(),
"filter[status]": z.union([SessionStatus, z.array(SessionStatus)]).optional(),
"filter[createdAt][period]": z.string().optional(),
"filter[createdAt][from]": z.coerce.number().int().optional(),
"filter[createdAt][to]": z.coerce.number().int().optional(),
})
.refine(
(value) => !(value["page[after]"] && value["page[before]"]),
{
message: "Cannot pass both page[after] and page[before] on the same request",
path: ["page[before]"],
}
);
export type ListSessionsQueryParams = z.infer<typeof ListSessionsQueryParams>;
/**
* Client-facing list options — flattened shape that
* {@link ApiClient.listSessions} converts into the `filter[*]` / `page[*]`
* query string before sending.
*/
export const ListSessionsOptions = z.object({
limit: z.number().int().min(1).max(100).optional(),
after: z.string().optional(),
before: z.string().optional(),
type: z.union([z.string(), z.array(z.string())]).optional(),
tag: z.union([z.string(), z.array(z.string())]).optional(),
taskIdentifier: z.union([z.string(), z.array(z.string())]).optional(),
externalId: z.string().optional(),
status: z.union([SessionStatus, z.array(SessionStatus)]).optional(),
period: z.string().optional(),
from: z.union([z.number(), z.date()]).optional(),
to: z.union([z.number(), z.date()]).optional(),
});
export type ListSessionsOptions = z.infer<typeof ListSessionsOptions>;
export const ListedSessionItem = SessionItem;
export type ListedSessionItem = z.infer<typeof ListedSessionItem>;
export const ListSessionsResponseBody = z.object({
data: z.array(ListedSessionItem),
pagination: z.object({
next: z.string().optional(),
previous: z.string().optional(),
}),
});
export type ListSessionsResponseBody = z.infer<typeof ListSessionsResponseBody>;
export const CompleteWaitpointTokenResponseBody = z.object({
success: z.literal(true),
});