fix(webapp): address review feedback on session route hardening

Scope the batched session run-id lookup to the caller environment and
project so a stale currentRunId pointer cannot resolve a run in another
tenant. Escape the user-supplied segments of the append idempotency key
so a colon in an externalId or X-Part-Id cannot collide and falsely
suppress an append. Keep the waitpoint drain running on an idempotent
retry: a duplicate append is skipped but still drains, so a retry whose
first attempt died before waking the waitpoint can still recover it.

Author: ericallam

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

@@ -108,7 +108,11 @@ export const loader = createLoaderApiRoute(
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
           }) as Session
-      )
+      ),
+      {
+        projectId: authentication.environment.projectId,
+        runtimeEnvironmentId: authentication.environment.id,
+      }
     );
 
     return json<ListSessionsResponseBody>({

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

@@ -151,49 +151,49 @@ const { action, loader } = createActionApiRoute(
     const partId = clientPartId ?? nanoid(7);
 
     // Idempotency on client-supplied part ids: a retried POST whose first
-    // attempt committed is acknowledged without a second append (which
-    // would duplicate the record and double-fire the waitpoint drain).
-    // The marker is only written after a successful append, so retries of
-    // genuinely failed appends still go through. Server-generated ids are
-    // per-request and carry no dedupe meaning.
-    if (
-      clientPartId &&
+    // attempt committed skips the second append (which would duplicate the
+    // record), but still falls through to the drain below — so a retry whose
+    // first attempt died before waking the waitpoint can still recover it.
+    const alreadyAppended =
+      !!clientPartId &&
       (await wasSessionStreamPartAppended(
         authentication.environment.id,
         addressingKey,
         params.io,
         clientPartId
-      ))
-    ) {
-      return json({ ok: true }, { status: 200 });
-    }
+      ));
 
-    const [appendError] = await tryCatch(
-      realtimeStream.appendPartToSessionStream(part, partId, addressingKey, params.io)
-    );
+    if (!alreadyAppended) {
+      const [appendError] = await tryCatch(
+        realtimeStream.appendPartToSessionStream(part, partId, addressingKey, params.io)
+      );
 
-    if (appendError) {
-      if (appendError instanceof ServiceValidationError) {
+      if (appendError) {
+        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 });
-    }
 
-    if (clientPartId) {
-      await markSessionStreamPartAppended(
-        authentication.environment.id,
-        addressingKey,
-        params.io,
-        clientPartId
-      );
+      if (clientPartId) {
+        await markSessionStreamPartAppended(
+          authentication.environment.id,
+          addressingKey,
+          params.io,
+          clientPartId
+        );
+      }
     }
 
     // Fire any run-scoped waitpoints registered against this channel. Best

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

@@ -134,13 +134,20 @@ export async function serializeSessionWithFriendlyRunId(
  * can't use with `runs.retrieve(...)`.
  */
 export async function serializeSessionsWithFriendlyRunIds(
-  sessions: Session[]
+  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 } },
+        where: {
+          id: { in: runIds },
+          projectId: scope.projectId,
+          runtimeEnvironmentId: scope.runtimeEnvironmentId,
+        },
         select: { id: true, friendlyId: true },
       })
     : [];

apps/webapp/app/services/sessionStreamWaitpointCache.server.ts

@@ -167,7 +167,12 @@ function buildAppendDedupeKey(
   io: "out" | "in",
   partId: string
 ): string {
-  return `${APPEND_DEDUPE_PREFIX}${environmentId}:${addressingKey}:${io}:${partId}`;
+  // Encode the free-form segments — `addressingKey` (externalId) and `partId`
+  // (X-Part-Id) are user-supplied and may contain `:`, which would otherwise
+  // let different tuples collide and falsely suppress an append.
+  return `${APPEND_DEDUPE_PREFIX}${encodeURIComponent(environmentId)}:${encodeURIComponent(
+    addressingKey
+  )}:${io}:${encodeURIComponent(partId)}`;
 }
 
 /**