From 1464758bbeea0f052bae378a7da8c8652b96aa02 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Tue, 11 Aug 2026 02:03:52 +0100 Subject: [PATCH 1/4] fix(webapp): keep session runs off the legacy realtime streams backend Runs created for a Session were triggered without a realtime streams version, so they fell through to the `realtimeStreamsVersion` column default of v1. A Session's own channels are always v2, so run-scoped `streams.*` calls made inside a session run resolved to a different backend than the session itself, for the whole life of the run. The trigger now resolves the version explicitly, degrading to v1 where v2 streams are not configured. --- .server-changes/session-run-streams-version.md | 6 ++++++ .../app/services/realtime/sessionRunManager.server.ts | 8 ++++++++ apps/webapp/test/realtimeServices.replicaLag.test.ts | 1 + 3 files changed, 15 insertions(+) create mode 100644 .server-changes/session-run-streams-version.md diff --git a/.server-changes/session-run-streams-version.md b/.server-changes/session-run-streams-version.md new file mode 100644 index 0000000000..09ae4b58e2 --- /dev/null +++ b/.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. diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts index 10cd966d95..1e6957d2a7 100644 --- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts +++ b/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; @@ -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) { diff --git a/apps/webapp/test/realtimeServices.replicaLag.test.ts b/apps/webapp/test/realtimeServices.replicaLag.test.ts index 02bb605afe..c9dba960a0 100644 --- a/apps/webapp/test/realtimeServices.replicaLag.test.ts +++ b/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). From 1826f9a73a7751c965433d3522a9d40da8b97125 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Tue, 11 Aug 2026 09:41:35 +0100 Subject: [PATCH 2/4] test(webapp): prove session runs route run-scoped streams to S2 Boots the real webapp plus Postgres, Redis and s2-lite, creates a Session through the public API so the run comes from the real trigger path, appends to a run-scoped stream the way `streams.append()` does, and asserts the version stamped on the run alongside where the bytes actually landed. The harness runs the webapp with `REALTIME_STREAMS_DEFAULT_VERSION` set to v2 against a live S2, so the pre-fix result is not a configuration gap: the run is stamped v1, S2 holds nothing, and the record sits in Redis. Exposes the harness Redis connection so a test can assert the backend directly rather than inferring it from an absence of records in S2. --- .../test/sessionRunStreamsBackend.e2e.test.ts | 135 ++++++++++++++++++ .../testcontainers/src/webapp.ts | 16 ++- 2 files changed, 150 insertions(+), 1 deletion(-) create mode 100644 apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts new file mode 100644 index 0000000000..886cab88ee --- /dev/null +++ b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts @@ -0,0 +1,135 @@ +/** + * 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 = "browserPreview"; +const PART_ID = "part-1"; + +/** 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}`; +} + +async function s2Body(streamName: string): Promise { + 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 appendRes = 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, + }, + body: JSON.stringify({ frame: "a".repeat(1024) }), + } + ); + + expect(appendRes.status).toBe(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; recordsInS2: boolean; keyInRedis: boolean }; + try { + observed = { + version: run.realtimeStreamsVersion, + recordsInS2: (await s2Body(streamName)).includes(PART_ID), + keyInRedis: (await redis.exists(redisStreamKey(created.runId, STREAM_ID))) === 1, + }; + } finally { + redis.disconnect(); + } + + expect(observed).toEqual({ version: "v2", recordsInS2: true, keyInRedis: false }); + }); +}); diff --git a/internal-packages/testcontainers/src/webapp.ts b/internal-packages/testcontainers/src/webapp.ts index 22d3485f8f..00f6abd7fc 100644 --- a/internal-packages/testcontainers/src/webapp.ts +++ b/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 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, + }; } From 31d490a323027fd24d0929226bbc0a60856ca95b Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Tue, 11 Aug 2026 12:45:43 +0100 Subject: [PATCH 3/4] test(webapp): exercise the real frame size in the streams-backend e2e A single 1KB append proved routing but not that the backend tolerates the payload shape this fixes: browser preview frames are roughly 250KB each. The test now appends eight of them and asserts every one is readable from S2, so a regression in either the route's body cap or S2's record cap fails here rather than in production. --- .../test/sessionRunStreamsBackend.e2e.test.ts | 48 ++++++++++++------- 1 file changed, 30 insertions(+), 18 deletions(-) diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts index 886cab88ee..8db4be9de3 100644 --- a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts +++ b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts @@ -34,7 +34,9 @@ afterAll(async () => { }, 120_000); const STREAM_ID = "browserPreview"; -const PART_ID = "part-1"; +const PART_ID = "part"; +const FRAME_BYTES = 250 * 1024; +const FRAME_COUNT = 8; /** Mirrors `S2RealtimeStreams.toStreamName` on the shared-basin prefix. */ function runStreamName(p: { @@ -52,6 +54,12 @@ 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 { const qs = new URLSearchParams({ seq_num: "0", clamp: "true", wait: "0" }); const res = await fetch( @@ -96,20 +104,24 @@ describe("session runs and the realtime streams backend", () => { select: { friendlyId: true, realtimeStreamsVersion: true, streamBasinName: true }, }); - const appendRes = 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, - }, - body: JSON.stringify({ frame: "a".repeat(1024) }), - } - ); - - expect(appendRes.status).toBe(200); + 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, @@ -119,17 +131,17 @@ describe("session runs and the realtime streams backend", () => { streamId: STREAM_ID, }); const redis = new Redis({ host: server.redis.host, port: server.redis.port }); - let observed: { version: string; recordsInS2: boolean; keyInRedis: boolean }; + let observed: { version: string; framesInS2: number; keyInRedis: boolean }; try { observed = { version: run.realtimeStreamsVersion, - recordsInS2: (await s2Body(streamName)).includes(PART_ID), + framesInS2: framesFound(await s2Body(streamName)), keyInRedis: (await redis.exists(redisStreamKey(created.runId, STREAM_ID))) === 1, }; } finally { redis.disconnect(); } - expect(observed).toEqual({ version: "v2", recordsInS2: true, keyInRedis: false }); + expect(observed).toEqual({ version: "v2", framesInS2: FRAME_COUNT, keyInRedis: false }); }); }); From 9a885267a0d369c6f39c91d6b96a3d3a70e71702 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Tue, 11 Aug 2026 13:16:29 +0100 Subject: [PATCH 4/4] test(webapp): use a neutral stream id in the streams-backend e2e --- apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts index 8db4be9de3..a2ba608da0 100644 --- a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts +++ b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts @@ -33,7 +33,7 @@ afterAll(async () => { await server?.stop(); }, 120_000); -const STREAM_ID = "browserPreview"; +const STREAM_ID = "frames"; const PART_ID = "part"; const FRAME_BYTES = 250 * 1024; const FRAME_COUNT = 8;