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
+---
+
+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;
-  }
-
-  if (
-    streamVersion === "v2" &&
-    env.REALTIME_STREAMS_S2_BASIN &&
-    (env.REALTIME_STREAMS_S2_ACCESS_TOKEN || env.REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS === "true")
-  ) {
-    return "v2";
-  }
+export type { RealtimeStreamsVersionConfig };
 
-  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,14 +407,18 @@ 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 } };
 
       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([]);
+  });
+});