feat(webapp): per-org S2 basin migration (#3516)

## Summary Move from a single shared S2 basin to **per-org basins** with retention tied to the org's billing plan. Stops S2 from deleting streams out from under live chat sessions when basin retention fires before the chat ends, and unlocks per-org cost attribution. OSS / s2-lite installs are unaffected: provisioning is gated by `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED` (default `false`), and the read precedence falls back to the global basin env var when an entity has no stamped basin. ``` basin = run.streamBasinName ?? session.streamBasinName ?? env.REALTIME_STREAMS_S2_BASIN ``` ## Design Three nullable `streamBasinName` columns (`Organization`, `TaskRun`, `Session`) plus a provisioner that idempotently creates the basin and reconfigures retention on plan changes. The trigger and session-create paths stamp the org's basin onto new rows; the realtime read path picks the basin from the entity context. Admin routes back-fill existing orgs and force-reconfigure a single org. ## Test plan - [x] `pnpm run typecheck --filter webapp --filter @internal/run-engine` - [x] Backfill admin route end-to-end (provision + DB stamp + S2 basin config). - [x] Reconfigure on plan change (all retention tiers). - [x] chat.agent multi-turn drives streams into the per-org basin. - [x] Legacy fallback when entity has no stamped basin. - [x] Provisioner is a no-op when the flag is off.
2026-05-05 10:06:58 +01:00
parent 3d418a9482
commit 386b4f65ff
23 changed files with 548 additions and 59 deletions
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Per-org S2 stream basins with retention tied to the org's billing plan, gated by `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED`. Stops basin retention from deleting streams out from under live chat sessions and unlocks per-org cost attribution.
@@ -3,6 +3,15 @@ import { MachinePresetName } from "@trigger.dev/core/v3";
 import { BoolEnv } from "./utils/boolEnv";
 import { isValidDatabaseUrl } from "./utils/db";
 import { isValidRegex } from "./utils/regex";
+import { isValidDuration } from "./services/realtime/duration.server";
+
+// `z.string()` constrained to a `parseDuration`-parseable string (e.g.
+// `7d`, `1h`). Validated at boot so a typo'd duration fails fast.
+function durationString() {
+  return z
+    .string()
+    .refine(isValidDuration, "must be a duration like 7d, 30d, 365d, 1h, 1y");
+}

 // Parses a CSV of machine preset names (e.g. "small-1x,small-2x") into a
 // non-empty array of MachinePresetName. Used by COMPUTE_TEMPLATE_MACHINE_PRESETS
@@ -1506,6 +1515,16 @@ 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),
+    // When "true", provision a dedicated S2 basin per org and stamp
+    // `streamBasinName` on new rows. Off keeps everything on the single
+    // basin defined by `REALTIME_STREAMS_S2_BASIN`.
+    REALTIME_STREAMS_PER_ORG_BASINS_ENABLED: z.enum(["true", "false"]).default("false"),
+    // Per-org basin name = `{prefix}-{env}-org-{orgId}`.
+    REALTIME_STREAMS_BASIN_NAME_PREFIX: z.string().default("triggerdotdev"),
+    REALTIME_STREAMS_BASIN_NAME_ENV: z.string().default("dev"),
+    REALTIME_STREAMS_BASIN_DEFAULT_RETENTION: durationString().default("30d"),
+    REALTIME_STREAMS_BASIN_STORAGE_CLASS: z.enum(["express", "standard"]).default("express"),
+    REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE: durationString().default("1h"),
    REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
    WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),

@@ -0,0 +1,47 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+import { isValidDuration } from "~/services/realtime/duration.server";
+import {
+  deprovisionBasinForOrg,
+  ensureBasinForOrg,
+} from "~/services/realtime/streamBasinProvisioner.server";
+
+const ParamsSchema = z.object({ organizationId: z.string() });
+
+const BodySchema = z.discriminatedUnion("action", [
+  z.object({
+    action: z.literal("ensure"),
+    retention: z
+      .string()
+      .refine(isValidDuration, "retention must be a duration like 7d, 30d, 365d, 1h, 1y"),
+  }),
+  z.object({ action: z.literal("deprovision") }),
+]);
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  await requireAdminApiRequest(request);
+
+  const { organizationId } = ParamsSchema.parse(params);
+
+  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 });
+  }
+
+  if (parsed.action === "ensure") {
+    const result = await ensureBasinForOrg(organizationId, parsed.retention);
+    return json({ ok: true, ...result });
+  }
+
+  const result = await deprovisionBasinForOrg(organizationId);
+  return json({ ok: true, ...result });
+}
@@ -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(
@@ -123,12 +123,14 @@ const { action, loader } = createActionApiRoute(
      // and remove the pending registration.
      if (!result.isCached) {
        try {
-          // Session streams are always v2 (S2) — the writer in
-          // `appendPartToSessionStream` and the SSE subscribe both
-          // 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");
+          // Match the writer's basin resolution exactly: session if the
+          // row exists, otherwise the org so we look at the same basin a
+          // fresh row would be stamped with. Mirrors the PUT/GET sister
+          // routes in `realtime.v1.sessions.$session.$io.ts`.
+          const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+            session: maybeSession,
+            organization: maybeSession ? null : authentication.environment.organization,
+          });

          if (realtimeStream instanceof S2RealtimeStreams) {
            const records = await realtimeStream.readSessionStreamRecords(
@@ -167,6 +167,7 @@ const { action } = createActionApiRoute(
            runtimeEnvironmentId: authentication.environment.id,
            environmentType: authentication.environment.type,
            organizationId: authentication.environment.organizationId,
+            streamBasinName: authentication.environment.organization.streamBasinName,
          },
          update: { triggerConfig: triggerConfigJson },
        });
@@ -186,6 +187,7 @@ const { action } = createActionApiRoute(
            runtimeEnvironmentId: authentication.environment.id,
            environmentType: authentication.environment.type,
            organizationId: authentication.environment.organizationId,
+            streamBasinName: authentication.environment.organization.streamBasinName,
          },
        });
      }
@@ -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(
@@ -59,7 +59,13 @@ const { action } = createActionApiRoute(
      });
    }

-    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
+    // No-row form: resolve via the org so the stream initialised here
+    // matches what later appends/subscribes will land on once the row
+    // is created.
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: maybeSession,
+      organization: maybeSession ? null : authentication.environment.organization,
+    });

    if (!(realtimeStream instanceof S2RealtimeStreams)) {
      return new Response("Session channels require the S2 realtime backend", {
@@ -122,7 +128,11 @@ const loader = createLoaderApiRoute(
    },
  },
  async ({ params, request, authentication, resource }) => {
-    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2");
+    // Same no-row fallback as PUT above.
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: resource.row,
+      organization: resource.row ? null : authentication.environment.organization,
+    });

    if (!(realtimeStream instanceof S2RealtimeStreams)) {
      return new Response("Session channels require the S2 realtime backend", {
@@ -29,6 +29,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
    select: {
      id: true,
      friendlyId: true,
+      streamBasinName: true,
      runtimeEnvironment: {
        include: {
          project: true,
@@ -64,7 +65,9 @@ export async function action({ request, params }: ActionFunctionArgs) {
  }

  // The runtimeEnvironment from the run is already in the correct shape for AuthenticatedEnvironment
-  const realtimeStream = getRealtimeStreamInstance(run.runtimeEnvironment, streamVersion);
+  const realtimeStream = getRealtimeStreamInstance(run.runtimeEnvironment, streamVersion, {
+    run,
+  });

  return realtimeStream.ingestData(
    request.body,
@@ -127,7 +130,8 @@ export const loader = createLoaderApiRoute(

    const realtimeStream = getRealtimeStreamInstance(
      authentication.environment,
-      run.realtimeStreamsVersion
+      run.realtimeStreamsVersion,
+      { run }
    );

    return realtimeStream.streamResponse(request, run.friendlyId, params.streamId, getRequestAbortSignal(), {
@@ -72,6 +72,7 @@ const { action } = createActionApiRoute(
        realtimeStreamsVersion: true,
        completedAt: true,
        id: true,
+        streamBasinName: true,
      },
    });

@@ -102,7 +103,8 @@ const { action } = createActionApiRoute(

    const realtimeStream = getRealtimeStreamInstance(
      authentication.environment,
-      targetRun.realtimeStreamsVersion
+      targetRun.realtimeStreamsVersion,
+      { run: targetRun }
    );

    const partId = request.headers.get("X-Part-Id") ?? nanoid(7);
@@ -26,14 +26,17 @@ const { action } = createActionApiRoute(
      select: {
        id: true,
        friendlyId: true,
+        streamBasinName: true,
        parentTaskRun: {
          select: {
            friendlyId: true,
+            streamBasinName: true,
          },
        },
        rootTaskRun: {
          select: {
            friendlyId: true,
+            streamBasinName: true,
          },
        },
      },
@@ -43,17 +46,20 @@ const { action } = createActionApiRoute(
      return new Response("Run not found", { status: 404 });
    }

-    const targetId =
+    const targetRun =
      params.target === "self"
-        ? run.friendlyId
+        ? run
        : params.target === "parent"
-        ? run.parentTaskRun?.friendlyId
-        : run.rootTaskRun?.friendlyId;
+        ? run.parentTaskRun
+        : run.rootTaskRun;

-    if (!targetId) {
+    if (!targetRun?.friendlyId) {
      return new Response("Target not found", { status: 404 });
    }

+    const targetId = targetRun.friendlyId;
+    const basinContext = { run: { streamBasinName: targetRun.streamBasinName ?? null } };
+
    if (request.method === "PUT") {
      // This is the "create" endpoint
      const updatedRun = await prisma.taskRun.update({
@@ -80,7 +86,8 @@ const { action } = createActionApiRoute(

      const realtimeStream = getRealtimeStreamInstance(
        authentication.environment,
-        updatedRun.realtimeStreamsVersion
+        updatedRun.realtimeStreamsVersion,
+        basinContext
      );

      const { responseHeaders } = await realtimeStream.initializeStream(targetId, params.streamId);
@@ -112,7 +119,11 @@ const { action } = createActionApiRoute(
        resumeFromChunkNumber = parsed;
      }

-      const realtimeStream = getRealtimeStreamInstance(authentication.environment, streamVersion);
+      const realtimeStream = getRealtimeStreamInstance(
+        authentication.environment,
+        streamVersion,
+        basinContext
+      );

      return realtimeStream.ingestData(
        request.body,
@@ -139,14 +150,17 @@ const loader = createLoaderApiRoute(
        select: {
          id: true,
          friendlyId: true,
+          streamBasinName: true,
          parentTaskRun: {
            select: {
              friendlyId: true,
+              streamBasinName: true,
            },
          },
          rootTaskRun: {
            select: {
              friendlyId: true,
+              streamBasinName: true,
            },
          },
        },
@@ -158,17 +172,19 @@ const loader = createLoaderApiRoute(
      return new Response("Run not found", { status: 404 });
    }

-    const targetId =
+    const targetRun =
      params.target === "self"
-        ? run.friendlyId
+        ? run
        : params.target === "parent"
-        ? run.parentTaskRun?.friendlyId
-        : run.rootTaskRun?.friendlyId;
+        ? run.parentTaskRun
+        : run.rootTaskRun;

-    if (!targetId) {
+    if (!targetRun?.friendlyId) {
      return new Response("Target not found", { status: 404 });
    }

+    const targetId = targetRun.friendlyId;
+
    // Handle HEAD request to get last chunk index
    if (request.method !== "HEAD") {
      return new Response("Only HEAD requests are allowed for this endpoint", { status: 405 });
@@ -178,7 +194,11 @@ const loader = createLoaderApiRoute(
    const clientId = request.headers.get("X-Client-Id") || "default";
    const streamVersion = request.headers.get("X-Stream-Version") || "v1";

-    const realtimeStream = getRealtimeStreamInstance(authentication.environment, streamVersion);
+    const realtimeStream = getRealtimeStreamInstance(
+      authentication.environment,
+      streamVersion,
+      { run: { streamBasinName: targetRun.streamBasinName ?? null } }
+    );

    const lastChunkIndex = await realtimeStream.getLastChunkIndex(
      targetId,
@@ -46,6 +46,7 @@ const { action } = createActionApiRoute(
        friendlyId: true,
        completedAt: true,
        realtimeStreamsVersion: true,
+        streamBasinName: true,
      },
    });

@@ -68,7 +69,8 @@ const { action } = createActionApiRoute(

    const realtimeStream = getRealtimeStreamInstance(
      authentication.environment,
-      run.realtimeStreamsVersion
+      run.realtimeStreamsVersion,
+      { run }
    );

    // Build the input stream record (raw user data, no wrapper)
@@ -155,7 +157,8 @@ const loader = createLoaderApiRoute(

    const realtimeStream = getRealtimeStreamInstance(
      authentication.environment,
-      run.realtimeStreamsVersion
+      run.realtimeStreamsVersion,
+      { run }
    );

    // Read from the internal S2 stream name (prefixed to avoid user stream collisions)
@@ -87,7 +87,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {

  const realtimeStream = getRealtimeStreamInstance(
    run.runtimeEnvironment,
-    run.realtimeStreamsVersion
+    run.realtimeStreamsVersion,
+    { run }
  );

  return realtimeStream.streamResponse(request, run.friendlyId, streamKey, getRequestAbortSignal(), {
@@ -395,6 +395,7 @@ export class RunEngineTriggerTaskService {
                bulkActionId: body.options?.bulkActionId,
                planType,
                realtimeStreamsVersion: options.realtimeStreamsVersion,
+                streamBasinName: environment.organization.streamBasinName,
                debounce: body.options?.debounce,
                annotations,
                // When debouncing with triggerAndWait, create a span for the debounced trigger
@@ -44,6 +44,17 @@ function initializeClient() {
 }

 const client = singleton("billingClient", initializeClient);
+
+/**
+ * `true` when the billing client was instantiated — i.e. we're running
+ * in a cloud-style install with `BILLING_API_URL` + `BILLING_API_KEY`
+ * configured. OSS / self-hosted installs return `false` here, which
+ * lets callers distinguish "no billing wired up, fall back to
+ * defaults" from "billing wired up but the call failed, retry."
+ */
+export function isBillingConfigured(): boolean {
+  return client !== undefined;
+}
 // Failures from @trigger.dev/platform billing client calls are tracked via
 // this metric (with low-cardinality {function, kind} labels) rather than
 // logged. Every task invocation hits these paths, so per-call logs were too
@@ -0,0 +1,49 @@
+/**
+ * Duration string parsing for stream-basin retention / delete-on-empty
+ * configuration. Used by `streamBasinProvisioner` (to convert to S2's
+ * integer-seconds wire format) and by `env.server.ts` (to validate
+ * duration-shaped env vars at boot rather than at first use).
+ *
+ * Accepts the short forms (`7d`, `30d`, `365d`, `1h`, `90m`, `45s`,
+ * `2w`, `1y`) and the human forms (`7days`, `1week`, `1year`).
+ */
+
+const PATTERN =
+  /^(\d+)\s*(s|sec|secs|seconds?|m|min|mins|minutes?|h|hour|hours?|d|day|days?|w|week|weeks?|y|year|years?)$/;
+
+export function isValidDuration(input: string): boolean {
+  return PATTERN.test(input.trim().toLowerCase());
+}
+
+/**
+ * Parse a duration string into seconds. Throws on garbage so a
+ * misconfigured env var fails loudly. Use {@link isValidDuration}
+ * for non-throwing validation (e.g. inside a Zod `.refine()`).
+ */
+export function parseDuration(input: string): number {
+  const trimmed = input.trim().toLowerCase();
+  const match = trimmed.match(PATTERN);
+  if (!match) {
+    throw new Error(`Invalid duration string: ${input}`);
+  }
+  const value = parseInt(match[1]!, 10);
+  const unit = match[2]!;
+  const multiplier =
+    /^s/.test(unit)
+      ? 1
+      : /^m(?:in|ins|inute|inutes)?$/.test(unit)
+        ? 60
+        : /^h/.test(unit)
+          ? 3600
+          : /^d/.test(unit)
+            ? 86400
+            : /^w/.test(unit)
+              ? 604800
+              : /^y/.test(unit)
+                ? 31_536_000
+                : NaN;
+  if (!Number.isFinite(multiplier)) {
+    throw new Error(`Invalid duration unit: ${unit}`);
+  }
+  return value * multiplier;
+}
@@ -464,7 +464,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
      return this.s2IssueAccessToken(id);
    }

-    const result = await this.cache.accessToken.swr(this.streamPrefix, async () => {
+    // Cache key includes basin so per-org basins never collide on
+    // cached tokens. `${basin}:${prefix}` is unique per (org-basin, env).
+    const cacheKey = `${this.basin}:${this.streamPrefix}`;
+    const result = await this.cache.accessToken.swr(cacheKey, async () => {
      return this.s2IssueAccessToken(id);
    });

@@ -0,0 +1,248 @@
+/**
+ * Per-org S2 basin provisioning. Gated by
+ * `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED`: when off, all orgs share
+ * `REALTIME_STREAMS_S2_BASIN` and this module no-ops.
+ *
+ * Pure retention-string in / S2-call out. Plan vocabulary lives in the
+ * cloud billing app, which calls into the admin sync route to drive
+ * provisioning + reconfiguration.
+ */
+import type { PrismaClientOrTransaction } from "~/db.server";
+import { prisma } from "~/db.server";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { parseDuration } from "./duration.server";
+
+export function isPerOrgBasinsEnabled(): boolean {
+  return env.REALTIME_STREAMS_PER_ORG_BASINS_ENABLED === "true";
+}
+
+export function defaultRetention(): string {
+  return env.REALTIME_STREAMS_BASIN_DEFAULT_RETENTION;
+}
+
+// Org id is a cuid — fixed-length and stable, so the basin name is
+// collision-free without truncation. Slugs are user-editable and would
+// drift.
+export function basinNameForOrg(org: { id: string }): string {
+  const prefix = env.REALTIME_STREAMS_BASIN_NAME_PREFIX;
+  const envName = env.REALTIME_STREAMS_BASIN_NAME_ENV;
+  return `${prefix}-${envName}-org-${org.id}`;
+}
+
+type ProvisionInput = {
+  id: string;
+  retention?: string;
+  streamBasinName: string | null | undefined;
+};
+
+type ProvisionResult =
+  | { kind: "skipped"; reason: "feature-disabled" | "already-provisioned"; basin: string | null }
+  | { kind: "provisioned"; basin: string; retention: string };
+
+// Idempotent. Treats S2 409 as success (race with another caller, or
+// previous run that crashed after S2 ack but before the column write).
+export async function provisionBasinForOrg(
+  org: ProvisionInput,
+  prismaClient: PrismaClientOrTransaction = prisma
+): Promise<ProvisionResult> {
+  if (!isPerOrgBasinsEnabled()) {
+    return { kind: "skipped", reason: "feature-disabled", basin: null };
+  }
+
+  if (org.streamBasinName) {
+    return { kind: "skipped", reason: "already-provisioned", basin: org.streamBasinName };
+  }
+
+  const accessToken = env.REALTIME_STREAMS_S2_ACCESS_TOKEN;
+  if (!accessToken) {
+    throw new Error(
+      "REALTIME_STREAMS_S2_ACCESS_TOKEN must be set when REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=true"
+    );
+  }
+
+  const basin = basinNameForOrg(org);
+  const retention = org.retention ?? defaultRetention();
+
+  await s2CreateBasin(basin, {
+    accessToken,
+    retentionPolicy: retention,
+    storageClass: env.REALTIME_STREAMS_BASIN_STORAGE_CLASS,
+    deleteOnEmptyMinAge: env.REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE,
+  });
+
+  await prismaClient.organization.update({
+    where: { id: org.id },
+    data: { streamBasinName: basin },
+  });
+
+  logger.info("[streamBasinProvisioner] provisioned basin for org", {
+    orgId: org.id,
+    basin,
+    retention,
+  });
+
+  return { kind: "provisioned", basin, retention };
+}
+
+export async function reconfigureBasinForOrg(
+  orgId: string,
+  retention: string
+): Promise<void> {
+  if (!isPerOrgBasinsEnabled()) return;
+
+  const accessToken = env.REALTIME_STREAMS_S2_ACCESS_TOKEN;
+  if (!accessToken) {
+    throw new Error(
+      "REALTIME_STREAMS_S2_ACCESS_TOKEN must be set when REALTIME_STREAMS_PER_ORG_BASINS_ENABLED=true"
+    );
+  }
+
+  const org = await prisma.organization.findFirst({
+    where: { id: orgId },
+    select: { id: true, streamBasinName: true },
+  });
+  if (!org?.streamBasinName) return;
+
+  await s2ReconfigureBasin(org.streamBasinName, { accessToken, retentionPolicy: retention });
+
+  logger.info("[streamBasinProvisioner] reconfigured basin retention", {
+    orgId,
+    basin: org.streamBasinName,
+    retention,
+  });
+}
+
+type EnsureResult =
+  | { kind: "skipped"; reason: "feature-disabled" | "org-not-found" }
+  | { kind: "provisioned"; basin: string; retention: string }
+  | { kind: "reconfigured"; basin: string; retention: string };
+
+// Idempotent: provisions if the org has no basin, PATCHes retention if
+// it does. The single entrypoint the cloud billing app drives — both
+// for the live plan-change path and the bulk backfill.
+export async function ensureBasinForOrg(
+  orgId: string,
+  retention: string
+): Promise<EnsureResult> {
+  if (!isPerOrgBasinsEnabled()) {
+    return { kind: "skipped", reason: "feature-disabled" };
+  }
+
+  const org = await prisma.organization.findFirst({
+    where: { id: orgId },
+    select: { id: true, streamBasinName: true },
+  });
+  if (!org) return { kind: "skipped", reason: "org-not-found" };
+
+  if (!org.streamBasinName) {
+    const result = await provisionBasinForOrg(
+      { id: org.id, streamBasinName: null, retention }
+    );
+    if (result.kind === "provisioned") {
+      return { kind: "provisioned", basin: result.basin, retention: result.retention };
+    }
+    return { kind: "skipped", reason: "feature-disabled" };
+  }
+
+  await reconfigureBasinForOrg(org.id, retention);
+  return { kind: "reconfigured", basin: org.streamBasinName, retention };
+}
+
+// Inverse of ensureBasinForOrg: nulls the column so future runs/sessions
+// land in the shared global basin. The S2 basin lingers; existing streams
+// age out on their original retention.
+export async function deprovisionBasinForOrg(
+  orgId: string
+): Promise<{ kind: "deprovisioned" } | { kind: "skipped"; reason: "no-basin" }> {
+  const org = await prisma.organization.findFirst({
+    where: { id: orgId },
+    select: { id: true, streamBasinName: true },
+  });
+  if (!org?.streamBasinName) return { kind: "skipped", reason: "no-basin" };
+
+  await prisma.organization.update({
+    where: { id: org.id },
+    data: { streamBasinName: null },
+  });
+
+  logger.info("[streamBasinProvisioner] deprovisioned basin for org", {
+    orgId,
+    previousBasin: org.streamBasinName,
+  });
+
+  return { kind: "deprovisioned" };
+}
+
+// S2 REST: POST /v1/basins to create, PATCH /v1/basins/{name} to
+// reconfigure. Wire shape takes integer seconds; we accept human strings
+// like "7d" / "1y" as env-var ergonomics and parse them here.
+
+type CreateBasinOptions = {
+  accessToken: string;
+  retentionPolicy: string;
+  storageClass: "express" | "standard";
+  deleteOnEmptyMinAge: string;
+};
+
+async function s2CreateBasin(name: string, opts: CreateBasinOptions): Promise<void> {
+  const url = `https://aws.s2.dev/v1/basins`;
+  const body = {
+    basin: name,
+    config: {
+      create_stream_on_append: true,
+      create_stream_on_read: true,
+      default_stream_config: {
+        storage_class: opts.storageClass,
+        retention_policy: { age: parseDuration(opts.retentionPolicy) },
+        delete_on_empty: { min_age_secs: parseDuration(opts.deleteOnEmptyMinAge) },
+      },
+    },
+  };
+
+  const res = await fetch(url, {
+    signal: AbortSignal.timeout(10_000),
+    method: "POST",
+    headers: {
+      Authorization: `Bearer ${opts.accessToken}`,
+      "Content-Type": "application/json",
+    },
+    body: JSON.stringify(body),
+  });
+
+  // 409 = basin already exists; treat as success (idempotent).
+  if (res.ok || res.status === 409) return;
+
+  const text = await res.text().catch(() => "");
+  throw new Error(`S2 createBasin failed: ${res.status} ${res.statusText} ${text}`);
+}
+
+type ReconfigureBasinOptions = {
+  accessToken: string;
+  retentionPolicy: string;
+};
+
+async function s2ReconfigureBasin(name: string, opts: ReconfigureBasinOptions): Promise<void> {
+  const url = `https://aws.s2.dev/v1/basins/${encodeURIComponent(name)}`;
+  const body = {
+    default_stream_config: {
+      retention_policy: { age: parseDuration(opts.retentionPolicy) },
+    },
+  };
+
+  const res = await fetch(url, {
+    signal: AbortSignal.timeout(10_000),
+    method: "PATCH",
+    headers: {
+      Authorization: `Bearer ${opts.accessToken}`,
+      "Content-Type": "application/json",
+    },
+    body: JSON.stringify(body),
+  });
+
+  if (res.ok) return;
+
+  const text = await res.text().catch(() => "");
+  throw new Error(`S2 reconfigureBasin failed: ${res.status} ${res.statusText} ${text}`);
+}
+
@@ -29,41 +29,69 @@ function initializeRedisRealtimeStreams() {

 export const v1RealtimeStreams = singleton("realtimeStreams", initializeRedisRealtimeStreams);

+/**
+ * Resolve a stream's basin. Precedence: run → session → org → global env.
+ * Pre-migration rows have `streamBasinName: null` and fall through to
+ * the global basin (where their streams actually live), so only pass
+ * `organization` when no run/session row exists at all — otherwise a
+ * null column would short-circuit to the org's *current* basin.
+ */
+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
+  );
+}
+
 export function getRealtimeStreamInstance(
  environment: AuthenticatedEnvironment,
-  streamVersion: string
+  streamVersion: string,
+  basinContext?: StreamBasinContext
 ): StreamIngestor & StreamResponder {
  if (streamVersion === "v1") {
    return v1RealtimeStreams;
-  } else {
-    if (
-      env.REALTIME_STREAMS_S2_BASIN &&
-      (env.REALTIME_STREAMS_S2_ACCESS_TOKEN ||
-        env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
-    ) {
-      return new S2RealtimeStreams({
-        basin: env.REALTIME_STREAMS_S2_BASIN,
-        accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN ?? "",
-        endpoint: env.REALTIME_STREAMS_S2_ENDPOINT,
-        skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true",
-        streamPrefix: [
-          "org",
-          environment.organization.id,
-          "env",
-          environment.slug,
-          environment.id,
-        ].join("/"),
-        logLevel: env.REALTIME_STREAMS_S2_LOG_LEVEL,
-        flushIntervalMs: env.REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS,
-        maxRetries: env.REALTIME_STREAMS_S2_MAX_RETRIES,
-        s2WaitSeconds: env.REALTIME_STREAMS_S2_WAIT_SECONDS,
-        accessTokenExpirationInMs: env.REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS,
-        cache: s2RealtimeStreamsCache,
-      });
-    }
-
-    throw new Error("Realtime streams v2 is required for this run but S2 configuration is missing");
  }
+
+  const resolvedBasin = resolveStreamBasin(basinContext ?? {});
+  if (
+    resolvedBasin &&
+    (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
+  ) {
+    return new S2RealtimeStreams({
+      basin: resolvedBasin,
+      accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN ?? "",
+      endpoint: env.REALTIME_STREAMS_S2_ENDPOINT,
+      skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true",
+      streamPrefix: streamPrefixFor(environment, resolvedBasin),
+      logLevel: env.REALTIME_STREAMS_S2_LOG_LEVEL,
+      flushIntervalMs: env.REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS,
+      maxRetries: env.REALTIME_STREAMS_S2_MAX_RETRIES,
+      s2WaitSeconds: env.REALTIME_STREAMS_S2_WAIT_SECONDS,
+      accessTokenExpirationInMs: env.REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS,
+      cache: s2RealtimeStreamsCache,
+    });
+  }
+
+  throw new Error("Realtime streams v2 is required for this run but S2 configuration is missing");
+}
+
+// Shared basin needs `org/{orgId}` to namespace; per-org basin already
+// isolates so the segment drops.
+function streamPrefixFor(environment: AuthenticatedEnvironment, basin: string): string {
+  const isPerOrgBasin = basin !== env.REALTIME_STREAMS_S2_BASIN;
+  const segments = isPerOrgBasin
+    ? ["env", environment.slug, environment.id]
+    : ["org", environment.organization.id, "env", environment.slug, environment.id];
+  return segments.join("/");
 }

 export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
@@ -0,0 +1,8 @@
+-- AlterTable
+ALTER TABLE "public"."Organization" ADD COLUMN IF NOT EXISTS "streamBasinName" TEXT;
+
+-- AlterTable
+ALTER TABLE "public"."Session" ADD COLUMN IF NOT EXISTS "streamBasinName" TEXT;
+
+-- AlterTable
+ALTER TABLE "public"."TaskRun" ADD COLUMN IF NOT EXISTS "streamBasinName" TEXT;
@@ -251,6 +251,13 @@ model Organization {

  platformNotifications PlatformNotification[]
  errorGroupStates      ErrorGroupState[]
+
+  /// S2 basin that holds this org's realtime streams. Null until the
+  /// per-org basin has been provisioned (OSS / s2-lite installs leave
+  /// it null forever; reads fall back to the global basin env var).
+  /// Set once at provisioning time; retention is reconfigured in-place
+  /// when the org's plan changes.
+  streamBasinName String?
 }

 model OrgMember {
@@ -758,6 +765,12 @@ model Session {
  createdAt DateTime @default(now())
  updatedAt DateTime @updatedAt

+  /// S2 basin where this session's stream pair lives. Stamped at create
+  /// time from `Organization.streamBasinName` so reads can resolve the
+  /// basin without joining org. Null when the org has no per-org basin
+  /// (OSS, or pre-backfill); reads fall back to the global basin.
+  streamBasinName String?
+
  runs SessionRun[]

  /// Idempotency: `(env, externalId)` uniquely identifies a session.
@@ -992,6 +1005,11 @@ model TaskRun {
  realtimeStreamsVersion String   @default("v1")
  /// Store the stream keys that are being used by the run
  realtimeStreams        String[] @default([])
+  /// S2 basin where this run's realtime streams live. Stamped at create
+  /// time from `Organization.streamBasinName` so reads can resolve the
+  /// basin without joining org. Null when the org has no per-org basin
+  /// (OSS, or pre-backfill); reads fall back to the global basin.
+  streamBasinName        String?

  @@unique([oneTimeUseToken])
  @@unique([runtimeEnvironmentId, taskIdentifier, idempotencyKey])
@@ -498,6 +498,7 @@ export class RunEngine {
      bulkActionId,
      planType,
      realtimeStreamsVersion,
+      streamBasinName,
      debounce,
      annotations,
      onDebounced,
@@ -660,6 +661,7 @@ export class RunEngine {
              bulkActionGroupIds: bulkActionId ? [bulkActionId] : undefined,
              planType,
              realtimeStreamsVersion,
+              streamBasinName,
              debounce: debounce
                ? {
                  key: debounce.key,
@@ -259,6 +259,7 @@ export type TriggerParams = {
  bulkActionId?: string;
  planType?: string;
  realtimeStreamsVersion?: string;
+  streamBasinName?: string | null;
  debounce?: {
    key: string;
    delay: string;