Keep a Codex turn going while the relay catches up on the app-server's output

backend/src/provider-connections/codex-subscription-relay.test.ts

@@ -313,6 +313,59 @@ describe("Codex subscription run relay", () => {
     ]);
   });
 
+  test("a burst of app-server output while its frames wait is delivered whole, never ending the turn", async () => {
+    // 35 minutes into a Codex turn the relay's frame queue passed 256 waiting
+    // lines, closed with 1008 ("serial task queue limit exceeded"), and the
+    // turn failed as "The provider event stream closed unexpectedly".
+    const server = startRelayServer();
+    const child = fakeAppServer();
+    let stall: PromiseWithResolvers<void> | null = null;
+    let stalled = 0;
+    setCodexSubscriptionRelayDependenciesForTest({
+      // Every frame is authorized before it is forwarded: a slow read holds them all.
+      selectRuntime: async () => {
+        if (stall) {
+          stalled += 1;
+          await stall.promise;
+        }
+        return runtime();
+      },
+      loadThreadBinding: async () => "provider-thread-1",
+      spawnAppServer: () => child.process,
+    });
+    const capability = issueCodexSubscriptionRelayCapability({
+      binding: binding(),
+      runtime: runtime(),
+      execServerUrl: "ws://127.0.0.1:43111/opaque-exec-grant",
+      codeModeHostUrl: "http://127.0.0.1:43112",
+      publicOrigin: `http://127.0.0.1:${server.port}`,
+    });
+    const socket = await opened(capability.url);
+    sockets.push(socket);
+    const closed = new Promise<string>((resolve) => socket.addEventListener("close", (event) => resolve(`closed ${event.code}`)));
+    socket.send(JSON.stringify({ id: 1, method: "initialize", params: { clientInfo: { name: "t3", version: "1.0.0" } } }));
+    await eventually(() => expect(child.received).toHaveLength(1));
+    await finishRelayInitialization(socket, child, 1);
+    socket.send(JSON.stringify({ id: 20, method: "thread/resume", params: { threadId: "provider-thread-1", cwd: "/root/work", model: "gpt-5.5" } }));
+    await eventually(() => expect(child.received.some((frame) => frame.includes('"thread/resume"'))).toBe(true));
+    const resumed = collectMessages(socket, 1);
+    child.stdout.write(`${JSON.stringify({ id: 20, result: { thread: { id: "provider-thread-1" } } })}\n`);
+    await resumed;
+
+    stall = Promise.withResolvers<void>();
+    const burst = Array.from({ length: 600 }, (_, index) =>
+      JSON.stringify({ method: "item/agentMessage/delta", params: { threadId: "provider-thread-1", turnId: "turn-1", itemId: "msg-1", delta: `d${index}` } }));
+    const forwarded = collectMessages(socket, burst.length);
+    for (const frame of burst) child.stdout.write(`${frame}\n`);
+    // The first frame waits on its authorization while the relay reads every other one.
+    await eventually(() => {
+      expect(stalled).toBeGreaterThan(0);
+      expect(child.stdout.writableLength + child.stdout.readableLength).toBe(0);
+    });
+    stall.resolve();
+    expect(await Promise.race([forwarded, closed])).toEqual(burst);
+  });
+
   test("makes capabilities one-use and rejects host account methods", async () => {
     const server = startRelayServer();
     const firstChild = fakeAppServer();

backend/src/provider-connections/codex-subscription-relay.ts

@@ -383,8 +383,14 @@ codexSubscriptionRelayRoutes.get(
       environmentBootstrap?.close();
       closeChild();
     };
+    // The sandbox's frames fail closed past the queue's limit. The app-server's
+    // are our own Codex's output, which must arrive whole: it writes bursts of
+    // hundreds of deltas faster than each frame is authorized and checked, and
+    // Bun reads a child's output whatever a pause asks, so its frames wait here
+    // without a limit. With one, 256 waiting frames closed the relay and failed
+    // a long turn as "The provider event stream closed unexpectedly".
     const clientFrames = createSerialTaskQueue(rejectRelay);
-    const serverFrames = createSerialTaskQueue(rejectRelay);
+    const serverFrames = createSerialTaskQueue(rejectRelay, { maxPendingTasks: Number.POSITIVE_INFINITY });
 
     return {
       onOpen: (_event, socket) => {

backend/src/provider-connections/serial-task-queue.ts

@@ -2,12 +2,16 @@ export interface SerialTaskQueue {
   enqueue(task: () => Promise<void>): void;
 }
 
+/** Runs tasks one at a time, in order; the first that throws ends the queue
+ *  (`onError`). Past `maxPendingTasks` waiting tasks it fails closed, for a
+ *  producer that must not build an unbounded backlog; a trusted one whose
+ *  output must arrive whole passes Number.POSITIVE_INFINITY. */
 export function createSerialTaskQueue(
   onError: (error: unknown) => void,
   options: { readonly maxPendingTasks?: number } = {},
 ): SerialTaskQueue {
   const maxPendingTasks = options.maxPendingTasks ?? 256;
-  if (!Number.isSafeInteger(maxPendingTasks) || maxPendingTasks < 1) {
+  if (!(maxPendingTasks === Number.POSITIVE_INFINITY || Number.isSafeInteger(maxPendingTasks)) || maxPendingTasks < 1) {
     throw new Error("serial task queue limit is invalid");
   }
   const pending: Array<() => Promise<void>> = [];