fix(webapp): row-optional session-channel routes default to org basin

PUT /realtime/v1/sessions/:session/:io and the SSE GET loader on the
same path are row-optional — `:session` may be a `chatId` (externalId)
that hasn't been upserted yet. When the row is missing, both used to
fall through `resolveStreamBasin` to the legacy global basin. If the
row was then created with a per-org basin stamp, follow-up appends and
SSE subscribes resolved to per-org while the PUT-returned headers
still pointed at legacy — caller writes via those headers landed in
the wrong place.

Resolve via the org's current basin when the row is absent. A fresh
session row would be stamped with that same basin at create time, so
all subsequent ops converge. Pre-migration rows (row exists, column
null) keep their legacy fallback because `organization` is only passed
in the no-row branch — `session.streamBasinName === null` still falls
through to the env var, not to the org column.

Verified by curl: PUT against a fresh externalId for an org with a
per-org basin returns `X-S2-Basin: triggerdotdev-dev-org-<orgId>`;
same call for a free-org key still returns the legacy basin.

Author: ericallam

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

@@ -59,8 +59,18 @@ const { action } = createActionApiRoute(
       });
     }
 
+    // When the row is missing (externalId form, row not yet upserted),
+    // default to the org's current basin instead of falling through to
+    // the legacy global. A fresh session row would be stamped with the
+    // org's basin at creation time, so subsequent appends/subscribes
+    // would resolve to the same place — without this, PUT-returned
+    // headers point at legacy and the actual writes go to per-org once
+    // the row exists. Pre-migration rows (row exists, column null) keep
+    // their existing legacy behaviour because we only fall back to org
+    // when there's no row at all.
     const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
       session: maybeSession,
+      organization: maybeSession ? null : authentication.environment.organization,
     });
 
     if (!(realtimeStream instanceof S2RealtimeStreams)) {
@@ -124,8 +134,13 @@ const loader = createLoaderApiRoute(
     },
   },
   async ({ params, request, authentication, resource }) => {
+    // Same row-optional reasoning as the PUT handler above: if no row
+    // exists yet, resolve via the org's current basin so the SSE
+    // subscribe lands in the same place that subsequent appends will
+    // (once the row gets created and stamped).
     const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
       session: resource.row,
+      organization: resource.row ? null : authentication.environment.organization,
     });
 
     if (!(realtimeStream instanceof S2RealtimeStreams)) {

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

@@ -34,27 +34,37 @@ export const v1RealtimeStreams = singleton("realtimeStreams", initializeRedisRea
  *
  *  1. `run.streamBasinName` (set at trigger time, immutable per-run)
  *  2. `session.streamBasinName` (set at session create time)
- *  3. `REALTIME_STREAMS_S2_BASIN` (the legacy / OSS / pre-backfill global)
+ *  3. `organization.streamBasinName` (current org basin — only useful
+ *     when neither a run nor a session row exists yet, e.g. PUT init
+ *     against an externalId before the row is created)
+ *  4. `REALTIME_STREAMS_S2_BASIN` (the legacy / OSS / pre-backfill global)
  *
  * Old runs / sessions that pre-date the per-org-basins migration carry
  * `null` columns and fall through to the global basin, which is the
  * one their streams were originally created in. Once the legacy basin
  * drains via S2 retention (~30d on prod today), this fallback can be
  * dropped — but it's cheap to keep as a safety net.
  *
+ * Callers should only pass `organization` when they know the row-bearing
+ * ref is absent (not when its column is null) — otherwise a pre-migration
+ * row's null column would short-circuit to the org's *current* basin
+ * instead of the legacy one its streams actually live in.
+ *
  * OSS / s2-lite installs always hit the global path because the
  * provisioner is gated by `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED`
  * and `streamBasinName` is never written.
  */
 export type StreamBasinContext = {
   run?: { streamBasinName: string | null } | null;
   session?: { streamBasinName: string | null } | null;
+  organization?: { streamBasinName: string | null } | null;
 };
 
 export function resolveStreamBasin(ctx: StreamBasinContext): string | undefined {
   return (
     ctx.run?.streamBasinName ??
     ctx.session?.streamBasinName ??
+    ctx.organization?.streamBasinName ??
     env.REALTIME_STREAMS_S2_BASIN ??
     undefined
   );