feat(webapp,run-engine): per-org S2 basin migration (TRI-9073)

Move from a single shared basin (`triggerdotdev-prod-streams`) with a
30d blanket retention to per-org basins with retention tied to the
org's billing plan (free 7d / hobby 30d / pro 365d). Fixes
TRIGGER-CLOUD-KK noise from S2 deleting streams out from under live
chat sessions when the basin retention fires before the session ends,
gives per-org cost attribution via S2's basin metrics API, and
narrows the blast radius of any leaked scoped token.

OSS / s2-lite installs are unaffected: provisioning is gated by
`REALTIME_STREAMS_PER_ORG_BASINS_ENABLED` and the read precedence
falls back to the global basin env var when an entity has no stamped
basin.

Schema: nullable `streamBasinName` on Organization / TaskRun / Session
(migration `20260504071227_add_stream_basin_name`). Stamped at
provisioning / trigger / session-create. Reads resolve via
`run.streamBasinName ?? session.streamBasinName ?? legacy basin`.

Provisioner: new `streamBasinProvisioner.server.ts` creates basins
via S2's `POST /v1/basins`, reconfigures via `PATCH /v1/basins/{name}`,
maps plan codes to retention durations. Idempotent on race / pre-existing
basin (treats 409 as success). Org create wires it synchronously with
soft-fail. Plan changes in `setPlan` enqueue
`v3.reconfigureStreamBasinForOrg` next to existing billing-cache
invalidations.

Worker jobs: `v3.provisionStreamBasinForOrg` (backfill / retry) and
`v3.reconfigureStreamBasinForOrg` (plan change) on commonWorker.

Read path: `getRealtimeStreamInstance` becomes a factory keyed by
`{ run, session }` basin context; stream prefix drops `org/{orgId}`
segment for per-org basins (basin already isolates) and keeps it for
the legacy basin (orgs share). Access-token cache key includes basin
to prevent cross-contamination.

Admin routes:
  POST /admin/api/v1/stream-basins/backfill — fan out provisioning
    jobs for every org with `streamBasinName: null`. dryRun + limit
    flags. GET returns progress (`provisioned / totalOrgs`).
  POST /admin/api/v1/stream-basins/reconfigure — enqueue worker job
    (queued mode) or run inline with `tier` override (escape hatch).

Run-engine: `streamBasinName` added to `TriggerParams` (optional);
the V2 trigger path stamps it onto the new TaskRun. No changes to
`MinimalAuthenticatedEnvironment` — stamping is a trigger-time concern,
not a queue concern.

Verified end-to-end with chat.agent on local: backfill creates basins
with right retention (7d free), reconfigure flips retention via plan
change (30d hobby / 365d pro), chat streams land in the per-org basin,
zero leakage to the legacy basin, multi-turn reuses the same in/out
stream pair.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1506,6 +1506,28 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
     REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
     REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
+    /// Per-org basin migration. When "true", the webapp provisions a
+    /// dedicated S2 basin per org with plan-tied retention and stamps
+    /// `streamBasinName` on new TaskRun / Session rows. OSS / s2-lite
+    /// installs leave this off and keep using the single basin defined
+    /// by `REALTIME_STREAMS_S2_BASIN`.
+    REALTIME_STREAMS_PER_ORG_BASINS_ENABLED: z.enum(["true", "false"]).default("false"),
+    /// Naming pattern for per-org basins: `{prefix}-{env}-org-{slug}`
+    /// e.g. `triggerdotdev-prod-org-acme-corp`. Cluster + tier shorthand
+    /// — kept short to stay under S2's basin-name length limit.
+    REALTIME_STREAMS_BASIN_NAME_PREFIX: z.string().default("triggerdotdev"),
+    REALTIME_STREAMS_BASIN_NAME_ENV: z.string().default("dev"),
+    /// Plan-tier retention strings (S2 duration syntax: 7d / 30d / 1y).
+    /// Free / hobby / pro line up with billing tiers; enterprise uses
+    /// the pro default and is reconfigured per-contract via the API.
+    REALTIME_STREAMS_BASIN_RETENTION_FREE: z.string().default("7d"),
+    REALTIME_STREAMS_BASIN_RETENTION_HOBBY: z.string().default("30d"),
+    REALTIME_STREAMS_BASIN_RETENTION_PRO: z.string().default("365d"),
+    /// Storage class applied to per-org basins at create time.
+    REALTIME_STREAMS_BASIN_STORAGE_CLASS: z.enum(["express", "standard"]).default("express"),
+    /// `delete_on_empty_min_age` applied to per-org basins. Streams
+    /// that go empty for this long are reaped automatically.
+    REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE: z.string().default("1h"),
     REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
     WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),
 

apps/webapp/app/models/organization.server.ts

@@ -14,6 +14,8 @@ import { env } from "~/env.server";
 import { featuresForUrl } from "~/features.server";
 import { createApiKeyForEnv, createPkApiKeyForEnv, envSlug } from "./api-key.server";
 import { getDefaultEnvironmentConcurrencyLimit } from "~/services/platform.v3.server";
+import { logger } from "~/services/logger.server";
+import { provisionBasinForOrg } from "~/services/realtime/streamBasinProvisioner.server";
 export type { Organization };
 
 const nanoid = customAlphabet("1234567890abcdef", 4);
@@ -82,6 +84,25 @@ export async function createOrganization(
     },
   });
 
+  // Provision the org's S2 basin synchronously so the very first run
+  // gets `streamBasinName` stamped via the existing org read. Soft-fail
+  // on S2 errors so a transient outage doesn't block signup — the
+  // backfill reconciler picks up any org left with `streamBasinName: null`.
+  // No-op when `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=false` (OSS mode).
+  try {
+    await provisionBasinForOrg({
+      id: organization.id,
+      slug: organization.slug,
+      tier: "free", // new orgs always start on free retention
+      streamBasinName: organization.streamBasinName,
+    });
+  } catch (error) {
+    logger.warn("[createOrganization] streamBasin provisioning failed; backfill will retry", {
+      orgId: organization.id,
+      error: error instanceof Error ? error.message : String(error),
+    });
+  }
+
   return { ...organization };
 }
 

apps/webapp/app/routes/admin.api.v1.stream-basins.backfill.ts

@@ -0,0 +1,165 @@
+import { json, type ActionFunctionArgs } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+import { isPerOrgBasinsEnabled } from "~/services/realtime/streamBasinProvisioner.server";
+import { commonWorker } from "~/v3/commonWorker.server";
+import { logger } from "~/services/logger.server";
+
+/**
+ * One-shot backfill that enqueues `v3.provisionStreamBasinForOrg` for
+ * every org with `streamBasinName: null`. Idempotent — re-running picks
+ * up only the orgs that haven't been provisioned yet, and the worker
+ * job itself is also idempotent (the provisioner short-circuits if the
+ * org column is already set).
+ *
+ *  - Admin auth via `requireAdminApiRequest` (PAT in `Authorization`).
+ *  - Refuses to run when `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=false`
+ *    so OSS / s2-lite installs can't accidentally trigger basin
+ *    creation against a misconfigured backend.
+ *  - `dryRun=true` (default false) returns the count without enqueueing.
+ *  - `limit` (default 1000, max 10000) caps a single invocation. Run
+ *    again to process more — the column filter naturally walks the
+ *    queue forward each call.
+ *  - Each job is keyed `provisionStreamBasin:<orgId>` so concurrent
+ *    backfill calls converge to one job per org instead of duplicating.
+ *
+ * Run from a shell:
+ *   curl -X POST -H "Authorization: Bearer $PAT" \
+ *     "https://api.trigger.dev/admin/api/v1/stream-basins/backfill?limit=200&dryRun=true"
+ */
+
+const BodySchema = z
+  .object({
+    dryRun: z.boolean().optional().default(false),
+    limit: z.number().int().min(1).max(10_000).optional().default(1000),
+  })
+  .strict();
+
+type BackfillResponse = {
+  ok: true;
+  dryRun: boolean;
+  enqueued: number;
+  pending: number;
+  remaining: number;
+  orgIds: string[];
+};
+
+export async function action({ request }: ActionFunctionArgs) {
+  await requireAdminApiRequest(request);
+
+  if (!isPerOrgBasinsEnabled()) {
+    return json(
+      {
+        ok: false,
+        error:
+          "Per-org stream basins are disabled. Set REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=true before running the backfill.",
+      },
+      { status: 400 }
+    );
+  }
+
+  // `application/json` POST body — empty body falls back to defaults so
+  // a parameterless POST does the right thing for the default backfill.
+  let parsed: z.infer<typeof BodySchema>;
+  try {
+    const text = await request.text();
+    const raw = text.length > 0 ? JSON.parse(text) : {};
+    const result = BodySchema.safeParse(raw);
+    if (!result.success) {
+      return json({ ok: false, error: result.error.flatten() }, { status: 400 });
+    }
+    parsed = result.data;
+  } catch {
+    return json({ ok: false, error: "Invalid JSON body" }, { status: 400 });
+  }
+
+  const { dryRun, limit } = parsed;
+
+  // Page candidate orgs. Ordered by createdAt so re-runs walk the queue
+  // forward predictably; deletedAt filter avoids resurrecting orgs.
+  const candidates = await prisma.organization.findMany({
+    where: {
+      streamBasinName: null,
+      deletedAt: null,
+    },
+    orderBy: { createdAt: "asc" },
+    take: limit,
+    select: { id: true },
+  });
+
+  // Total count of remaining nulls (for progress reporting).
+  const remainingTotal = await prisma.organization.count({
+    where: { streamBasinName: null, deletedAt: null },
+  });
+
+  if (dryRun) {
+    const response: BackfillResponse = {
+      ok: true,
+      dryRun: true,
+      enqueued: 0,
+      pending: candidates.length,
+      remaining: Math.max(0, remainingTotal - candidates.length),
+      orgIds: candidates.map((o) => o.id),
+    };
+    return json(response);
+  }
+
+  // Enqueue one job per org. Per-org dedupe key collapses concurrent
+  // backfill calls into a single pending job, and a job that's already
+  // run (basin set) is a no-op on the worker side.
+  let enqueued = 0;
+  for (const org of candidates) {
+    try {
+      await commonWorker.enqueue({
+        job: "v3.provisionStreamBasinForOrg",
+        payload: { orgId: org.id },
+        id: `provisionStreamBasin:${org.id}`,
+      });
+      enqueued += 1;
+    } catch (error) {
+      logger.error("[stream-basins-backfill] enqueue failed", {
+        orgId: org.id,
+        error: error instanceof Error ? error.message : String(error),
+      });
+    }
+  }
+
+  const response: BackfillResponse = {
+    ok: true,
+    dryRun: false,
+    enqueued,
+    pending: candidates.length,
+    remaining: Math.max(0, remainingTotal - enqueued),
+    orgIds: candidates.map((o) => o.id),
+  };
+
+  logger.info("[stream-basins-backfill] enqueued provisioning jobs", {
+    enqueued,
+    candidates: candidates.length,
+    remaining: response.remaining,
+  });
+
+  return json(response);
+}
+
+// GET returns the current state without doing anything — useful for
+// monitoring "is the backfill done yet?" from a dashboard / curl.
+export async function loader({ request }: ActionFunctionArgs) {
+  await requireAdminApiRequest(request);
+
+  const totalOrgs = await prisma.organization.count({ where: { deletedAt: null } });
+  const provisioned = await prisma.organization.count({
+    where: { deletedAt: null, NOT: { streamBasinName: null } },
+  });
+  const remaining = totalOrgs - provisioned;
+
+  return json({
+    ok: true,
+    perOrgBasinsEnabled: isPerOrgBasinsEnabled(),
+    totalOrgs,
+    provisioned,
+    remaining,
+    completion: totalOrgs === 0 ? 1 : provisioned / totalOrgs,
+  });
+}

apps/webapp/app/routes/admin.api.v1.stream-basins.reconfigure.ts

@@ -0,0 +1,63 @@
+import { json, type ActionFunctionArgs } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+import {
+  isPerOrgBasinsEnabled,
+  reconfigureBasinForOrg,
+  type StreamBasinTier,
+} from "~/services/realtime/streamBasinProvisioner.server";
+import { commonWorker } from "~/v3/commonWorker.server";
+
+/**
+ * Admin trigger for `v3.reconfigureStreamBasinForOrg`. The plan-change
+ * path in `setPlan` already enqueues this automatically in cloud mode;
+ * this route exists for ops + e2e testing.
+ *
+ * - Default (`{ orgId }`): enqueues the worker job which resolves the
+ *   tier via `getCurrentPlan` and PATCHes the basin to match. No-op
+ *   locally because `getCurrentPlan` is gated to cloud hosts.
+ * - With `tier`: bypasses the billing lookup and runs reconfigure
+ *   inline against the given tier. Useful for validating the PATCH
+ *   wire shape end-to-end and as a manual override (e.g. enterprise
+ *   contract retention).
+ */
+const BodySchema = z
+  .object({
+    orgId: z.string(),
+    tier: z.enum(["free", "hobby", "pro"]).optional(),
+  })
+  .strict();
+
+export async function action({ request }: ActionFunctionArgs) {
+  await requireAdminApiRequest(request);
+
+  if (!isPerOrgBasinsEnabled()) {
+    return json(
+      { ok: false, error: "Per-org stream basins are disabled." },
+      { status: 400 }
+    );
+  }
+
+  const text = await request.text();
+  const parsed = BodySchema.safeParse(JSON.parse(text));
+  if (!parsed.success) {
+    return json({ ok: false, error: parsed.error.flatten() }, { status: 400 });
+  }
+
+  if (parsed.data.tier) {
+    // Direct, synchronous reconfigure with the explicit tier override.
+    // Skips the worker queue + billing lookup so the PATCH is verifiable
+    // in the response. Errors surface as 500.
+    const tier: StreamBasinTier = parsed.data.tier;
+    await reconfigureBasinForOrg(parsed.data.orgId, tier);
+    return json({ ok: true, mode: "inline", orgId: parsed.data.orgId, tier });
+  }
+
+  await commonWorker.enqueue({
+    job: "v3.reconfigureStreamBasinForOrg",
+    payload: { orgId: parsed.data.orgId },
+    id: `reconfigureStreamBasin:${parsed.data.orgId}`,
+  });
+
+  return json({ ok: true, mode: "queued", enqueued: parsed.data.orgId });
+}

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

@@ -40,6 +40,7 @@ const { action, loader } = createActionApiRoute(
           id: true,
           friendlyId: true,
           realtimeStreamsVersion: true,
+          streamBasinName: true,
         },
       });
 
@@ -98,7 +99,8 @@ const { action, loader } = createActionApiRoute(
         try {
           const realtimeStream = getRealtimeStreamInstance(
             authentication.environment,
-            run.realtimeStreamsVersion
+            run.realtimeStreamsVersion,
+            { run }
           );
 
           const records = await realtimeStream.readRecords(

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

@@ -128,7 +128,9 @@ const { action, loader } = createActionApiRoute(
           // 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");
+          const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+            session: maybeSession,
+          });
 
           if (realtimeStream instanceof S2RealtimeStreams) {
             const records = await realtimeStream.readSessionStreamRecords(

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

@@ -167,6 +167,10 @@ const { action } = createActionApiRoute(
             runtimeEnvironmentId: authentication.environment.id,
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
+            // Stamp the org's S2 basin so realtime reads on this
+            // session's `.in/.out` channels resolve without joining
+            // Organization. Null until per-org basins are provisioned.
+            streamBasinName: authentication.environment.organization.streamBasinName,
           },
           update: { triggerConfig: triggerConfigJson },
         });
@@ -186,6 +190,10 @@ const { action } = createActionApiRoute(
             runtimeEnvironmentId: authentication.environment.id,
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
+            // Stamp the org's S2 basin so realtime reads on this
+            // session's `.in/.out` channels resolve without joining
+            // Organization. Null until per-org basins are provisioned.
+            streamBasinName: authentication.environment.organization.streamBasinName,
           },
         });
       }

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

@@ -81,7 +81,9 @@ const { action, loader } = createActionApiRoute(
       );
     }
 
-    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session,
+    });
 
     if (!(realtimeStream instanceof S2RealtimeStreams)) {
       return json(

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

@@ -59,7 +59,9 @@ const { action } = createActionApiRoute(
       });
     }
 
-    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: maybeSession,
+    });
 
     if (!(realtimeStream instanceof S2RealtimeStreams)) {
       return new Response("Session channels require the S2 realtime backend", {
@@ -122,7 +124,9 @@ const loader = createLoaderApiRoute(
     },
   },
   async ({ params, request, authentication, resource }) => {
-    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: resource.row,
+    });
 
     if (!(realtimeStream instanceof S2RealtimeStreams)) {
       return new Response("Session channels require the S2 realtime backend", {