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

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

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Runs no longer end up with realtime streams that cannot be read or written.

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/v1StreamsGlobal.server.ts

@@ -96,20 +96,59 @@ 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 = {
+  defaultVersion: "v1" | "v2";
+  /** A basin that will actually resolve at read/write time, or undefined if none will. */
+  basin?: string;
+  accessToken?: string;
+  skipAccessTokens: boolean;
+};
 
-  if (
-    streamVersion === "v2" &&
-    env.REALTIME_STREAMS_S2_BASIN &&
-    (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
-  ) {
-    return "v2";
+/**
+ * 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";
   }
 
-  return "v1";
+  const hasCredentials = Boolean(config.accessToken) || config.skipAccessTokens;
+
+  return hasCredentials && Boolean(config.basin) ? "v2" : "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/determineRealtimeStreamsVersion.test.ts

@@ -0,0 +1,86 @@
+import { describe, expect, it } from "vitest";
+import {
+  resolveRealtimeStreamsVersion,
+  type RealtimeStreamsVersionConfig,
+} from "~/services/realtime/v1StreamsGlobal.server";
+
+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");
+  });
+});