fix(webapp): keep session .out writable when S2 access tokens are skipped

Self-hosted deployments that run S2 with REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS
and no configured access token got an empty token back from the session
channel initialize call. The SDK's session-stream writer treats an empty
access token as missing S2 credentials and never opens the writer, so a
chat.agent turn's assistant output was silently dropped: .out stayed empty
with no error surfaced.

The initialize response now hands back a non-empty placeholder token when
access tokens are skipped (S2 ignores it in that mode, but the client requires
a token to be present). The session-agent e2e runs the skip-access-tokens path
with no token so the regression stays covered.

Author: ericallam

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -74,6 +74,15 @@ export type S2RealtimeStreamsOptions = {
 const S2_TOKEN_OPS = ["append", "create-stream", "trim"] as const;
 const S2_TOKEN_OPS_FINGERPRINT = [...S2_TOKEN_OPS].sort().join(",");
 
+/**
+ * Placeholder handed back as the S2 access token when `skipAccessTokens` is set
+ * and no token is configured (self-hosted s2-lite ignores the token entirely).
+ * The SDK's session-stream writer rejects an empty access token as "no S2
+ * credentials" and never opens the writer, so the token must be non-empty even
+ * when it is semantically unused.
+ */
+const SKIP_ACCESS_TOKENS_SENTINEL = "s2-skip-access-tokens";
+
 type S2IssueAccessTokenResponse = { access_token: string };
 type S2AppendInput = { records: { body: string }[] };
 type S2AppendAck = {
@@ -168,7 +177,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     relativeName: string
   ): Promise<{ responseHeaders?: Record<string, string> }> {
     const accessToken = this.skipAccessTokens
-      ? this.token
+      ? this.token || SKIP_ACCESS_TOKENS_SENTINEL
       : await this.getS2AccessToken(randomUUID());
 
     return {

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

@@ -66,6 +66,24 @@ afterAll(async () => {
   await server?.stop();
 }, 120_000);
 
+/**
+ * A synchronous mock model stream: enqueues every chunk in one tick and closes.
+ * `simulateReadableStream`'s default 0ms delay still yields a `setTimeout(0)`
+ * macrotask between chunks, which a loaded CI event loop can starve so the
+ * later chunks never emit and the turn produces an empty `.out`. Emitting
+ * synchronously keeps the mock deterministic across environments.
+ */
+function simStream(chunks: readonly unknown[]): ReadableStream<any> {
+  return new ReadableStream({
+    start(controller) {
+      for (const chunk of chunks) {
+        controller.enqueue(chunk);
+      }
+      controller.close();
+    },
+  });
+}
+
 function textModel(text: string) {
   const chunks = [
     { type: "text-start", id: "t1" },
@@ -81,7 +99,7 @@ function textModel(text: string) {
     },
   ];
   return new MockLanguageModelV3({
-    doStream: async () => ({ stream: simulateReadableStream({ chunks: chunks as never }) }),
+    doStream: async () => ({ stream: simStream(chunks) }),
   });
 }
 
@@ -139,7 +157,7 @@ function echoModel() {
           },
         },
       ];
-      return { stream: simulateReadableStream({ chunks: chunks as never }) };
+      return { stream: simStream(chunks) };
     },
   });
 }
@@ -239,7 +257,7 @@ function toolCallThenText(opts: {
   let idx = 0;
   return new MockLanguageModelV3({
     doStream: async () => ({
-      stream: simulateReadableStream({ chunks: (idx++ === 0 ? call1 : call2) as never }),
+      stream: simStream(idx++ === 0 ? call1 : call2),
     }),
   });
 }
@@ -264,7 +282,7 @@ function sequenceModel(texts: string[]) {
         { type: "text-end", id: "t1" },
         FINISH_STOP,
       ];
-      return { stream: simulateReadableStream({ chunks: chunks as never }) };
+      return { stream: simStream(chunks) };
     },
   });
 }

internal-packages/testcontainers/src/webapp.ts

@@ -312,6 +312,7 @@ export async function startSessionStreamTestServer(): Promise<SessionStreamTestS
       { host: rc.getHost(), port: rc.getPort() },
       {
         extraEnv: {
+          REALTIME_STREAMS_S2_ACCESS_TOKEN: "",
           REALTIME_STREAMS_S2_BASIN: s2.basin,
           REALTIME_STREAMS_S2_ENDPOINT: `${s2.endpoint}/v1`,
           REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS: "true",