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.

Author: matt-aitken

.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.

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")
   ) {

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");
+  });
+});