test(webapp): cover more session-stream scenarios

Adds wire-level legs drawn from the ai-chat smoke-test catalog: the
`.in/append` client-channel round-trip, the server-side peek fast-close
(`X-Session-Settled`) that pairs with the client caught-up close, a 413 on an
oversized `.in/append` body still carrying CORS headers, and rejection of a
subscribe with an invalid token. Seeds a real Session row so the append path
resolves.

Author: ericallam

apps/webapp/test/helpers/sessionStream.ts

@@ -99,10 +99,84 @@ export type SubscribeOptions = {
   lastEventId?: string;
   timeoutInSeconds?: number;
   peekSettled?: boolean;
+  io?: "out" | "in";
 };
 
+function sessionChannelUrl(baseUrl: string, addressingKey: string, io: "out" | "in"): string {
+  return `${baseUrl}/realtime/v1/sessions/${encodeURIComponent(addressingKey)}/${io}`;
+}
+
+/**
+ * Append a record to a session's `.in` channel through the webapp route the
+ * browser uses (client -> agent). Returns the status, the reflected
+ * Access-Control-Allow-Origin (when an Origin was sent), and the parsed body.
+ */
+export async function appendInput(opts: {
+  baseUrl: string;
+  addressingKey: string;
+  token: string;
+  body: string;
+  partId?: string;
+  origin?: string;
+}): Promise<{ status: number; acao: string | null; json: unknown }> {
+  const url = `${sessionChannelUrl(opts.baseUrl, opts.addressingKey, "in")}/append`;
+  const res = await fetch(url, {
+    method: "POST",
+    headers: {
+      Authorization: `Bearer ${opts.token}`,
+      "Content-Type": "application/json",
+      ...(opts.partId ? { "X-Part-Id": opts.partId } : {}),
+      ...(opts.origin ? { Origin: opts.origin } : {}),
+    },
+    body: opts.body,
+  });
+  let json: unknown;
+  try {
+    json = await res.json();
+  } catch {}
+  return { status: res.status, acao: res.headers.get("access-control-allow-origin"), json };
+}
+
+/**
+ * Raw SSE GET against a session channel, exposing the response status +
+ * headers (which `SSEStreamSubscription` hides). Reads the body to close and
+ * reports how long that took, for asserting the server-side peek fast-close.
+ */
+export async function openChannelRaw(
+  opts: SubscribeOptions & { maxMs?: number }
+): Promise<{ status: number; sessionSettled: string | null; closedMs: number; body: string }> {
+  const url = sessionChannelUrl(opts.baseUrl, opts.addressingKey, opts.io ?? "out");
+  const started = performance.now();
+  const res = await fetch(url, {
+    headers: {
+      Authorization: `Bearer ${opts.token}`,
+      Accept: "text/event-stream",
+      ...(opts.lastEventId ? { "Last-Event-ID": opts.lastEventId } : {}),
+      ...(opts.timeoutInSeconds ? { "Timeout-Seconds": String(opts.timeoutInSeconds) } : {}),
+      ...(opts.peekSettled ? { "X-Peek-Settled": "1" } : {}),
+    },
+  });
+  const sessionSettled = res.headers.get("x-session-settled");
+  let body = "";
+  if (res.body) {
+    const reader = res.body.getReader();
+    const decoder = new TextDecoder();
+    const deadline = started + (opts.maxMs ?? 15_000);
+    try {
+      while (performance.now() < deadline) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        body += decoder.decode(value, { stream: true });
+      }
+    } finally {
+      await reader.cancel().catch(() => {});
+    }
+  }
+  return { status: res.status, sessionSettled, closedMs: performance.now() - started, body };
+}
+
 export function subscribeSessionOut(opts: SubscribeOptions): SSEStreamSubscription {
-  const url = `${opts.baseUrl}/realtime/v1/sessions/${encodeURIComponent(opts.addressingKey)}/out`;
+  const url = sessionChannelUrl(opts.baseUrl, opts.addressingKey, opts.io ?? "out");
   return new SSEStreamSubscription(url, {
     headers: {
       Authorization: `Bearer ${opts.token}`,

apps/webapp/test/session-stream.e2e.test.ts

@@ -16,10 +16,12 @@ import type { SessionStreamTestServer } from "@internal/testcontainers/webapp";
 import { startSessionStreamTestServer } from "@internal/testcontainers/webapp";
 import { seedTestEnvironment } from "./helpers/seedTestEnvironment";
 import {
+  appendInput,
   collectSessionOut,
   collectUntilCaughtUp,
   isTurnComplete,
   mintSessionToken,
+  openChannelRaw,
   SessionStreamProducer,
   sessionStreamName,
 } from "./helpers/sessionStream";
@@ -37,8 +39,21 @@ afterAll(async () => {
 }, 120_000);
 
 async function setupSession() {
-  const { organization, environment, apiKey } = await seedTestEnvironment(server.prisma);
+  const { organization, project, environment, apiKey } = await seedTestEnvironment(server.prisma);
   const addressingKey = `sess-${randomBytes(6).toString("hex")}`;
+  await server.prisma.session.create({
+    data: {
+      friendlyId: `session_${randomBytes(8).toString("hex")}`,
+      externalId: addressingKey,
+      type: "chat.agent",
+      projectId: project.id,
+      runtimeEnvironmentId: environment.id,
+      environmentType: environment.type,
+      organizationId: organization.id,
+      taskIdentifier: "chat-agent",
+      triggerConfig: {},
+    },
+  });
   const token = await mintSessionToken({ apiKey, envId: environment.id, addressingKey });
   const streamName = sessionStreamName({
     orgId: organization.id,
@@ -213,4 +228,81 @@ describe("session stream e2e", () => {
     expect(dataChunks).toEqual([0, 1, 2]);
     expect(parts.some(isTurnComplete)).toBe(true);
   });
+
+  it("E8 in/append 413 for an oversized body still carries CORS headers", async () => {
+    const { addressingKey, token, baseUrl } = await setupSession();
+
+    const oversized = "x".repeat(2 * 1024 * 1024);
+    const { status, acao } = await appendInput({
+      baseUrl,
+      addressingKey,
+      token,
+      origin: "http://example.com",
+      body: JSON.stringify({ kind: "message", payload: { big: oversized } }),
+    });
+
+    expect(status).toBe(413);
+    expect(acao).not.toBeNull();
+  });
+
+  it("E9 server peek fast-closes at a turn-complete tail with X-Session-Settled", async () => {
+    const { addressingKey, token, producer, baseUrl } = await setupSession();
+
+    await producer.appendData({ n: 0 }, "p0");
+    const tc = await producer.appendTurnComplete();
+
+    const { status, sessionSettled, closedMs } = await openChannelRaw({
+      baseUrl,
+      addressingKey,
+      token,
+      lastEventId: String(tc),
+      peekSettled: true,
+      timeoutInSeconds: 30,
+      maxMs: 10_000,
+    });
+
+    expect(status).toBe(200);
+    expect(sessionSettled).toBe("true");
+    expect(closedMs).toBeLessThan(5_000);
+  });
+
+  it("E11 in/append delivers the record on the .in channel", async () => {
+    const { addressingKey, token, baseUrl } = await setupSession();
+
+    const payload = JSON.stringify({ kind: "message", text: "hello from client" });
+    const appended = await appendInput({
+      baseUrl,
+      addressingKey,
+      token,
+      partId: "in-0",
+      body: payload,
+    });
+    expect(appended.status).toBe(200);
+
+    const { parts } = await collectSessionOut({
+      baseUrl,
+      addressingKey,
+      token,
+      io: "in",
+      until: (p) => p.some((x) => x.chunk != null),
+      maxMs: 15_000,
+    });
+
+    const got = parts.find((p) => p.chunk != null);
+    expect(got).toBeTruthy();
+    expect(String(got?.chunk)).toContain("hello from client");
+  });
+
+  it("E12 subscribe with an invalid token is rejected", async () => {
+    const { addressingKey, baseUrl } = await setupSession();
+
+    const { status } = await openChannelRaw({
+      baseUrl,
+      addressingKey,
+      token: "tr_pub_invalid_not_a_real_token",
+      maxMs: 5_000,
+    });
+
+    expect([401, 403]).toContain(status);
+  });
 });