test(core): temporary diagnostic for failing agent S2 writes

Logs the S2 append pipeline error (basin, stream, endpoint, message) when
a session .out write fails, to capture why the agent's direct-to-S2 writes
return an empty stream in CI while succeeding locally. To be reverted.

Author: ericallam

apps/webapp/test/helpers/agentHarness.ts

@@ -112,6 +112,17 @@ export function runRealChatAgent(opts: RunRealChatAgentOptions): RunningAgent {
     ) as Promise<void>
   ).finally(restore);
 
+  done.catch((e) => {
+    const err = e instanceof Error ? e : new Error(String(e));
+    console.error(
+      "[AGENT-DIAG] run rejected",
+      opts.agentId,
+      runId,
+      `${err.name}: ${err.message}`,
+      (err.stack ?? "").split("\n").slice(1, 6).join(" || ")
+    );
+  });
+
   return {
     done,
     close: async () => {

apps/webapp/test/helpers/testChatAgent.ts

@@ -55,6 +55,16 @@ export const testChatAgent = chat
 
     run: async ({ messages, signal }) => {
       const model = locals.get(testChatModelLocal);
+      console.error(
+        "[TURN-DIAG] e2e-test-chat run: messages=",
+        messages?.length,
+        "hasModel=",
+        !!model,
+        "aborted=",
+        signal?.aborted,
+        "lastRole=",
+        messages?.[messages.length - 1]?.role
+      );
       if (!model) {
         throw new Error("test model not injected via locals");
       }

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) };
     },
   });
 }
@@ -317,6 +335,18 @@ describe("session agent e2e (real chat.agent loop)", () => {
         maxMs: 30_000,
       });
 
+      const dbg = await fetch(
+        `${baseUrl}/realtime/v1/sessions/${encodeURIComponent(addressingKey)}/out/records`,
+        { headers: { Authorization: `Bearer ${token}` } }
+      );
+      console.error(
+        "[EA1-DIAG] collected parts:",
+        parts.length,
+        "readback:",
+        dbg.status,
+        (await dbg.text()).slice(0, 700)
+      );
+
       const text = parts
         .filter((p) => p.chunk != null)
         .map((p) => JSON.stringify(p.chunk))

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -152,6 +152,14 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
 
       this.log(`[S2MetadataStream] Starting stream pipeline`);
 
+      console.error(
+        "[S2WRITER-DIAG] pipeline START stream=",
+        this.options.stream,
+        "flushMs=",
+        this.flushIntervalMs
+      );
+      let recordCount = 0;
+
       // Convert source stream to AppendRecord format and pipe to S2
       await this.serverStream
         .pipeThrough(
@@ -166,6 +174,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
                 controller.error(encoded.error);
                 return;
               }
+              recordCount++;
               controller.enqueue(AppendRecord.string({ body: encoded.body }));
             },
           })
@@ -182,6 +191,15 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       // Get final position to verify completion
       const lastAcked = session.lastAckedPosition();
 
+      console.error(
+        "[S2WRITER-DIAG] pipeline DONE stream=",
+        this.options.stream,
+        "recordsIn=",
+        recordCount,
+        "lastAckedEnd=",
+        JSON.stringify(lastAcked?.end)
+      );
+
       if (lastAcked?.end) {
         this.lastSeqNum = lastAcked.end.seqNum;
         this.log(
@@ -193,6 +211,17 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
         this.log("[S2MetadataStream] Stream error occurred but stream was aborted");
         return;
       }
+      console.error(
+        "[S2WRITER-DIAG] pipeline error basin=",
+        this.options.basin,
+        "stream=",
+        this.options.stream,
+        "endpoint=",
+        this.options.endpoint,
+        "err=",
+        error instanceof Error ? `${error.name}: ${error.message}` : String(error),
+        (error instanceof Error ? error.stack ?? "" : "").split("\n").slice(1, 5).join(" || ")
+      );
       this.logError("[S2MetadataStream] Error in stream pipeline:", error);
       throw error;
     }