feat(webapp,core): Session channel waitpoints — server side

ericallam · ericallam · commit 4cadc19bc010 · 2026-04-23T10:00:43.000+01:00
Give Session channels run-engine waitpoint semantics so a task can
suspend while idle on a session channel and resume when an external
client sends a record — parallel to what streams.input offers
run-scoped streams.

Webapp
- POST /api/v1/runs/:runFriendlyId/session-streams/wait — creates a
  manual waitpoint attached to {sessionId, io} and race-checks the S2
  stream starting at lastSeqNum so pre-arrived data fires it
  immediately. Mirrors the existing input-stream waitpoint route.
- sessionStreamWaitpointCache.server.ts — Redis set keyed on
  {sessionFriendlyId, io}, drained atomically on each append so
  concurrent multi-tab waiters all wake together.
- realtime.v1.sessions.$session.$io.append now drains pending
  waitpoints after every record lands and completes each with the
  appended body.
- S2RealtimeStreams.readSessionStreamRecords — session-channel
  parallel of readRecords, feeds the race-check path.

Core
- CreateSessionStreamWaitpoint request/response schemas alongside
  the existing Session CRUD schemas. Server API contract only —
  the client ApiClient + SDK wrapper ship on the AI-chat branch.
diff --git a/.changeset/session-primitive.md b/.changeset/session-primitive.md
@@ -2,4 +2,4 @@
 "@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.
+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.
diff --git a/.server-changes/session-primitive.md b/.server-changes/session-primitive.md
@@ -4,3 +4,5 @@ type: feature
 ---
 
 Add `Session` primitive — a durable, typed, bidirectional I/O primitive that outlives a single run, intended for agent/chat use cases. Ships the Postgres schema (`Session` table), control-plane CRUD routes (`POST/GET/PATCH /api/v1/sessions`, `POST /api/v1/sessions/:session/close` — polymorphic on friendlyId or externalId), `sessions` JWT scope, ClickHouse `sessions_v1` table, and `SessionsReplicationService` (logical replication from Postgres `Session` → ClickHouse `sessions_v1`). Run-scoped realtime streams (`streams.pipe`/`streams.input`) are unchanged and do **not** create Session rows.
+
+Adds `POST /api/v1/runs/:runFriendlyId/session-streams/wait` (session-stream waitpoint creation) and wires `POST /realtime/v1/sessions/:session/:io/append` to fire any pending waitpoints on the channel. Gives `session.in` run-engine waitpoint semantics matching run-scoped input streams: a task can suspend while idle on a session channel and resume when an external client sends a record. Redis-backed pending-waitpoint set (`ssw:{sessionFriendlyId}:{io}`) is drained atomically on each append so multiple concurrent waiters (e.g. multi-tab chat) all resume together.
diff --git a/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts b/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts
@@ -0,0 +1,165 @@
+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 { 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 { 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 });
+      }
+
+      const session = await resolveSessionByIdOrExternalId(
+        $replica,
+        authentication.environment.id,
+        body.session
+      );
+
+      if (!session) {
+        return json({ error: "Session not found" }, { status: 404 });
+      }
+
+      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 (sessionFriendlyId, io) — both runs on a
+      // multi-tab session wake on the same record.
+      const ttlMs = timeout ? timeout.getTime() - Date.now() : undefined;
+      await addSessionStreamWaitpoint(
+        session.friendlyId,
+        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 {
+          const realtimeStream = getRealtimeStreamInstance(
+            authentication.environment,
+            run.realtimeStreamsVersion
+          );
+
+          if (realtimeStream instanceof S2RealtimeStreams) {
+            const records = await realtimeStream.readSessionStreamRecords(
+              session.friendlyId,
+              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(
+                session.friendlyId,
+                body.io,
+                result.waitpoint.id
+              );
+            }
+          }
+        } catch {
+          // Non-fatal: pending registration stays in Redis; the next append
+          // will complete the waitpoint via the append handler path.
+        }
+      }
+
+      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 });
+      } else if (error instanceof Error) {
+        return json({ error: error.message }, { status: 500 });
+      }
+
+      return json({ error: "Something went wrong" }, { status: 500 });
+    }
+  }
+);
+
+export { action, loader };
diff --git a/apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts b/apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts
@@ -3,10 +3,13 @@ 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 { 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({
@@ -81,6 +84,43 @@ const { action } = createActionApiRoute(
       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).
+    const [drainError, waitpointIds] = await tryCatch(
+      drainSessionStreamWaitpoints(session.friendlyId, params.io)
+    );
+    if (drainError) {
+      logger.error("Failed to drain session stream waitpoints", {
+        sessionFriendlyId: session.friendlyId,
+        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", {
+              sessionFriendlyId: session.friendlyId,
+              io: params.io,
+              waitpointId,
+              error: completeError,
+            });
+          }
+        })
+      );
+    }
+
     return json({ ok: true }, { status: 200 });
   }
 );
diff --git a/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts b/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts
@@ -186,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();
diff --git a/apps/webapp/app/services/sessionStreamWaitpointCache.server.ts b/apps/webapp/app/services/sessionStreamWaitpointCache.server.ts
@@ -0,0 +1,117 @@
+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);
+
+/**
+ * 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.sadd(key, waitpointId);
+    await redis.pexpire(key, 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,
+    });
+  }
+}
diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts
