fix(webapp): harden the realtime session routes (#3890)

## Summary

Reliability and authorization fixes for realtime chat sessions:

- Session-stream waitpoint delivery is scoped to the environment, so two
environments using the same session `externalId` can no longer complete
each other's waitpoints.
- The session snapshot-url routes now enforce per-session authorization,
and appending to a session's `out` channel requires secret-key auth, so
a session-scoped token can't read another session's snapshot or forge
assistant output.
- Appends that carry an `X-Part-Id` header are deduplicated on retry, so
a retried send can't duplicate a message.
- Session creation rejects expired sessions (instead of triggering a run
that can never receive input), `externalId` is immutable after creation,
and the sessions list endpoint returns friendly `run_*` ids to match the
single-session routes.

## Rollout

The waitpoint cache key gains an environment prefix. To keep waitpoints
registered by the previous deploy working across the boundary, the drain
reads both the new and the previous key for this release; the legacy
read can be removed a release later once no pre-deploy waitpoints
remain.

Author: ericallam

.server-changes/session-route-hardening.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Hardening fixes for realtime sessions: stricter authorization on snapshot URLs and out-channel appends, environment-scoped message delivery for waiting runs, and idempotent appends via the X-Part-Id header. Session creation now rejects expired sessions, externalId can no longer be changed after creation, and the sessions list returns friendly run ids.

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts

@@ -106,12 +106,14 @@ const { action, loader } = createActionApiRoute(
       });
 
       // 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.
+      // append fires it. Keyed by (environmentId, addressingKey, io) — the
+      // canonical string for the row, scoped to the environment because
+      // externalIds are only unique per environment. The append handler
+      // drains by the same 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(
+        authentication.environment.id,
         addressingKey,
         body.io,
         result.waitpoint.id,
@@ -152,6 +154,7 @@ const { action, loader } = createActionApiRoute(
               });
 
               await removeSessionStreamWaitpoint(
+                authentication.environment.id,
                 addressingKey,
                 body.io,
                 result.waitpoint.id

apps/webapp/app/routes/api.v1.sessions.$session.ts

@@ -74,6 +74,24 @@ const { action } = createActionApiRoute(
       return json({ error: "Session not found" }, { status: 404 });
     }
 
+    // The externalId is the canonical addressing key once set: the S2
+    // stream names, the waitpoint cache key, and the minted session PAT
+    // scope all derive from it. Re-keying a session would orphan its
+    // streams (the chat goes silent) and invalidate the PAT's scope, so
+    // reject any change. Same-value PATCHes stay idempotent.
+    if (
+      body.externalId !== undefined &&
+      body.externalId !== existing.externalId
+    ) {
+      return json(
+        {
+          error:
+            "externalId cannot be changed after creation; close this session and create a new one with the desired externalId",
+        },
+        { status: 422 }
+      );
+    }
+
     try {
       const updated = await prisma.session.update({
         where: { id: existing.id },

apps/webapp/app/routes/api.v1.sessions.$sessionId.snapshot-url.ts

@@ -4,6 +4,7 @@ import { $replica } from "~/db.server";
 import { chatSnapshotStorageKey } from "~/services/realtime/chatSnapshot.server";
 import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
 import {
+  anyResource,
   createActionApiRoute,
   createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
@@ -21,8 +22,31 @@ const routeConfig = {
     resolveSessionByIdOrExternalId($replica, auth.environment.id, params.sessionId),
 };
 
+// Authorize against the union of the URL form, friendlyId, and externalId —
+// same shape as the sibling session routes. Without an authorization block
+// the route builder skips scope checks entirely, so any session-scoped JWT
+// in the environment could presign URLs for any other session's snapshot.
+function sessionResource(
+  paramId: string,
+  session: { friendlyId: string; externalId: string | null } | null | undefined
+) {
+  const ids = new Set<string>([paramId]);
+  if (session) {
+    ids.add(session.friendlyId);
+    if (session.externalId) ids.add(session.externalId);
+  }
+  return anyResource([...ids].map((id) => ({ type: "sessions" as const, id })));
+}
+
 export const { action } = createActionApiRoute(
-  { ...routeConfig, method: "PUT" },
+  {
+    ...routeConfig,
+    method: "PUT",
+    authorization: {
+      action: "write",
+      resource: (params, _, __, ___, session) => sessionResource(params.sessionId, session),
+    },
+  },
   async ({ authentication, resource: session }) => {
     if (!session) {
       return json({ error: "Session not found" }, { status: 404 });
@@ -42,7 +66,15 @@ export const { action } = createActionApiRoute(
   }
 );
 
-export const loader = createLoaderApiRoute(routeConfig, async ({ authentication, resource: session }) => {
+export const loader = createLoaderApiRoute(
+  {
+    ...routeConfig,
+    authorization: {
+      action: "read",
+      resource: (session, params) => sessionResource(params.sessionId, session),
+    },
+  },
+  async ({ authentication, resource: session }) => {
   if (!session) {
     return json({ error: "Session not found" }, { status: 404 });
   }

apps/webapp/app/routes/api.v1.sessions.ts

@@ -18,7 +18,10 @@ import {
   type SessionTriggerConfig,
 } from "~/services/realtime/sessionRunManager.server";
 import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
-import { serializeSession } from "~/services/realtime/sessions.server";
+import {
+  serializeSession,
+  serializeSessionsWithFriendlyRunIds,
+} from "~/services/realtime/sessions.server";
 import { SessionsRepository } from "~/services/sessionsRepository/sessionsRepository.server";
 import {
   anyResource,
@@ -91,17 +94,29 @@ export const loader = createLoaderApiRoute(
       },
     });
 
-    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)
+    // Batched friendlyId translation: `currentRunId` on the wire is the
+    // public `run_*` form, matching the single-session routes. One `IN`
+    // lookup per page.
+    const data = await serializeSessionsWithFriendlyRunIds(
+      rows.map(
+        (row) =>
+          ({
+            ...row,
+            // Columns the list query doesn't select — filled so the
+            // serializer can operate on a narrowed payload without type errors.
+            projectId: authentication.environment.projectId,
+            environmentType: authentication.environment.type,
+            organizationId: authentication.environment.organizationId,
+          }) as Session
       ),
+      {
+        projectId: authentication.environment.projectId,
+        runtimeEnvironmentId: authentication.environment.id,
+      }
+    );
+
+    return json<ListSessionsResponseBody>({
+      data,
       pagination: {
         ...(pagination.nextCursor ? { next: pagination.nextCursor } : {}),
         ...(pagination.previousCursor ? { previous: pagination.previousCursor } : {}),
@@ -225,6 +240,17 @@ const { action } = createActionApiRoute(
         );
       }
 
+      // Same guard as the append / end-and-continue handlers: an expired
+      // row must not spawn a run, because every subsequent `.in/append`
+      // would 400 on the expiry check — a run boots but the chat can
+      // never receive input.
+      if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
+        return json(
+          { error: "Session is expired; use a different externalId to create a new session" },
+          { status: 409 }
+        );
+      }
+
       // Session is task-bound — every session has a live run by
       // construction. `ensureRunForSession` is idempotent: on the
       // cached path it sees `currentRunId` is alive and returns it

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts

@@ -11,7 +11,11 @@ import {
   resolveSessionByIdOrExternalId,
 } from "~/services/realtime/sessions.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
-import { drainSessionStreamWaitpoints } from "~/services/sessionStreamWaitpointCache.server";
+import {
+  claimSessionStreamPart,
+  drainSessionStreamWaitpoints,
+  releaseSessionStreamPart,
+} from "~/services/sessionStreamWaitpointCache.server";
 import {
   anyResource,
   createActionApiRoute,
@@ -91,6 +95,17 @@ const { action, loader } = createActionApiRoute(
       );
     }
 
+    // `.out` is the agent→client channel. Only PRIVATE (secret key) auth —
+    // i.e. the agent run itself — may write to it. Session-scoped JWTs carry
+    // `write:sessions:<key>` for `.in`; without this gate they could forge
+    // assistant chunks and complete `.out` waitpoints on their own session.
+    if (params.io === "out" && authentication.type !== "PRIVATE") {
+      return json(
+        { ok: false, error: "Appending to the out channel requires secret key authentication" },
+        { status: 403 }
+      );
+    }
+
     const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
       session,
     });
@@ -132,25 +147,54 @@ const { action, loader } = createActionApiRoute(
     const addressingKey = canonicalSessionAddressingKey(session, params.session);
 
     const part = await request.text();
-    const partId = request.headers.get("X-Part-Id") ?? nanoid(7);
+    const clientPartId = request.headers.get("X-Part-Id");
+    const partId = clientPartId ?? nanoid(7);
 
-    const [appendError] = await tryCatch(
-      realtimeStream.appendPartToSessionStream(part, partId, addressingKey, params.io)
-    );
+    // Idempotency on client-supplied part ids: atomically claim the id before
+    // appending. A concurrent or retried POST that loses the claim skips the
+    // append (no duplicate record) but still falls through to the drain below,
+    // so a retry whose first attempt died before waking the waitpoint can still
+    // recover it. The claim is released on append failure so a genuine retry
+    // can re-claim and proceed.
+    const wonClaim = clientPartId
+      ? await claimSessionStreamPart(
+          authentication.environment.id,
+          addressingKey,
+          params.io,
+          clientPartId
+        )
+      : true;
+
+    if (wonClaim) {
+      const [appendError] = await tryCatch(
+        realtimeStream.appendPartToSessionStream(part, partId, addressingKey, params.io)
+      );
 
-    if (appendError) {
-      if (appendError instanceof ServiceValidationError) {
+      if (appendError) {
+        if (clientPartId) {
+          await releaseSessionStreamPart(
+            authentication.environment.id,
+            addressingKey,
+            params.io,
+            clientPartId
+          );
+        }
+        if (appendError instanceof ServiceValidationError) {
+          return json(
+            { ok: false, error: appendError.message },
+            { status: appendError.status ?? 422 }
+          );
+        }
+        logger.error("Failed to append to session stream", {
+          sessionId: session.id,
+          io: params.io,
+          error: appendError,
+        });
         return json(
-          { ok: false, error: appendError.message },
-          { status: appendError.status ?? 422 }
+          { ok: false, error: "Something went wrong, please try again." },
+          { status: 500 }
         );
       }
-      logger.error("Failed to append to session stream", {
-        sessionId: session.id,
-        io: params.io,
-        error: appendError,
-      });
-      return json({ ok: false, error: "Something went wrong, please try again." }, { status: 500 });
     }
 
     // Fire any run-scoped waitpoints registered against this channel. Best
@@ -160,7 +204,7 @@ const { action, loader } = createActionApiRoute(
     // `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)
+      drainSessionStreamWaitpoints(authentication.environment.id, addressingKey, params.io)
     );
     if (drainError) {
       logger.error("Failed to drain session stream waitpoints", {

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.realtime.v1.sessions.$session.$io.append.ts

@@ -128,7 +128,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
   // Drain any waitpoints registered for this channel — same as the
   // public append. Best-effort; failure doesn't fail the append.
   const [drainError, waitpointIds] = await tryCatch(
-    drainSessionStreamWaitpoints(addressingKey, io)
+    drainSessionStreamWaitpoints(environment.id, addressingKey, io)
   );
   if (drainError) {
     logger.error("Failed to drain session stream waitpoints (playground)", {

apps/webapp/app/services/realtime/sessions.server.ts

@@ -75,10 +75,10 @@ export function canonicalSessionAddressingKey(
  *
  * 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.
+ * that emit `SessionItem`s must translate: single-row endpoints via
+ * {@link serializeSessionWithFriendlyRunId}, list endpoints via the
+ * batched {@link serializeSessionsWithFriendlyRunIds}. Never put this
+ * raw form on the wire directly.
  */
 export function serializeSession(session: Session): SessionItem {
   return {
@@ -125,3 +125,38 @@ export async function serializeSessionWithFriendlyRunId(
     currentRunId: run?.friendlyId ?? null,
   };
 }
+
+/**
+ * Batched form of {@link serializeSessionWithFriendlyRunId} for list
+ * endpoints: one `IN` lookup per page instead of N+1. `currentRunId` on
+ * the wire is always the public `run_*` friendlyId — the raw
+ * {@link serializeSession} form leaks the internal cuid, which customers
+ * can't use with `runs.retrieve(...)`.
+ */
+export async function serializeSessionsWithFriendlyRunIds(
+  sessions: Session[],
+  scope: { projectId: string; runtimeEnvironmentId: string }
+): Promise<SessionItem[]> {
+  const runIds = [...new Set(sessions.map((s) => s.currentRunId).filter((id): id is string => !!id))];
+
+  // `currentRunId` is a plain string pointer (no FK), so scope the lookup to
+  // the caller's tenant — a stale value must not resolve a run in another env.
+  const runs = runIds.length
+    ? await $replica.taskRun.findMany({
+        where: {
+          id: { in: runIds },
+          projectId: scope.projectId,
+          runtimeEnvironmentId: scope.runtimeEnvironmentId,
+        },
+        select: { id: true, friendlyId: true },
+      })
+    : [];
+  const friendlyIdByRunId = new Map(runs.map((run) => [run.id, run.friendlyId]));
+
+  return sessions.map((session) => ({
+    ...serializeSession(session),
+    currentRunId: session.currentRunId
+      ? friendlyIdByRunId.get(session.currentRunId) ?? null
+      : null,
+  }));
+}