From bf441c988fb29bcf49383fc50b30444050b56488 Mon Sep 17 00:00:00 2001 From: UseAgent Date: Sat, 7 Nov 2026 00:48:58 +0530 Subject: [PATCH] Keep a Codex turn going while the relay catches up on the app-server's output The subscription relay authorizes and checks each line the host Codex app-server writes, one at a time, and the queue in front of that work failed closed once 256 lines waited. Codex writes bursts of hundreds of deltas faster than that, so a slow stretch closed the relay (1008, "serial task queue limit exceeded") and failed a healthy turn as "The provider event stream closed unexpectedly". Pausing the child's output does not help: Bun keeps delivering a child's stdout after pause(). The app-server's frames now wait without a limit; the sandbox's frames keep the fail-closed limit of 256. The new relay test holds one frame's authorization while 600 lines arrive; the old code closes the relay. --- .../codex-subscription-relay.test.ts | 53 +++++++++++++++++++ .../codex-subscription-relay.ts | 8 ++- .../provider-connections/serial-task-queue.ts | 6 ++- 3 files changed, 65 insertions(+), 2 deletions(-) diff --git a/backend/src/provider-connections/codex-subscription-relay.test.ts b/backend/src/provider-connections/codex-subscription-relay.test.ts index e6d89a6ac..00c8088ab 100644 --- a/backend/src/provider-connections/codex-subscription-relay.test.ts +++ b/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 | 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((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(); + 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(); diff --git a/backend/src/provider-connections/codex-subscription-relay.ts b/backend/src/provider-connections/codex-subscription-relay.ts index ce9d3fcf8..e9ec1c58e 100644 --- a/backend/src/provider-connections/codex-subscription-relay.ts +++ b/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) => { diff --git a/backend/src/provider-connections/serial-task-queue.ts b/backend/src/provider-connections/serial-task-queue.ts index 040d5c598..682c13adf 100644 --- a/backend/src/provider-connections/serial-task-queue.ts +++ b/backend/src/provider-connections/serial-task-queue.ts @@ -2,12 +2,16 @@ export interface SerialTaskQueue { enqueue(task: () => Promise): 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> = [];