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.

Author: matt-aitken

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([]);
+  });
+});

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

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

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 ?? {}),
         },
       }
     );