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). diff --git a/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts b/apps/webapp/test/sessionRunStreamsBackend.e2e.test.ts new file mode 100644 index 0000000000..a2ba608da0 --- /dev/null +++ b/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 { + 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 }); + }); +}); 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, + }; }