From 2eb19be5fc096dcce823c179d1f93d03b4d49a5f Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 11:01:10 +0100
Subject: [PATCH 01/11] fix(webapp): never resolve realtime streams v2 without
 S2 configured

The default-version branch returned `REALTIME_STREAMS_DEFAULT_VERSION` verbatim
while the explicit branch checked that a basin and credentials were present. A
deployment that set the default to v2 without configuring S2 therefore stamped
runs v2, and every read and write against those runs' streams threw for the
life of the run.

Both paths now go through the same check, and an unsatisfiable v2 degrades to
v1, which is a working backend.
---
 .server-changes/streams-version-s2-guard.md   |  6 ++
 .../realtime/v1StreamsGlobal.server.ts        | 15 ++--
 .../determineRealtimeStreamsVersion.test.ts   | 70 +++++++++++++++++++
 3 files changed, 87 insertions(+), 4 deletions(-)
 create mode 100644 .server-changes/streams-version-s2-guard.md
 create mode 100644 apps/webapp/test/determineRealtimeStreamsVersion.test.ts

diff --git a/.server-changes/streams-version-s2-guard.md b/.server-changes/streams-version-s2-guard.md
new file mode 100644
index 0000000000..2c607652e4
--- /dev/null
+++ b/.server-changes/streams-version-s2-guard.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Stop creating runs against a realtime streams backend the deployment cannot serve.
diff --git a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
index a6801b8a1b..8bd1f9d1a5 100644
--- a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
+++ b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
@@ -96,13 +96,20 @@ function streamPrefixFor(environment: AuthenticatedEnvironment, basin: string):
   return segments.join("/");
 }
 
+/**
+ * Resolve the streams version to stamp on a run, falling back to
+ * `REALTIME_STREAMS_DEFAULT_VERSION` when the caller expresses no preference.
+ *
+ * v2 is only ever returned when S2 is actually configured. 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.
+ */
 export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
-  if (!streamVersion) {
-    return env.REALTIME_STREAMS_DEFAULT_VERSION;
-  }
+  const requested = streamVersion ?? env.REALTIME_STREAMS_DEFAULT_VERSION;
 
   if (
-    streamVersion === "v2" &&
+    requested === "v2" &&
     env.REALTIME_STREAMS_S2_BASIN &&
     (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
   ) {
diff --git a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
new file mode 100644
index 0000000000..6d73070856
--- /dev/null
+++ b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
@@ -0,0 +1,70 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const envMock = vi.hoisted(() => ({
+  REALTIME_STREAMS_DEFAULT_VERSION: "v1" as "v1" | "v2",
+  REALTIME_STREAMS_S2_BASIN: undefined as string | undefined,
+  REALTIME_STREAMS_S2_ACCESS_TOKEN: undefined as string | undefined,
+  REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS: "false",
+}));
+
+vi.mock("~/env.server", () => ({ env: envMock }));
+
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
+
+function configureS2() {
+  envMock.REALTIME_STREAMS_S2_BASIN = "a-basin";
+  envMock.REALTIME_STREAMS_S2_ACCESS_TOKEN = "a-token";
+}
+
+beforeEach(() => {
+  envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v1";
+  envMock.REALTIME_STREAMS_S2_BASIN = undefined;
+  envMock.REALTIME_STREAMS_S2_ACCESS_TOKEN = undefined;
+  envMock.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS = "false";
+});
+
+describe("determineRealtimeStreamsVersion", () => {
+  it("honours an explicit v2 when S2 is configured", () => {
+    configureS2();
+    expect(determineRealtimeStreamsVersion("v2")).toBe("v2");
+  });
+
+  it("accepts a skip-tokens deployment as configured", () => {
+    envMock.REALTIME_STREAMS_S2_BASIN = "a-basin";
+    envMock.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS = "true";
+    expect(determineRealtimeStreamsVersion("v2")).toBe("v2");
+  });
+
+  it("degrades an explicit v2 to v1 when S2 is not configured", () => {
+    expect(determineRealtimeStreamsVersion("v2")).toBe("v1");
+  });
+
+  it("falls back to the default version when the caller expresses no preference", () => {
+    configureS2();
+    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
+    expect(determineRealtimeStreamsVersion()).toBe("v2");
+  });
+
+  it("degrades a v2 default to v1 when S2 is not configured", () => {
+    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
+    expect(determineRealtimeStreamsVersion()).toBe("v1");
+  });
+
+  it("requires a basin, not just a token", () => {
+    envMock.REALTIME_STREAMS_S2_ACCESS_TOKEN = "a-token";
+    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
+    expect(determineRealtimeStreamsVersion()).toBe("v1");
+    expect(determineRealtimeStreamsVersion("v2")).toBe("v1");
+  });
+
+  it("keeps an explicit v1 on v1 even where S2 is available", () => {
+    configureS2();
+    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
+    expect(determineRealtimeStreamsVersion("v1")).toBe("v1");
+  });
+
+  it("treats an unrecognised version as v1", () => {
+    configureS2();
+    expect(determineRealtimeStreamsVersion("v3")).toBe("v1");
+  });
+});

From dab295751b0c8543dbd3ee636d1395c9c28de06d Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 11:15:28 +0100
Subject: [PATCH 02/11] fix(webapp): count per-org basins as S2 being
 configured

Gating v2 on the global basin alone would have degraded every run to v1 on a
deployment that provisions a basin per organization and sets no global one,
even though S2 is fully working there. `resolveStreamBasin` already resolves
run, session and organization basins ahead of the global setting, so either
source now satisfies the basin requirement.

Splits the pure resolver out from the env lookup so the version matrix can be
tested without reaching for `env.server`.
---
 .server-changes/streams-version-s2-guard.md   |   2 +-
 .../realtime/v1StreamsGlobal.server.ts        |  54 ++++++---
 .../determineRealtimeStreamsVersion.test.ts   | 105 ++++++++++--------
 3 files changed, 101 insertions(+), 60 deletions(-)

diff --git a/.server-changes/streams-version-s2-guard.md b/.server-changes/streams-version-s2-guard.md
index 2c607652e4..6114ee44eb 100644
--- a/.server-changes/streams-version-s2-guard.md
+++ b/.server-changes/streams-version-s2-guard.md
@@ -3,4 +3,4 @@ area: webapp
 type: fix
 ---
 
-Stop creating runs against a realtime streams backend the deployment cannot serve.
+Runs no longer end up with realtime streams that cannot be read or written.
diff --git a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
index 8bd1f9d1a5..435198e32a 100644
--- a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
+++ b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
@@ -96,27 +96,51 @@ function streamPrefixFor(environment: AuthenticatedEnvironment, basin: string):
   return segments.join("/");
 }
 
+export type RealtimeStreamsVersionConfig = {
+  defaultVersion: "v1" | "v2";
+  basin?: string;
+  accessToken?: string;
+  skipAccessTokens: boolean;
+  perOrgBasinsEnabled: boolean;
+};
+
 /**
- * Resolve the streams version to stamp on a run, falling back to
- * `REALTIME_STREAMS_DEFAULT_VERSION` when the caller expresses no preference.
+ * 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.
  *
- * v2 is only ever returned when S2 is actually configured. 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.
+ * A basin can come from the global setting or from per-org provisioning, so
+ * either satisfies the basin requirement. This mirrors {@link resolveStreamBasin},
+ * which resolves run, session and organization basins ahead of the global one.
  */
-export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
-  const requested = streamVersion ?? env.REALTIME_STREAMS_DEFAULT_VERSION;
+export function resolveRealtimeStreamsVersion(
+  streamVersion: string | undefined,
+  config: RealtimeStreamsVersionConfig
+): "v1" | "v2" {
+  const requested = streamVersion ?? config.defaultVersion;
 
-  if (
-    requested === "v2" &&
-    env.REALTIME_STREAMS_S2_BASIN &&
-    (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
-  ) {
-    return "v2";
+  if (requested !== "v2") {
+    return "v1";
   }
 
-  return "v1";
+  const hasCredentials = Boolean(config.accessToken) || config.skipAccessTokens;
+  const hasBasin = Boolean(config.basin) || config.perOrgBasinsEnabled;
+
+  return hasCredentials && hasBasin ? "v2" : "v1";
+}
+
+export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
+  return resolveRealtimeStreamsVersion(streamVersion, {
+    defaultVersion: env.REALTIME_STREAMS_DEFAULT_VERSION,
+    basin: env.REALTIME_STREAMS_S2_BASIN,
+    accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN,
+    skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true",
+    perOrgBasinsEnabled: env.REALTIME_STREAMS_PER_ORG_BASINS_ENABLED === "true",
+  });
 }
 
 const s2RealtimeStreamsCache = singleton(
diff --git a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
index 6d73070856..ec5b78a978 100644
--- a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
+++ b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
@@ -1,70 +1,87 @@
-import { beforeEach, describe, expect, it, vi } from "vitest";
+import { describe, expect, it } from "vitest";
+import {
+  resolveRealtimeStreamsVersion,
+  type RealtimeStreamsVersionConfig,
+} from "~/services/realtime/v1StreamsGlobal.server";
 
-const envMock = vi.hoisted(() => ({
-  REALTIME_STREAMS_DEFAULT_VERSION: "v1" as "v1" | "v2",
-  REALTIME_STREAMS_S2_BASIN: undefined as string | undefined,
-  REALTIME_STREAMS_S2_ACCESS_TOKEN: undefined as string | undefined,
-  REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS: "false",
-}));
+const NO_S2: RealtimeStreamsVersionConfig = {
+  defaultVersion: "v1",
+  basin: undefined,
+  accessToken: undefined,
+  skipAccessTokens: false,
+  perOrgBasinsEnabled: false,
+};
 
-vi.mock("~/env.server", () => ({ env: envMock }));
+const GLOBAL_BASIN: RealtimeStreamsVersionConfig = {
+  ...NO_S2,
+  basin: "a-basin",
+  accessToken: "a-token",
+};
 
-import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
+const PER_ORG_BASINS: RealtimeStreamsVersionConfig = {
+  ...NO_S2,
+  accessToken: "a-token",
+  perOrgBasinsEnabled: true,
+};
 
-function configureS2() {
-  envMock.REALTIME_STREAMS_S2_BASIN = "a-basin";
-  envMock.REALTIME_STREAMS_S2_ACCESS_TOKEN = "a-token";
-}
-
-beforeEach(() => {
-  envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v1";
-  envMock.REALTIME_STREAMS_S2_BASIN = undefined;
-  envMock.REALTIME_STREAMS_S2_ACCESS_TOKEN = undefined;
-  envMock.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS = "false";
-});
+describe("resolveRealtimeStreamsVersion", () => {
+  it("honours an explicit v2 when a global basin is configured", () => {
+    expect(resolveRealtimeStreamsVersion("v2", GLOBAL_BASIN)).toBe("v2");
+  });
 
-describe("determineRealtimeStreamsVersion", () => {
-  it("honours an explicit v2 when S2 is configured", () => {
-    configureS2();
-    expect(determineRealtimeStreamsVersion("v2")).toBe("v2");
+  it("honours an explicit v2 when only per-org basins are configured", () => {
+    expect(resolveRealtimeStreamsVersion("v2", PER_ORG_BASINS)).toBe("v2");
   });
 
-  it("accepts a skip-tokens deployment as configured", () => {
-    envMock.REALTIME_STREAMS_S2_BASIN = "a-basin";
-    envMock.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS = "true";
-    expect(determineRealtimeStreamsVersion("v2")).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(determineRealtimeStreamsVersion("v2")).toBe("v1");
+    expect(resolveRealtimeStreamsVersion("v2", NO_S2)).toBe("v1");
   });
 
   it("falls back to the default version when the caller expresses no preference", () => {
-    configureS2();
-    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
-    expect(determineRealtimeStreamsVersion()).toBe("v2");
+    expect(
+      resolveRealtimeStreamsVersion(undefined, { ...GLOBAL_BASIN, defaultVersion: "v2" })
+    ).toBe("v2");
   });
 
   it("degrades a v2 default to v1 when S2 is not configured", () => {
-    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
-    expect(determineRealtimeStreamsVersion()).toBe("v1");
+    expect(resolveRealtimeStreamsVersion(undefined, { ...NO_S2, defaultVersion: "v2" })).toBe("v1");
+  });
+
+  it("keeps a v2 default on v2 when only per-org basins are configured", () => {
+    expect(
+      resolveRealtimeStreamsVersion(undefined, { ...PER_ORG_BASINS, 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 a token", () => {
-    envMock.REALTIME_STREAMS_S2_ACCESS_TOKEN = "a-token";
-    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
-    expect(determineRealtimeStreamsVersion()).toBe("v1");
-    expect(determineRealtimeStreamsVersion("v2")).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", () => {
-    configureS2();
-    envMock.REALTIME_STREAMS_DEFAULT_VERSION = "v2";
-    expect(determineRealtimeStreamsVersion("v1")).toBe("v1");
+    expect(resolveRealtimeStreamsVersion("v1", { ...GLOBAL_BASIN, defaultVersion: "v2" })).toBe(
+      "v1"
+    );
   });
 
   it("treats an unrecognised version as v1", () => {
-    configureS2();
-    expect(determineRealtimeStreamsVersion("v3")).toBe("v1");
+    expect(resolveRealtimeStreamsVersion("v3", GLOBAL_BASIN)).toBe("v1");
   });
 });

From 6780a977480e5c1a0ff086b6f60714c9b25f1d78 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 11:31:18 +0100
Subject: [PATCH 03/11] fix(webapp): require a basin that resolves, not the
 per-org flag

Treating `REALTIME_STREAMS_PER_ORG_BASINS_ENABLED` as proof of a basin was
wrong. The flag says the feature is on, not that an organization has been
provisioned, and provisioning happens out of band. An unprovisioned
organization on a deployment with no global basin would still have been stamped
v2 and thrown on every stream operation, which is the failure this is meant to
prevent.

Callers that hold the organization now pass its `streamBasinName`, mirroring
the organization step of `resolveStreamBasin`. A provisioned organization
resolves v2 with no global basin configured; an unprovisioned one degrades to
v1.
---
 .../app/routes/api.v1.tasks.$taskId.batch.ts  |  3 ++-
 .../routes/api.v1.tasks.$taskId.trigger.ts    |  3 ++-
 apps/webapp/app/routes/api.v1.tasks.batch.ts  |  3 ++-
 apps/webapp/app/routes/api.v2.tasks.batch.ts  |  3 ++-
 apps/webapp/app/routes/api.v3.batches.ts      |  3 ++-
 .../realtime/v1StreamsGlobal.server.ts        | 26 ++++++++++++-------
 .../app/v3/services/replayTaskRun.server.ts   |  3 ++-
 .../determineRealtimeStreamsVersion.test.ts   | 17 ++++++------
 8 files changed, 37 insertions(+), 24 deletions(-)

diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts
index 2bd7bc5650..904bc9d4a9 100644
--- a/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts
+++ b/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
         ),
       });
 
diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts
index 6165049330..c0de0c59f7 100644
--- a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts
+++ b/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"
diff --git a/apps/webapp/app/routes/api.v1.tasks.batch.ts b/apps/webapp/app/routes/api.v1.tasks.batch.ts
index 5c9202d6fe..2ea5ebb3b2 100644
--- a/apps/webapp/app/routes/api.v1.tasks.batch.ts
+++ b/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",
diff --git a/apps/webapp/app/routes/api.v2.tasks.batch.ts b/apps/webapp/app/routes/api.v2.tasks.batch.ts
index 5dcbf13e0f..8bdfdb569d 100644
--- a/apps/webapp/app/routes/api.v2.tasks.batch.ts
+++ b/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",
diff --git a/apps/webapp/app/routes/api.v3.batches.ts b/apps/webapp/app/routes/api.v3.batches.ts
index 071bb783b8..ce301718f0 100644
--- a/apps/webapp/app/routes/api.v3.batches.ts
+++ b/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"),
       });
diff --git a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
index 435198e32a..557a23e816 100644
--- a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
+++ b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
@@ -98,10 +98,10 @@ function streamPrefixFor(environment: AuthenticatedEnvironment, basin: string):
 
 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;
-  perOrgBasinsEnabled: boolean;
 };
 
 /**
@@ -113,9 +113,10 @@ export type RealtimeStreamsVersionConfig = {
  * 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.
  *
- * A basin can come from the global setting or from per-org provisioning, so
- * either satisfies the basin requirement. This mirrors {@link resolveStreamBasin},
- * which resolves run, session and organization basins ahead of the global one.
+ * 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,
@@ -128,18 +129,25 @@ export function resolveRealtimeStreamsVersion(
   }
 
   const hasCredentials = Boolean(config.accessToken) || config.skipAccessTokens;
-  const hasBasin = Boolean(config.basin) || config.perOrgBasinsEnabled;
 
-  return hasCredentials && hasBasin ? "v2" : "v1";
+  return hasCredentials && Boolean(config.basin) ? "v2" : "v1";
 }
 
-export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
+/**
+ * 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: env.REALTIME_STREAMS_S2_BASIN,
+    basin: organizationBasinName ?? env.REALTIME_STREAMS_S2_BASIN,
     accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN,
     skipAccessTokens: env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true",
-    perOrgBasinsEnabled: env.REALTIME_STREAMS_PER_ORG_BASINS_ENABLED === "true",
   });
 }
 
diff --git a/apps/webapp/app/v3/services/replayTaskRun.server.ts b/apps/webapp/app/v3/services/replayTaskRun.server.ts
index d626ec2ae8..750427fb32 100644
--- a/apps/webapp/app/v3/services/replayTaskRun.server.ts
+++ b/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",
diff --git a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
index ec5b78a978..9e28c12872 100644
--- a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
+++ b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
@@ -9,7 +9,6 @@ const NO_S2: RealtimeStreamsVersionConfig = {
   basin: undefined,
   accessToken: undefined,
   skipAccessTokens: false,
-  perOrgBasinsEnabled: false,
 };
 
 const GLOBAL_BASIN: RealtimeStreamsVersionConfig = {
@@ -18,10 +17,10 @@ const GLOBAL_BASIN: RealtimeStreamsVersionConfig = {
   accessToken: "a-token",
 };
 
-const PER_ORG_BASINS: RealtimeStreamsVersionConfig = {
+const ORG_BASIN: RealtimeStreamsVersionConfig = {
   ...NO_S2,
+  basin: "an-org-basin",
   accessToken: "a-token",
-  perOrgBasinsEnabled: true,
 };
 
 describe("resolveRealtimeStreamsVersion", () => {
@@ -29,8 +28,8 @@ describe("resolveRealtimeStreamsVersion", () => {
     expect(resolveRealtimeStreamsVersion("v2", GLOBAL_BASIN)).toBe("v2");
   });
 
-  it("honours an explicit v2 when only per-org basins are configured", () => {
-    expect(resolveRealtimeStreamsVersion("v2", PER_ORG_BASINS)).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", () => {
@@ -57,10 +56,10 @@ describe("resolveRealtimeStreamsVersion", () => {
     expect(resolveRealtimeStreamsVersion(undefined, { ...NO_S2, defaultVersion: "v2" })).toBe("v1");
   });
 
-  it("keeps a v2 default on v2 when only per-org basins are configured", () => {
-    expect(
-      resolveRealtimeStreamsVersion(undefined, { ...PER_ORG_BASINS, defaultVersion: "v2" })
-    ).toBe("v2");
+  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", () => {

From 1f90e9c62d94e4a26dbcd55ad255c8795ccac125 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 02:03:52 +0100
Subject: [PATCH 04/11] fix(webapp): keep session runs off the legacy realtime
 streams backend

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 channels are always v2, so run-scoped `streams.*` calls made
inside a session run resolved to a different backend than the session itself,
for the whole life of the run.

The trigger now resolves the version explicitly, degrading to v1 where v2
streams are not configured.
---
 .server-changes/session-run-streams-version.md            | 6 ++++++
 .../app/services/realtime/sessionRunManager.server.ts     | 8 ++++++++
 apps/webapp/test/realtimeServices.replicaLag.test.ts      | 1 +
 3 files changed, 15 insertions(+)
 create mode 100644 .server-changes/session-run-streams-version.md

diff --git a/.server-changes/session-run-streams-version.md b/.server-changes/session-run-streams-version.md
new file mode 100644
index 0000000000..09ae4b58e2
--- /dev/null
+++ b/.server-changes/session-run-streams-version.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Run-scoped realtime streams written inside a chat session run now use the same streams backend as the session itself, instead of falling back to the older one.
diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts
index 10cd966d95..1e6957d2a7 100644
--- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts
+++ b/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,7 @@ async function triggerSessionRun(params: {
   const result = await service.call(session.taskIdentifier, environment, body, {
     triggerSource: "session",
     triggerAction: "trigger",
+    realtimeStreamsVersion: determineRealtimeStreamsVersion("v2"),
   });
 
   if (!result) {
diff --git a/apps/webapp/test/realtimeServices.replicaLag.test.ts b/apps/webapp/test/realtimeServices.replicaLag.test.ts
index 02bb605afe..c9dba960a0 100644
--- a/apps/webapp/test/realtimeServices.replicaLag.test.ts
+++ b/apps/webapp/test/realtimeServices.replicaLag.test.ts
@@ -412,6 +412,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(triggerState.calls[0]!.options.realtimeStreamsVersion).toBeDefined();
       expect(replica.wasHit("taskRun")).toBe(true);
 
       // Proof the null was lag-induced: the primary holds the resolvable friendlyId (≠ the cuid).

From 542dcc1b6cc95ee99cb743d96dd9d6ae3f9d4361 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 09:41:35 +0100
Subject: [PATCH 05/11] test(webapp): prove session runs route run-scoped
 streams to S2

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 to
a run-scoped stream the way `streams.append()` does, and asserts the version
stamped on the run alongside where the bytes actually landed.

The harness runs the webapp with `REALTIME_STREAMS_DEFAULT_VERSION` set to v2
against a live S2, so the pre-fix result is not a configuration gap: the run is
stamped v1, S2 holds nothing, and the record sits in Redis.

Exposes the harness Redis connection so a test can assert the backend directly
rather than inferring it from an absence of records in S2.
---
 .../test/sessionRunStreamsBackend.e2e.test.ts | 135 ++++++++++++++++++
 .../testcontainers/src/webapp.ts              |  16 ++-
 2 files changed, 150 insertions(+), 1 deletion(-)
 create mode 100644 apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts

diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
new file mode 100644
index 0000000000..886cab88ee
--- /dev/null
+++ b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
@@ -0,0 +1,135 @@
+/**
+ * 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 = "browserPreview";
+const PART_ID = "part-1";
+
+/** 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}`;
+}
+
+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: { friendlyId: true, realtimeStreamsVersion: true, streamBasinName: true },
+    });
+
+    const appendRes = 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,
+        },
+        body: JSON.stringify({ frame: "a".repeat(1024) }),
+      }
+    );
+
+    expect(appendRes.status).toBe(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; recordsInS2: boolean; keyInRedis: boolean };
+    try {
+      observed = {
+        version: run.realtimeStreamsVersion,
+        recordsInS2: (await s2Body(streamName)).includes(PART_ID),
+        keyInRedis: (await redis.exists(redisStreamKey(created.runId, STREAM_ID))) === 1,
+      };
+    } finally {
+      redis.disconnect();
+    }
+
+    expect(observed).toEqual({ version: "v2", recordsInS2: true, keyInRedis: false });
+  });
+});
diff --git a/internal-packages/testcontainers/src/webapp.ts b/internal-packages/testcontainers/src/webapp.ts
index 22d3485f8f..00f6abd7fc 100644
--- a/internal-packages/testcontainers/src/webapp.ts
+++ b/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 };
 }
 
 /**
@@ -348,5 +354,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,
+  };
 }

From 493d2836aad5c5c751b67fad5223d3d7301502a9 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 13:52:19 +0100
Subject: [PATCH 06/11] fix(webapp): follow the session's basin and consolidate
 the streams-version work

Threads the organization basin into the session run's version resolution, so a
per-org-basin deployment with no global basin resolves v2 for the session run
the same way the trigger, batch and replay paths already do. Without it the
session path would silently degrade to v1 on exactly the configuration the
resolver change exists to preserve.

Also folds the session-run work into this branch so the two halves ship as one
change, uses a neutral stream id in the e2e, and appends records at a realistic
size so the backend's per-record limit is exercised.
---
 .../session-run-streams-version.md            |  6 ---
 .server-changes/streams-version-s2-guard.md   |  2 +-
 .../realtime/sessionRunManager.server.ts      |  5 +-
 .../test/sessionRunStreamsBackend.e2e.test.ts | 50 ++++++++++++-------
 4 files changed, 36 insertions(+), 27 deletions(-)
 delete mode 100644 .server-changes/session-run-streams-version.md

diff --git a/.server-changes/session-run-streams-version.md b/.server-changes/session-run-streams-version.md
deleted file mode 100644
index 09ae4b58e2..0000000000
--- a/.server-changes/session-run-streams-version.md
+++ /dev/null
@@ -1,6 +0,0 @@
----
-area: webapp
-type: fix
----
-
-Run-scoped realtime streams written inside a chat session run now use the same streams backend as the session itself, instead of falling back to the older one.
diff --git a/.server-changes/streams-version-s2-guard.md b/.server-changes/streams-version-s2-guard.md
index 6114ee44eb..0fbfcb1514 100644
--- a/.server-changes/streams-version-s2-guard.md
+++ b/.server-changes/streams-version-s2-guard.md
@@ -3,4 +3,4 @@ area: webapp
 type: fix
 ---
 
-Runs no longer end up with realtime streams that cannot be read or written.
+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.
diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts
index 1e6957d2a7..53d436e1e5 100644
--- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts
+++ b/apps/webapp/app/services/realtime/sessionRunManager.server.ts
@@ -317,7 +317,10 @@ async function triggerSessionRun(params: {
   const result = await service.call(session.taskIdentifier, environment, body, {
     triggerSource: "session",
     triggerAction: "trigger",
-    realtimeStreamsVersion: determineRealtimeStreamsVersion("v2"),
+    realtimeStreamsVersion: determineRealtimeStreamsVersion(
+      "v2",
+      environment.organization.streamBasinName
+    ),
   });
 
   if (!result) {
diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
index 886cab88ee..a2ba608da0 100644
--- a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
+++ b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
@@ -33,8 +33,10 @@ afterAll(async () => {
   await server?.stop();
 }, 120_000);
 
-const STREAM_ID = "browserPreview";
-const PART_ID = "part-1";
+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: {
@@ -52,6 +54,12 @@ 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(
@@ -96,20 +104,24 @@ describe("session runs and the realtime streams backend", () => {
       select: { friendlyId: true, realtimeStreamsVersion: true, streamBasinName: true },
     });
 
-    const appendRes = 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,
-        },
-        body: JSON.stringify({ frame: "a".repeat(1024) }),
-      }
-    );
-
-    expect(appendRes.status).toBe(200);
+    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,
@@ -119,17 +131,17 @@ describe("session runs and the realtime streams backend", () => {
       streamId: STREAM_ID,
     });
     const redis = new Redis({ host: server.redis.host, port: server.redis.port });
-    let observed: { version: string; recordsInS2: boolean; keyInRedis: boolean };
+    let observed: { version: string; framesInS2: number; keyInRedis: boolean };
     try {
       observed = {
         version: run.realtimeStreamsVersion,
-        recordsInS2: (await s2Body(streamName)).includes(PART_ID),
+        framesInS2: framesFound(await s2Body(streamName)),
         keyInRedis: (await redis.exists(redisStreamKey(created.runId, STREAM_ID))) === 1,
       };
     } finally {
       redis.disconnect();
     }
 
-    expect(observed).toEqual({ version: "v2", recordsInS2: true, keyInRedis: false });
+    expect(observed).toEqual({ version: "v2", framesInS2: FRAME_COUNT, keyInRedis: false });
   });
 });

From afaf3b4c2522c42b47d077355ca8c20cd817a301 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 14:08:46 +0100
Subject: [PATCH 07/11] fix(webapp): resolve a session run's version from the
 session's own basin

The organization basin can be cleared, so it is not a safe stand-in for the
session's. `deprovisionBasinForOrg` nulls the column, and a session created
while a basin existed keeps its pinned copy and keeps serving v2 channels. A
run triggered for that session afterwards would have read the now-null org
basin, degraded to v1, and reintroduced the session-to-run mismatch this
change exists to remove.

The session's own basin now takes precedence, mirroring `resolveStreamBasin`,
which resolves run then session then organization then the global setting.

Also completes the replica-lag fixtures, which passed an environment with no
organization and only worked while the basin was never read.
---
 .../realtime/sessionRunManager.server.ts      | 21 +++++++++++++++----
 .../test/realtimeServices.replicaLag.test.ts  | 10 +++++++--
 2 files changed, 25 insertions(+), 6 deletions(-)

diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts
index 53d436e1e5..06873182be 100644
--- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts
+++ b/apps/webapp/app/services/realtime/sessionRunManager.server.ts
@@ -52,7 +52,13 @@ type EnsureRunForSessionParams = {
    */
   session: Pick<
     Session,
-    "id" | "friendlyId" | "taskIdentifier" | "triggerConfig" | "currentRunId" | "currentRunVersion"
+    | "id"
+    | "friendlyId"
+    | "taskIdentifier"
+    | "triggerConfig"
+    | "currentRunId"
+    | "currentRunVersion"
+    | "streamBasinName"
   >;
   environment: AuthenticatedEnvironment;
   reason: EnsureRunReason;
@@ -235,6 +241,7 @@ export async function ensureRunForSession(
       triggerConfig: true,
       currentRunId: true,
       currentRunVersion: true,
+      streamBasinName: true,
     },
   });
 
@@ -284,7 +291,7 @@ export async function ensureRunForSession(
  * degrades to v1 where v2 streams are not configured.
  */
 async function triggerSessionRun(params: {
-  session: Pick<Session, "id" | "taskIdentifier">;
+  session: Pick<Session, "id" | "taskIdentifier" | "streamBasinName">;
   config: SessionTriggerConfig;
   environment: AuthenticatedEnvironment;
   payloadOverrides?: Record<string, unknown>;
@@ -319,7 +326,7 @@ async function triggerSessionRun(params: {
     triggerAction: "trigger",
     realtimeStreamsVersion: determineRealtimeStreamsVersion(
       "v2",
-      environment.organization.streamBasinName
+      session.streamBasinName ?? environment.organization.streamBasinName
     ),
   });
 
@@ -341,7 +348,13 @@ type SwapSessionRunParams = {
    */
   session: Pick<
     Session,
-    "id" | "friendlyId" | "taskIdentifier" | "triggerConfig" | "currentRunId" | "currentRunVersion"
+    | "id"
+    | "friendlyId"
+    | "taskIdentifier"
+    | "triggerConfig"
+    | "currentRunId"
+    | "currentRunVersion"
+    | "streamBasinName"
   >;
   /**
    * The run requesting the swap. Optimistic claim requires
diff --git a/apps/webapp/test/realtimeServices.replicaLag.test.ts b/apps/webapp/test/realtimeServices.replicaLag.test.ts
index c9dba960a0..0cf65f7617 100644
--- a/apps/webapp/test/realtimeServices.replicaLag.test.ts
+++ b/apps/webapp/test/realtimeServices.replicaLag.test.ts
@@ -331,7 +331,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",
       });
 
@@ -401,7 +404,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",
       });
 

From c36fecbd074a4b6b320647a7a1d0e77c84746503 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 14:21:53 +0100
Subject: [PATCH 08/11] fix(webapp): gate the session run's version on the
 basin the run will carry

The run is created with the organization's basin, and run-scoped stream routes
resolve their basin from the run alone. The session's own basin never enters
that path, so gating the version on it can stamp a run v2 whose basin resolves
to nothing, and every stream operation on that run then throws for its whole
life.

Reverts to the organization basin, which is the value the run is actually
created with. Where an organization has been deprovisioned, its sessions keep
serving v2 channels from their pinned basin while newly triggered runs degrade
to v1, which still works, rather than failing outright.
---
 .../realtime/sessionRunManager.server.ts      | 21 ++++---------------
 1 file changed, 4 insertions(+), 17 deletions(-)

diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts
index 06873182be..53d436e1e5 100644
--- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts
+++ b/apps/webapp/app/services/realtime/sessionRunManager.server.ts
@@ -52,13 +52,7 @@ type EnsureRunForSessionParams = {
    */
   session: Pick<
     Session,
-    | "id"
-    | "friendlyId"
-    | "taskIdentifier"
-    | "triggerConfig"
-    | "currentRunId"
-    | "currentRunVersion"
-    | "streamBasinName"
+    "id" | "friendlyId" | "taskIdentifier" | "triggerConfig" | "currentRunId" | "currentRunVersion"
   >;
   environment: AuthenticatedEnvironment;
   reason: EnsureRunReason;
@@ -241,7 +235,6 @@ export async function ensureRunForSession(
       triggerConfig: true,
       currentRunId: true,
       currentRunVersion: true,
-      streamBasinName: true,
     },
   });
 
@@ -291,7 +284,7 @@ export async function ensureRunForSession(
  * degrades to v1 where v2 streams are not configured.
  */
 async function triggerSessionRun(params: {
-  session: Pick<Session, "id" | "taskIdentifier" | "streamBasinName">;
+  session: Pick<Session, "id" | "taskIdentifier">;
   config: SessionTriggerConfig;
   environment: AuthenticatedEnvironment;
   payloadOverrides?: Record<string, unknown>;
@@ -326,7 +319,7 @@ async function triggerSessionRun(params: {
     triggerAction: "trigger",
     realtimeStreamsVersion: determineRealtimeStreamsVersion(
       "v2",
-      session.streamBasinName ?? environment.organization.streamBasinName
+      environment.organization.streamBasinName
     ),
   });
 
@@ -348,13 +341,7 @@ type SwapSessionRunParams = {
    */
   session: Pick<
     Session,
-    | "id"
-    | "friendlyId"
-    | "taskIdentifier"
-    | "triggerConfig"
-    | "currentRunId"
-    | "currentRunVersion"
-    | "streamBasinName"
+    "id" | "friendlyId" | "taskIdentifier" | "triggerConfig" | "currentRunId" | "currentRunVersion"
   >;
   /**
    * The run requesting the swap. Optimistic claim requires

From ea239f242e8da2875d9b7aa935ba3ad45f67ce23 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 14:54:07 +0100
Subject: [PATCH 09/11] test(webapp): cover the basin configuration that hid
 two bugs in review

Three gaps, all of which let a wrong basin through unnoticed.

The resolver suite now asserts an invariant across the whole configuration
matrix rather than a handful of cases: returning v2 requires a basin, so no
future combination can hand back a version the run cannot serve.

The session swap case pins which basin the trigger path reads. It seeds a
session with a basin of its own and asserts the organization's reaches the
resolver, because the run row carries the organization's and the run-scoped
stream routes resolve against that alone.

A new e2e runs the harness with no global basin and per-org basins enabled,
the configuration where basin resolution actually decides something. Every
existing test ran with a global basin set, which makes any basin value work
and hides the whole class.
---
 .../determineRealtimeStreamsVersion.test.ts   |  29 +++
 .../test/realtimeServices.replicaLag.test.ts  |  13 ++
 .../sessionRunStreamsPerOrgBasin.e2e.test.ts  | 173 ++++++++++++++++++
 .../testcontainers/src/webapp.ts              |   5 +-
 4 files changed, 219 insertions(+), 1 deletion(-)
 create mode 100644 apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts

diff --git a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
index 9e28c12872..6258382101 100644
--- a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
+++ b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
@@ -84,3 +84,32 @@ describe("resolveRealtimeStreamsVersion", () => {
     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 is 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 };
+              if (resolveRealtimeStreamsVersion(requested, config) === "v2" && !basin) {
+                counterexamples.push(JSON.stringify({ requested, ...config }));
+              }
+            }
+          }
+        }
+      }
+    }
+
+    expect(counterexamples).toEqual([]);
+  });
+});
diff --git a/apps/webapp/test/realtimeServices.replicaLag.test.ts b/apps/webapp/test/realtimeServices.replicaLag.test.ts
index 0cf65f7617..569e18a11c 100644
--- a/apps/webapp/test/realtimeServices.replicaLag.test.ts
+++ b/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 } },
@@ -389,6 +398,7 @@ describe("realtime-svc — replica-lag guards", () => {
           triggerConfig: { basePayload: {} },
           currentRunId: callingRunId,
           currentRunVersion: 0,
+          streamBasinName: "session-pinned-basin",
         },
       });
 
@@ -397,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 } };
@@ -419,6 +430,8 @@ describe("realtime-svc — replica-lag guards", () => {
       expect(triggerState.calls).toHaveLength(1);
       expect(triggerState.calls[0]!.body.payload.previousRunId).toBe(callingRunId);
       expect(triggerState.calls[0]!.options.realtimeStreamsVersion).toBeDefined();
+
+      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).
diff --git a/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts b/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts
new file mode 100644
index 0000000000..874518f738
--- /dev/null
+++ b/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts
@@ -0,0 +1,173 @@
+/**
+ * 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.
+ *
+ * The unprovisioned case asserts the run carries a null basin and still serves
+ * its streams. 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 has 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 and still serves its streams", 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 });
+  });
+});
diff --git a/internal-packages/testcontainers/src/webapp.ts b/internal-packages/testcontainers/src/webapp.ts
index 00f6abd7fc..2ff73b1b52 100644
--- a/internal-packages/testcontainers/src/webapp.ts
+++ b/internal-packages/testcontainers/src/webapp.ts
@@ -286,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;
@@ -328,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 ?? {}),
         },
       }
     );

From e48296727290ed25d5c95994357e035542acb0b6 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 17:42:05 +0100
Subject: [PATCH 10/11] test(webapp): remove assertions that could not fail and
 claims the tests do not make

Stubbing the resolver to capture its arguments made the neighbouring version
assertion unfalsifiable, since the stub returns a version unconditionally. The
argument assertion is the one carrying weight, so the other is gone.

The per-org e2e claimed an unprovisioned organization still serves its streams.
It only drives run-scoped streams; that organization's session channels cannot
resolve a basin and fail. Retitled and scoped so nobody reads it as evidence a
session is healthy.

The resolver invariant now covers an empty basin, which is what the harness
actually passes for the global setting, and asserts credentials as well as a
basin, since serving v2 needs both.
---
 .../test/determineRealtimeStreamsVersion.test.ts |  7 ++++---
 .../test/realtimeServices.replicaLag.test.ts     |  2 --
 .../test/sessionRunStreamsBackend.e2e.test.ts    |  2 +-
 .../sessionRunStreamsPerOrgBasin.e2e.test.ts     | 16 ++++++++++------
 4 files changed, 15 insertions(+), 12 deletions(-)

diff --git a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
index 6258382101..b5d6592e4a 100644
--- a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
+++ b/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
@@ -86,13 +86,13 @@ describe("resolveRealtimeStreamsVersion", () => {
 });
 
 describe("resolveRealtimeStreamsVersion invariant", () => {
-  const BASINS = [undefined, "a-basin"];
+  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 is present, for every configuration", () => {
+  it("only returns v2 when a basin and credentials are both present, for every configuration", () => {
     const counterexamples: string[] = [];
 
     for (const basin of BASINS) {
@@ -101,7 +101,8 @@ describe("resolveRealtimeStreamsVersion invariant", () => {
           for (const defaultVersion of DEFAULTS) {
             for (const requested of REQUESTED) {
               const config = { defaultVersion, basin, accessToken, skipAccessTokens };
-              if (resolveRealtimeStreamsVersion(requested, config) === "v2" && !basin) {
+              const usable = Boolean(basin) && (Boolean(accessToken) || skipAccessTokens);
+              if (resolveRealtimeStreamsVersion(requested, config) === "v2" && !usable) {
                 counterexamples.push(JSON.stringify({ requested, ...config }));
               }
             }
diff --git a/apps/webapp/test/realtimeServices.replicaLag.test.ts b/apps/webapp/test/realtimeServices.replicaLag.test.ts
index 569e18a11c..6a302dcfd9 100644
--- a/apps/webapp/test/realtimeServices.replicaLag.test.ts
+++ b/apps/webapp/test/realtimeServices.replicaLag.test.ts
@@ -429,8 +429,6 @@ 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(triggerState.calls[0]!.options.realtimeStreamsVersion).toBeDefined();
-
       expect(versionCalls.at(-1)).toEqual({ requested: "v2", basin: null });
       expect(replica.wasHit("taskRun")).toBe(true);
 
diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
index a2ba608da0..e500a3eb2b 100644
--- a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
+++ b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts
@@ -101,7 +101,7 @@ describe("session runs and the realtime streams backend", () => {
 
     const run = await server.prisma.taskRun.findFirstOrThrow({
       where: { friendlyId: created.runId },
-      select: { friendlyId: true, realtimeStreamsVersion: true, streamBasinName: true },
+      select: { realtimeStreamsVersion: true },
     });
 
     const appendStatuses: number[] = [];
diff --git a/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts b/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts
index 874518f738..5e7527c901 100644
--- a/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts
+++ b/apps/webapp/test/sessionRunStreamsPerOrgBasin.e2e.test.ts
@@ -10,11 +10,15 @@
  * are asserted: a provisioned organization reaches S2, and an unprovisioned one
  * degrades to v1 and keeps working on Redis.
  *
- * The unprovisioned case asserts the run carries a null basin and still serves
- * its streams. 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 has one
- * of its own.
+ * 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
  */
@@ -139,7 +143,7 @@ describe("session runs with per-org basins and no global basin", () => {
     expect(observed).toEqual({ version: "v2", runBasin: basin, inS2: true, keyInRedis: false });
   });
 
-  it("degrades to v1 for an unprovisioned organization and still serves its streams", async () => {
+  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({

From ec290c676badb8334106b7430079c7289bdf8a87 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Tue, 11 Aug 2026 18:02:52 +0100
Subject: [PATCH 11/11] refactor(webapp): move the pure streams-version
 resolver out of the env-bound module

The resolver was written to take its configuration rather than read `env`, but
it still lived beside the singleton that reads it, so the unit test pulled
`env.server` in transitively, which the webapp rules forbid. Same split as
nativeRealtimeClient and its instance module: the pure function has no imports
at all, and the env-bound wrapper stays where the singletons are.

Renames the test to match the function it exercises.
---
 .../realtime/realtimeStreamsVersion.ts        | 44 +++++++++++++++++++
 .../realtime/v1StreamsGlobal.server.ts        | 41 +++--------------
 ...test.ts => realtimeStreamsVersion.test.ts} |  2 +-
 3 files changed, 50 insertions(+), 37 deletions(-)
 create mode 100644 apps/webapp/app/services/realtime/realtimeStreamsVersion.ts
 rename apps/webapp/test/{determineRealtimeStreamsVersion.test.ts => realtimeStreamsVersion.test.ts} (98%)

diff --git a/apps/webapp/app/services/realtime/realtimeStreamsVersion.ts b/apps/webapp/app/services/realtime/realtimeStreamsVersion.ts
new file mode 100644
index 0000000000..a7d207c713
--- /dev/null
+++ b/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";
+}
diff --git a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts b/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
index 557a23e816..27305f676e 100644
--- a/apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts
+++ b/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,42 +100,7 @@ function streamPrefixFor(environment: AuthenticatedEnvironment, basin: string):
   return segments.join("/");
 }
 
-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";
-}
+export type { RealtimeStreamsVersionConfig };
 
 /**
  * Pass `organizationBasinName` wherever the caller has it. It mirrors the
diff --git a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts b/apps/webapp/test/realtimeStreamsVersion.test.ts
similarity index 98%
rename from apps/webapp/test/determineRealtimeStreamsVersion.test.ts
rename to apps/webapp/test/realtimeStreamsVersion.test.ts
index b5d6592e4a..4ffa973d21 100644
--- a/apps/webapp/test/determineRealtimeStreamsVersion.test.ts
+++ b/apps/webapp/test/realtimeStreamsVersion.test.ts
@@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest";
 import {
   resolveRealtimeStreamsVersion,
   type RealtimeStreamsVersionConfig,
-} from "~/services/realtime/v1StreamsGlobal.server";
+} from "~/services/realtime/realtimeStreamsVersion";
 
 const NO_S2: RealtimeStreamsVersionConfig = {
   defaultVersion: "v1",