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

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

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

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

apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts

@@ -0,0 +1,147 @@
+/**
+ * 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 = "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: {
+  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}`;
+}
+
+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(
+    `${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 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,
+      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; framesInS2: number; keyInRedis: boolean };
+    try {
+      observed = {
+        version: run.realtimeStreamsVersion,
+        framesInS2: framesFound(await s2Body(streamName)),
+        keyInRedis: (await redis.exists(redisStreamKey(created.runId, STREAM_ID))) === 1,
+      };
+    } finally {
+      redis.disconnect();
+    }
+
+    expect(observed).toEqual({ version: "v2", framesInS2: FRAME_COUNT, keyInRedis: false });
+  });
+});

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