fix(webapp): keep session runs off the legacy realtime streams backend (#4564)

## Summary

Runs created for a Session were triggered without a realtime streams
version, so they fell through to the `realtimeStreamsVersion` column
default of `v1`. A Session's own `.in` / `.out` channels are always
`v2`, so any run-scoped `streams.append()` or `streams.pipe()` call made
inside a session run wrote to a different backend than the session it
belongs to, and stayed there for the life of the run.

The API trigger routes were never affected. They call
`determineRealtimeStreamsVersion` with the client's
`x-trigger-realtime-streams-version` header and always pass an explicit
value, so a current SDK asking for v2 gets it. Only the internal callers
that build trigger options by hand were leaning on the column default,
which no env var can influence because that path never calls the
resolver at all.

## The version resolver

Fixing the call site exposed a second problem in
`determineRealtimeStreamsVersion`. Its two paths disagreed: an explicit
`v2` was checked against the S2 configuration first, but when the caller
expressed no preference it returned `REALTIME_STREAMS_DEFAULT_VERSION`
verbatim with no check. A deployment that set the default to `v2`
without configuring S2 therefore stamped runs `v2`, nothing failed at
trigger time, and every later read or write against those runs' streams
threw `Realtime streams v2 is required for this run but S2 configuration
is missing` for the life of the run.

Both paths now resolve through one pure function that takes its
configuration rather than reading `env`:

```ts
const requested = streamVersion ?? config.defaultVersion;
if (requested !== "v2") return "v1";

const hasCredentials = Boolean(config.accessToken) || config.skipAccessTokens;
return hasCredentials && Boolean(config.basin) ? "v2" : "v1";
```

## The basin requirement

`resolveStreamBasin` resolves run, session and organization basins ahead
of the global setting, so a deployment that provisions a basin per
organization can serve v2 with no global basin at all. Gating purely on
the global setting would degrade every run there to `v1`.

`determineRealtimeStreamsVersion` therefore takes an optional
organization basin, and every caller that holds one passes it, including
the session path:

```ts
basin: organizationBasinName ?? env.REALTIME_STREAMS_S2_BASIN,
```

This is deliberately the resolved basin and not the
`REALTIME_STREAMS_PER_ORG_BASINS_ENABLED` flag. The flag says the
feature is on, not that a given organization has been provisioned, and
provisioning happens out of band. Keying off the flag would stamp `v2`
on runs for unprovisioned organizations, recreating the failure this
removes.

**This widens behaviour for explicit `v2` requests**, which previously
required the global basin: a provisioned organization on a per-org
deployment now resolves `v2` where it used to get `v1`. That is
intentional, and it makes every path agree.

## Scope

Only newly created runs change. A run already stamped `v1` keeps that
version for its lifetime by design, since readers resolve the backend
from the same column and its existing streams have to stay readable.
Scheduled runs reach the same column default through
`scheduleEngine.server.ts` and are deliberately left alone: that one is
a policy question about `REALTIME_STREAMS_DEFAULT_VERSION` rather than
an inconsistency inside a single feature.

## Verification

A full-stack e2e boots the real webapp plus Postgres, Redis and s2-lite,
creates a Session through the public API so the run comes from the real
trigger path, appends records the way `streams.append()` does, and
asserts three things at once: the version stamped on the run, that the
payload is readable from S2, and that no key exists in Redis. It appends
at a realistic record size so the route's body cap and S2's per-record
cap are both exercised. Reverting the session-path change flips all
three observations, so it fails against the old behaviour rather than
passing vacuously.

Unit tests cover the resolver matrix, including organization-basin-only
and credential-only configurations; two of them fail against the
previous resolver.

Also verified by hand against a local stack: a real `chat.agent` session
run writing 8 records of 250KB through `streams.append()` put 2,049,072
bytes into S2 with no Redis key, while the same agent with the
session-path change removed put 2,102,360 bytes into Redis and nothing
into S2.

.server-changes/streams-version-s2-guard.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Realtime streams written inside a chat session run now use the same backend as the session itself, and runs are no longer created against a backend that cannot serve them.

apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts

@@ -97,7 +97,8 @@ const { action } = createActionApiRoute(
         traceContext,
         spanParentAsLink: spanParentAsLink === 1,
         realtimeStreamsVersion: determineRealtimeStreamsVersion(
-          realtimeStreamsVersion ?? undefined
+          realtimeStreamsVersion ?? undefined,
+          authentication.environment.organization.streamBasinName
         ),
       });
 

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -144,7 +144,8 @@ const { action, loader } = createActionApiRoute(
           spanParentAsLink: spanParentAsLink === 1,
           oneTimeUseToken,
           realtimeStreamsVersion: determineRealtimeStreamsVersion(
-            realtimeStreamsVersion ?? undefined
+            realtimeStreamsVersion ?? undefined,
+            authentication.environment.organization.streamBasinName
           ),
           triggerSource: isFromWorker
             ? "sdk"

apps/webapp/app/routes/api.v1.tasks.batch.ts

@@ -116,7 +116,8 @@ const { action, loader } = createActionApiRoute(
         spanParentAsLink: spanParentAsLink === 1,
         oneTimeUseToken,
         realtimeStreamsVersion: determineRealtimeStreamsVersion(
-          realtimeStreamsVersion ?? undefined
+          realtimeStreamsVersion ?? undefined,
+          authentication.environment.organization.streamBasinName
         ),
         triggerSource: isFromWorker ? "sdk" : (sanitizeTriggerSource(triggerSourceHeader) ?? "api"),
         triggerAction: "trigger",

apps/webapp/app/routes/api.v2.tasks.batch.ts

@@ -143,7 +143,8 @@ const { action, loader } = createActionApiRoute(
         spanParentAsLink: spanParentAsLink === 1,
         oneTimeUseToken,
         realtimeStreamsVersion: determineRealtimeStreamsVersion(
-          realtimeStreamsVersion ?? undefined
+          realtimeStreamsVersion ?? undefined,
+          authentication.environment.organization.streamBasinName
         ),
         triggerSource: isFromWorker ? "sdk" : (sanitizeTriggerSource(triggerSourceHeader) ?? "api"),
         triggerAction: "trigger",

apps/webapp/app/routes/api.v3.batches.ts

@@ -167,7 +167,8 @@ const { action, loader } = createActionApiRoute(
         spanParentAsLink: spanParentAsLink === 1,
         oneTimeUseToken,
         realtimeStreamsVersion: determineRealtimeStreamsVersion(
-          realtimeStreamsVersion ?? undefined
+          realtimeStreamsVersion ?? undefined,
+          authentication.environment.organization.streamBasinName
         ),
         triggerSource: isFromWorker ? "sdk" : (sanitizeTriggerSource(triggerSourceHeader) ?? "api"),
       });

apps/webapp/app/services/realtime/realtimeStreamsVersion.ts

@@ -0,0 +1,44 @@
+/**
+ * Pure realtime-streams version resolution. Deliberately free of `env` and of
+ * any module-scope singletons so it can be tested with injected values, the
+ * same split as `nativeRealtimeClient` and `nativeRealtimeClientInstance`.
+ * The env-bound wrapper is `determineRealtimeStreamsVersion` in
+ * `v1StreamsGlobal.server.ts`.
+ */
+
+export type RealtimeStreamsVersionConfig = {
+  defaultVersion: "v1" | "v2";
+  /** A basin that will actually resolve at read/write time, or undefined if none will. */
+  basin?: string;
+  accessToken?: string;
+  skipAccessTokens: boolean;
+};
+
+/**
+ * Resolve the streams version to stamp on a run, falling back to the
+ * deployment default when the caller expresses no preference.
+ *
+ * v2 is only ever returned when S2 can actually serve it. A run stamped v2 on a
+ * deployment without S2 is unusable: `getRealtimeStreamInstance` throws for the
+ * life of the run, and no read or write against its streams can succeed. v1 is
+ * a working backend, so an unsatisfiable v2 degrades to it.
+ *
+ * The basin must be one that will actually resolve later. Enabling per-org
+ * basins is not enough on its own: provisioning is out of band, so an
+ * unprovisioned organization has no basin and a global setting may not exist
+ * to fall back to.
+ */
+export function resolveRealtimeStreamsVersion(
+  streamVersion: string | undefined,
+  config: RealtimeStreamsVersionConfig
+): "v1" | "v2" {
+  const requested = streamVersion ?? config.defaultVersion;
+
+  if (requested !== "v2") {
+    return "v1";
+  }
+
+  const hasCredentials = Boolean(config.accessToken) || config.skipAccessTokens;
+
+  return hasCredentials && Boolean(config.basin) ? "v2" : "v1";
+}

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

@@ -8,6 +8,7 @@ import { logger } from "~/services/logger.server";
 import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server";
 import { TriggerTaskService } from "~/v3/services/triggerTask.server";
 import { isFinalRunStatus } from "~/v3/taskStatus";
+import { determineRealtimeStreamsVersion } from "./v1StreamsGlobal.server";
 
 /**
  * Schema for `Session.triggerConfig` (stored as JSONB). The wire-format
@@ -275,6 +276,12 @@ export async function ensureRunForSession(
  * Trigger a single run for a session. Builds `TriggerTaskRequestBody`
  * by shallow-merging `payloadOverrides` over `config.basePayload` and
  * threading `config`'s machine/queue/tags through the trigger options.
+ *
+ * A session's own channels are always v2, so the run is stamped to match
+ * rather than inheriting the `realtimeStreamsVersion` column default. Without
+ * this, run-scoped `streams.*` calls inside a session run resolve to v1 while
+ * the session it belongs to is on v2. `determineRealtimeStreamsVersion`
+ * degrades to v1 where v2 streams are not configured.
  */
 async function triggerSessionRun(params: {
   session: Pick<Session, "id" | "taskIdentifier">;
@@ -310,6 +317,10 @@ async function triggerSessionRun(params: {
   const result = await service.call(session.taskIdentifier, environment, body, {
     triggerSource: "session",
     triggerAction: "trigger",
+    realtimeStreamsVersion: determineRealtimeStreamsVersion(
+      "v2",
+      environment.organization.streamBasinName
+    ),
   });
 
   if (!result) {

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

@@ -10,6 +10,10 @@ import { singleton } from "~/utils/singleton";
 import type { AuthenticatedEnvironment } from "../apiAuth.server";
 import { RedisRealtimeStreams } from "./redisRealtimeStreams.server";
 import { S2RealtimeStreams } from "./s2realtimeStreams.server";
+import {
+  resolveRealtimeStreamsVersion,
+  type RealtimeStreamsVersionConfig,
+} from "./realtimeStreamsVersion";
 import type { StreamIngestor, StreamResponder } from "./types";
 
 function initializeRedisRealtimeStreams() {
@@ -96,20 +100,24 @@ function streamPrefixFor(environment: AuthenticatedEnvironment, basin: string):
   return segments.join("/");
 }
 
-export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
-  if (!streamVersion) {
-    return env.REALTIME_STREAMS_DEFAULT_VERSION;
-  }
+export type { RealtimeStreamsVersionConfig };
 
-  if (
-    streamVersion === "v2" &&
-    env.REALTIME_STREAMS_S2_BASIN &&
-    (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
-  ) {
-    return "v2";
-  }
-
-  return "v1";
+/**
+ * Pass `organizationBasinName` wherever the caller has it. It mirrors the
+ * organization step of {@link resolveStreamBasin}, and is what lets a
+ * per-org-basin deployment with no global setting resolve v2 for a
+ * provisioned organization while an unprovisioned one still degrades to v1.
+ */
+export function determineRealtimeStreamsVersion(
+  streamVersion?: string,
+  organizationBasinName?: string | null
+): "v1" | "v2" {
+  return resolveRealtimeStreamsVersion(streamVersion, {
+    defaultVersion: env.REALTIME_STREAMS_DEFAULT_VERSION,
+    basin: organizationBasinName ?? env.REALTIME_STREAMS_S2_BASIN,
+    accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN,
+    skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true",
+  });
 }
 
 const s2RealtimeStreamsCache = singleton(

apps/webapp/app/v3/services/replayTaskRun.server.ts

@@ -163,7 +163,8 @@ export class ReplayTaskRunService extends BaseService {
             traceparent: `00-${existingTaskRun.traceId}-${existingTaskRun.spanId}-01`,
           },
           realtimeStreamsVersion: determineRealtimeStreamsVersion(
-            existingTaskRun.realtimeStreamsVersion
+            existingTaskRun.realtimeStreamsVersion,
+            authenticatedEnvironment.organization.streamBasinName
           ),
           triggerSource: overrideOptions.triggerSource ?? "api",
           triggerAction: "replay",

apps/webapp/test/realtimeServices.replicaLag.test.ts

@@ -39,6 +39,15 @@ const replicaHolder = vi.hoisted(() => ({ client: undefined as any }));
 const storeHolder = vi.hoisted(() => ({ store: undefined as any }));
 // Records every TriggerTaskService.call so read 3 can assert NO double-trigger and read 4 can assert
 // which previousRunId the resolveRunFriendlyId fallback forwarded.
+const versionCalls = vi.hoisted(() => [] as Array<{ requested?: string; basin?: string | null }>);
+
+vi.mock("~/services/realtime/v1StreamsGlobal.server", () => ({
+  determineRealtimeStreamsVersion: (requested?: string, basin?: string | null) => {
+    versionCalls.push({ requested, basin });
+    return "v2";
+  },
+}));
+
 const triggerState = vi.hoisted(() => ({
   calls: [] as Array<{ taskIdentifier: string; body: any; options: any }>,
   result: { run: { id: "", friendlyId: "" } } as { run: { id: string; friendlyId: string } },
@@ -331,7 +340,10 @@ describe("realtime-svc — replica-lag guards", () => {
 
       const result = await ensureRunForSession({
         session,
-        environment: { id: seed.environment.id } as unknown as AuthenticatedEnvironment,
+        environment: {
+          id: seed.environment.id,
+          organization: { streamBasinName: null },
+        } as unknown as AuthenticatedEnvironment,
         reason: "manual",
       });
 
@@ -386,6 +398,7 @@ describe("realtime-svc — replica-lag guards", () => {
           triggerConfig: { basePayload: {} },
           currentRunId: callingRunId,
           currentRunVersion: 0,
+          streamBasinName: "session-pinned-basin",
         },
       });
 
@@ -394,6 +407,7 @@ describe("realtime-svc — replica-lag guards", () => {
       replicaHolder.client = replica.client;
       storeHolder.store = writerStore;
       triggerState.calls.length = 0;
+      versionCalls.length = 0;
       const newRunId = cuidRunId(`sn${seq}`);
       const newFriendlyId = `run_${suffix}_new`;
       triggerState.result = { run: { id: newRunId, friendlyId: newFriendlyId } };
@@ -401,7 +415,10 @@ describe("realtime-svc — replica-lag guards", () => {
       const result = await swapSessionRun({
         session: sessionRow,
         callingRunId,
-        environment: { id: seed.environment.id } as unknown as AuthenticatedEnvironment,
+        environment: {
+          id: seed.environment.id,
+          organization: { streamBasinName: null },
+        } as unknown as AuthenticatedEnvironment,
         reason: "upgrade",
       });
 
@@ -412,6 +429,7 @@ describe("realtime-svc — replica-lag guards", () => {
       // previousRunId forwarded to the triggered run is the calling run's cuid (documented fallback).
       expect(triggerState.calls).toHaveLength(1);
       expect(triggerState.calls[0]!.body.payload.previousRunId).toBe(callingRunId);
+      expect(versionCalls.at(-1)).toEqual({ requested: "v2", basin: null });
       expect(replica.wasHit("taskRun")).toBe(true);
 
       // Proof the null was lag-induced: the primary holds the resolvable friendlyId (≠ the cuid).

apps/webapp/test/realtimeStreamsVersion.test.ts

@@ -0,0 +1,116 @@
+import { describe, expect, it } from "vitest";
+import {
+  resolveRealtimeStreamsVersion,
+  type RealtimeStreamsVersionConfig,
+} from "~/services/realtime/realtimeStreamsVersion";
+
+const NO_S2: RealtimeStreamsVersionConfig = {
+  defaultVersion: "v1",
+  basin: undefined,
+  accessToken: undefined,
+  skipAccessTokens: false,
+};
+
+const GLOBAL_BASIN: RealtimeStreamsVersionConfig = {
+  ...NO_S2,
+  basin: "a-basin",
+  accessToken: "a-token",
+};
+
+const ORG_BASIN: RealtimeStreamsVersionConfig = {
+  ...NO_S2,
+  basin: "an-org-basin",
+  accessToken: "a-token",
+};
+
+describe("resolveRealtimeStreamsVersion", () => {
+  it("honours an explicit v2 when a global basin is configured", () => {
+    expect(resolveRealtimeStreamsVersion("v2", GLOBAL_BASIN)).toBe("v2");
+  });
+
+  it("honours an explicit v2 when only an org basin is resolvable", () => {
+    expect(resolveRealtimeStreamsVersion("v2", ORG_BASIN)).toBe("v2");
+  });
+
+  it("accepts a skip-tokens deployment as credentialed", () => {
+    expect(
+      resolveRealtimeStreamsVersion("v2", {
+        ...NO_S2,
+        basin: "a-basin",
+        skipAccessTokens: true,
+      })
+    ).toBe("v2");
+  });
+
+  it("degrades an explicit v2 to v1 when S2 is not configured", () => {
+    expect(resolveRealtimeStreamsVersion("v2", NO_S2)).toBe("v1");
+  });
+
+  it("falls back to the default version when the caller expresses no preference", () => {
+    expect(
+      resolveRealtimeStreamsVersion(undefined, { ...GLOBAL_BASIN, defaultVersion: "v2" })
+    ).toBe("v2");
+  });
+
+  it("degrades a v2 default to v1 when S2 is not configured", () => {
+    expect(resolveRealtimeStreamsVersion(undefined, { ...NO_S2, defaultVersion: "v2" })).toBe("v1");
+  });
+
+  it("keeps a v2 default on v2 when only an org basin is resolvable", () => {
+    expect(resolveRealtimeStreamsVersion(undefined, { ...ORG_BASIN, defaultVersion: "v2" })).toBe(
+      "v2"
+    );
+  });
+
+  it("requires credentials, not just a basin", () => {
+    const basinOnly = { ...NO_S2, basin: "a-basin", defaultVersion: "v2" as const };
+    expect(resolveRealtimeStreamsVersion(undefined, basinOnly)).toBe("v1");
+    expect(resolveRealtimeStreamsVersion("v2", basinOnly)).toBe("v1");
+  });
+
+  it("requires a basin, not just credentials", () => {
+    const tokenOnly = { ...NO_S2, accessToken: "a-token", defaultVersion: "v2" as const };
+    expect(resolveRealtimeStreamsVersion(undefined, tokenOnly)).toBe("v1");
+    expect(resolveRealtimeStreamsVersion("v2", tokenOnly)).toBe("v1");
+  });
+
+  it("keeps an explicit v1 on v1 even where S2 is available", () => {
+    expect(resolveRealtimeStreamsVersion("v1", { ...GLOBAL_BASIN, defaultVersion: "v2" })).toBe(
+      "v1"
+    );
+  });
+
+  it("treats an unrecognised version as v1", () => {
+    expect(resolveRealtimeStreamsVersion("v3", GLOBAL_BASIN)).toBe("v1");
+  });
+});
+
+describe("resolveRealtimeStreamsVersion invariant", () => {
+  const BASINS = [undefined, "", "a-basin"];
+  const TOKENS = [undefined, "a-token"];
+  const SKIPS = [false, true];
+  const DEFAULTS: Array<"v1" | "v2"> = ["v1", "v2"];
+  const REQUESTED = [undefined, "v1", "v2", "v3"];
+
+  it("only returns v2 when a basin and credentials are both present, for every configuration", () => {
+    const counterexamples: string[] = [];
+
+    for (const basin of BASINS) {
+      for (const accessToken of TOKENS) {
+        for (const skipAccessTokens of SKIPS) {
+          for (const defaultVersion of DEFAULTS) {
+            for (const requested of REQUESTED) {
+              const config = { defaultVersion, basin, accessToken, skipAccessTokens };
+              const usable = Boolean(basin) && (Boolean(accessToken) || skipAccessTokens);
+              if (resolveRealtimeStreamsVersion(requested, config) === "v2" && !usable) {
+                counterexamples.push(JSON.stringify({ requested, ...config }));
+              }
+            }
+          }
+        }
+      }
+    }
+
+    expect(counterexamples).toEqual([]);
+  });
+});

apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts

@@ -0,0 +1,147 @@
+/**
+ * Full-stack e2e for which realtime streams backend a Session's run lands on.
+ *
+ * Boots the real webapp + Postgres + Redis + s2-lite (via
+ * startSessionStreamTestServer), creates a Session through the public API so
+ * the run is triggered by the real `sessionRunManager` path, then appends to a
+ * run-scoped stream exactly as `streams.append()` does and checks where the
+ * bytes actually went.
+ *
+ * The harness starts the webapp with `REALTIME_STREAMS_DEFAULT_VERSION: "v2"`
+ * and a live S2, so a run landing on v1 here is not a configuration gap. It
+ * means the trigger path never asked, and fell through to the
+ * `realtimeStreamsVersion` column default.
+ *
+ * Requires a pre-built webapp: pnpm run build --filter webapp
+ */
+import { randomBytes } from "crypto";
+import Redis from "ioredis";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import type { SessionStreamTestServer } from "@internal/testcontainers/webapp";
+import { startSessionStreamTestServer } from "@internal/testcontainers/webapp";
+import { seedTestEnvironment } from "./helpers/seedTestEnvironment";
+
+vi.setConfig({ testTimeout: 120_000, hookTimeout: 180_000 });
+
+let server: SessionStreamTestServer;
+
+beforeAll(async () => {
+  server = await startSessionStreamTestServer();
+}, 180_000);
+
+afterAll(async () => {
+  await server?.stop();
+}, 120_000);
+
+const STREAM_ID = "frames";
+const PART_ID = "part";
+const FRAME_BYTES = 250 * 1024;
+const FRAME_COUNT = 8;
+
+/** Mirrors `S2RealtimeStreams.toStreamName` on the shared-basin prefix. */
+function runStreamName(p: {
+  orgId: string;
+  envSlug: string;
+  envId: string;
+  runId: string;
+  streamId: string;
+}): string {
+  return `org/${p.orgId}/env/${p.envSlug}/${p.envId}/runs/${p.runId}/${p.streamId}`;
+}
+
+/** Mirrors the `keyPrefix` + key shape in `v1StreamsGlobal` / `RedisRealtimeStreams`. */
+function redisStreamKey(runId: string, streamId: string): string {
+  return `tr:realtime:streams:stream:${runId}:${streamId}`;
+}
+
+function framesFound(body: string): number {
+  return Array.from({ length: FRAME_COUNT }, (_, i) => `${PART_ID}-${i}`).filter((id) =>
+    body.includes(id)
+  ).length;
+}
+
+async function s2Body(streamName: string): Promise<string> {
+  const qs = new URLSearchParams({ seq_num: "0", clamp: "true", wait: "0" });
+  const res = await fetch(
+    `${server.s2.endpoint}/v1/streams/${encodeURIComponent(streamName)}/records?${qs}`,
+    {
+      headers: {
+        Authorization: "Bearer ignored",
+        Accept: "text/event-stream",
+        "S2-Format": "raw",
+        "S2-Basin": server.s2.basin,
+      },
+    }
+  );
+
+  if (res.status === 404) return "";
+  expect(res.ok).toBe(true);
+
+  return res.text();
+}
+
+describe("session runs and the realtime streams backend", () => {
+  it("stamps the run v2 and routes a run-scoped stream to S2, not Redis", async () => {
+    const { organization, environment, apiKey } = await seedTestEnvironment(server.prisma);
+
+    const createRes = await fetch(`${server.webapp.baseUrl}/api/v1/sessions`, {
+      method: "POST",
+      headers: { Authorization: `Bearer ${apiKey}`, "Content-Type": "application/json" },
+      body: JSON.stringify({
+        type: "chat.agent",
+        externalId: `e2e-${randomBytes(6).toString("hex")}`,
+        taskIdentifier: "e2e-browser-agent",
+        triggerConfig: { basePayload: {} },
+      }),
+    });
+
+    expect(createRes.ok).toBe(true);
+    const created = (await createRes.json()) as { runId: string };
+    expect(created.runId).toBeTruthy();
+
+    const run = await server.prisma.taskRun.findFirstOrThrow({
+      where: { friendlyId: created.runId },
+      select: { realtimeStreamsVersion: true },
+    });
+
+    const appendStatuses: number[] = [];
+    for (let i = 0; i < FRAME_COUNT; i++) {
+      const res = await fetch(
+        `${server.webapp.baseUrl}/realtime/v1/streams/${created.runId}/self/${STREAM_ID}/append`,
+        {
+          method: "POST",
+          headers: {
+            Authorization: `Bearer ${apiKey}`,
+            "Content-Type": "text/plain",
+            "X-Part-Id": `${PART_ID}-${i}`,
+          },
+          body: JSON.stringify({ i, frame: "a".repeat(FRAME_BYTES) }),
+        }
+      );
+      appendStatuses.push(res.status);
+    }
+
+    expect(appendStatuses).toEqual(Array.from({ length: FRAME_COUNT }, () => 200));
+
+    const streamName = runStreamName({
+      orgId: organization.id,
+      envSlug: environment.slug,
+      envId: environment.id,
+      runId: created.runId,
+      streamId: STREAM_ID,
+    });
+    const redis = new Redis({ host: server.redis.host, port: server.redis.port });
+    let observed: { version: string; framesInS2: number; keyInRedis: boolean };
+    try {
+      observed = {
+        version: run.realtimeStreamsVersion,
+        framesInS2: framesFound(await s2Body(streamName)),
+        keyInRedis: (await redis.exists(redisStreamKey(created.runId, STREAM_ID))) === 1,
+      };
+    } finally {
+      redis.disconnect();
+    }
+
+    expect(observed).toEqual({ version: "v2", framesInS2: FRAME_COUNT, keyInRedis: false });
+  });
+});

apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts

@@ -0,0 +1,177 @@
+/**
+ * Full-stack e2e for the per-org-basin configuration: S2 credentials present,
+ * no global basin, so whether a run can use v2 depends entirely on whether its
+ * organization has been provisioned one.
+ *
+ * The sibling `sessionRunStreamsBackend` e2e runs with a global basin set,
+ * which makes every basin value work and hides this whole class of bug. Here a
+ * run stamped v2 without a resolvable basin is not a degraded experience, it
+ * throws on every stream operation for the life of the run, so both directions
+ * are asserted: a provisioned organization reaches S2, and an unprovisioned one
+ * degrades to v1 and keeps working on Redis.
+ *
+ * Scope: both cases assert run-scoped streams only. Neither drives a session
+ * channel, so neither says anything about `.in`/`.out`. That matters for the
+ * unprovisioned case, where the session's own channels cannot resolve a basin
+ * at all and fail: the run degrading to v1 is what keeps working there, not the
+ * session. Do not read these as evidence that a session is healthy.
+ *
+ * Which basin the trigger path reads is pinned separately, by the swap case in
+ * `realtimeServices.replicaLag.test.ts`, which asserts the organization's basin
+ * reaches the resolver even when the session row carries one of its own.
+ *
+ * Requires a pre-built webapp: pnpm run build --filter webapp
+ */
+import { randomBytes } from "crypto";
+import Redis from "ioredis";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import type { SessionStreamTestServer } from "@internal/testcontainers/webapp";
+import { startSessionStreamTestServer } from "@internal/testcontainers/webapp";
+import { seedTestEnvironment } from "./helpers/seedTestEnvironment";
+
+vi.setConfig({ testTimeout: 120_000, hookTimeout: 180_000 });
+
+let server: SessionStreamTestServer;
+
+beforeAll(async () => {
+  server = await startSessionStreamTestServer({
+    extraEnv: {
+      REALTIME_STREAMS_S2_BASIN: "",
+      REALTIME_STREAMS_PER_ORG_BASINS_ENABLED: "true",
+    },
+  });
+}, 180_000);
+
+afterAll(async () => {
+  await server?.stop();
+}, 120_000);
+
+const STREAM_ID = "frames";
+
+/** Per-org basins drop the `org/{id}` segment; see `streamPrefixFor`. */
+function perOrgStreamName(p: { envSlug: string; envId: string; runId: string }): string {
+  return `env/${p.envSlug}/${p.envId}/runs/${p.runId}/${STREAM_ID}`;
+}
+
+function redisStreamKey(runId: string): string {
+  return `tr:realtime:streams:stream:${runId}:${STREAM_ID}`;
+}
+
+async function s2HasRecords(basin: string, streamName: string): Promise<boolean> {
+  const qs = new URLSearchParams({ seq_num: "0", clamp: "true", wait: "0" });
+  const res = await fetch(
+    `${server.s2.endpoint}/v1/streams/${encodeURIComponent(streamName)}/records?${qs}`,
+    {
+      headers: {
+        Authorization: "Bearer ignored",
+        Accept: "text/event-stream",
+        "S2-Format": "raw",
+        "S2-Basin": basin,
+      },
+    }
+  );
+  if (!res.ok) return false;
+  return (await res.text()).includes(STREAM_ID);
+}
+
+async function createSessionRun(apiKey: string, taskIdentifier: string): Promise<string> {
+  const res = await fetch(`${server.webapp.baseUrl}/api/v1/sessions`, {
+    method: "POST",
+    headers: { Authorization: `Bearer ${apiKey}`, "Content-Type": "application/json" },
+    body: JSON.stringify({
+      type: "chat.agent",
+      externalId: `e2e-${randomBytes(6).toString("hex")}`,
+      taskIdentifier,
+      triggerConfig: { basePayload: {} },
+    }),
+  });
+  expect(res.ok).toBe(true);
+  return ((await res.json()) as { runId: string }).runId;
+}
+
+async function appendFrame(apiKey: string, runId: string): Promise<number> {
+  const res = await fetch(
+    `${server.webapp.baseUrl}/realtime/v1/streams/${runId}/self/${STREAM_ID}/append`,
+    {
+      method: "POST",
+      headers: {
+        Authorization: `Bearer ${apiKey}`,
+        "Content-Type": "text/plain",
+        "X-Part-Id": STREAM_ID,
+      },
+      body: JSON.stringify({ frame: "a".repeat(1024) }),
+    }
+  );
+  return res.status;
+}
+
+describe("session runs with per-org basins and no global basin", () => {
+  it("reaches S2 for a provisioned organization", async () => {
+    const { organization, environment, apiKey } = await seedTestEnvironment(server.prisma);
+    const basin = server.s2.basin;
+
+    await server.prisma.organization.update({
+      where: { id: organization.id },
+      data: { streamBasinName: basin },
+    });
+
+    const runId = await createSessionRun(apiKey, "e2e-per-org-provisioned");
+
+    const run = await server.prisma.taskRun.findFirstOrThrow({
+      where: { friendlyId: runId },
+      select: { realtimeStreamsVersion: true, streamBasinName: true },
+    });
+
+    expect(await appendFrame(apiKey, runId)).toBe(200);
+
+    const redis = new Redis({ host: server.redis.host, port: server.redis.port });
+    let observed;
+    try {
+      observed = {
+        version: run.realtimeStreamsVersion,
+        runBasin: run.streamBasinName,
+        inS2: await s2HasRecords(
+          basin,
+          perOrgStreamName({ envSlug: environment.slug, envId: environment.id, runId })
+        ),
+        keyInRedis: (await redis.exists(redisStreamKey(runId))) === 1,
+      };
+    } finally {
+      redis.disconnect();
+    }
+
+    expect(observed).toEqual({ version: "v2", runBasin: basin, inS2: true, keyInRedis: false });
+  });
+
+  it("degrades to v1 for an unprovisioned organization, keeping its run-scoped streams usable", async () => {
+    const { organization, apiKey } = await seedTestEnvironment(server.prisma);
+
+    await server.prisma.organization.update({
+      where: { id: organization.id },
+      data: { streamBasinName: null },
+    });
+
+    const runId = await createSessionRun(apiKey, "e2e-per-org-unprovisioned");
+
+    const run = await server.prisma.taskRun.findFirstOrThrow({
+      where: { friendlyId: runId },
+      select: { realtimeStreamsVersion: true, streamBasinName: true },
+    });
+
+    expect(await appendFrame(apiKey, runId)).toBe(200);
+
+    const redis = new Redis({ host: server.redis.host, port: server.redis.port });
+    let observed;
+    try {
+      observed = {
+        version: run.realtimeStreamsVersion,
+        runBasin: run.streamBasinName,
+        keyInRedis: (await redis.exists(redisStreamKey(runId))) === 1,
+      };
+    } finally {
+      redis.disconnect();
+    }
+
+    expect(observed).toEqual({ version: "v1", runBasin: null, keyInRedis: true });
+  });
+});

internal-packages/testcontainers/src/webapp.ts

@@ -272,6 +272,12 @@ export type { StartedS2Container } from "./s2";
 export interface SessionStreamTestServer extends TestServer {
   s2: StartedS2Container;
   minio: StartedMinIOContainer;
+  /**
+   * Mapped connection for the same Redis the webapp under test uses. Lets a
+   * test assert which backend a stream actually landed on, rather than
+   * inferring it from the absence of records in S2.
+   */
+  redis: { host: string; port: number };
 }
 
 /**
@@ -280,7 +286,9 @@ export interface SessionStreamTestServer extends TestServer {
  * process reaching every container over its mapped port, so the S2 endpoint is
  * the mapped localhost URL (the docker-network alias is unusable from the host).
  */
-export async function startSessionStreamTestServer(): Promise<SessionStreamTestServer> {
+export async function startSessionStreamTestServer(
+  options: StartWebappOptions = {}
+): Promise<SessionStreamTestServer> {
   const network = await new Network().start();
 
   let pgContainer: Awaited<ReturnType<typeof createPostgresContainer>>["container"] | undefined;
@@ -322,6 +330,7 @@ export async function startSessionStreamTestServer(): Promise<SessionStreamTestS
           OBJECT_STORE_ACCESS_KEY_ID: minioConfig.accessKeyId,
           OBJECT_STORE_SECRET_ACCESS_KEY: minioConfig.secretAccessKey,
           OBJECT_STORE_REGION: minioConfig.region,
+          ...(options.extraEnv ?? {}),
         },
       }
     );
@@ -348,5 +357,13 @@ export async function startSessionStreamTestServer(): Promise<SessionStreamTestS
     await network.stop().catch((err) => console.error("network.stop failed:", err));
   };
 
-  return { webapp, prisma: prisma!, databaseUrl: pgUrl!, s2: s2!, minio: minio!, stop };
+  return {
+    webapp,
+    prisma: prisma!,
+    databaseUrl: pgUrl!,
+    s2: s2!,
+    minio: minio!,
+    redis: { host: redisContainer!.getHost(), port: redisContainer!.getPort() },
+    stop,
+  };
 }