fix(ai): remove bounded websocket inbound queues (#50026)

packages/ai/src/route/transport/websocket.ts

@@ -215,7 +215,9 @@ export const fromWebSocket = (
 ): Effect.Effect<WebSocketConnection, AIError> =>
   Effect.gen(function* () {
     yield* waitOpen(ws, input)
-    const messages = yield* Queue.bounded<string | Uint8Array, AIError | Cause.Done<void>>(128)
+    // The socket pushes frames synchronously and cannot be paused, so the hand-off to the consumer
+    // fiber must absorb whole read buffers. Bun delivers over a thousand small frames in one tick.
+    const messages = yield* Queue.unbounded<string | Uint8Array, AIError | Cause.Done<void>>()
 
     const oversized = (message: string | Uint8Array) =>
       typeof message === "string" ? new Blob([message]).size > MAX_FRAME_BYTES : message.byteLength > MAX_FRAME_BYTES
@@ -238,19 +240,7 @@ export const fromWebSocket = (
     }
     const offer = (message: string | Uint8Array) => {
       if (rejectOversized(message)) return
-      if (Queue.offerUnsafe(messages, message)) return
-      Queue.failCauseUnsafe(
-        messages,
-        Cause.fail(
-          transportError("WebSocket inbound queue overflow", {
-            body: typeof message === "string" ? message : new TextDecoder().decode(message),
-            url: input.url,
-            operation: "read",
-            code: "queue-overflow",
-            phase: "receive",
-          }),
-        ),
-      )
+      Queue.offerUnsafe(messages, message)
     }
 
     const onMessage = (event: MessageEvent) => {

packages/core/src/session/model-transport.ts

@@ -20,7 +20,6 @@ import { SessionSchema } from "./schema.js"
 import { webSocketConstructor } from "../effect/app-node-platform.js"
 
 const ROTATE_AFTER_MS = 55 * 60 * 1000
-const INBOUND_CAPACITY = 128
 const CONNECT_TIMEOUT = "10 seconds"
 const IDLE_TIMEOUT = "5 minutes"
 const events = Metric.counter("opencode_session_websocket_events_total", {
@@ -169,11 +168,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
           active: channel.active !== undefined,
         })
         if (channel.active) Queue.failCauseUnsafe(channel.active.queue, Cause.fail(error))
-        yield* metric(
-          error.reason._tag === "Transport" && error.reason.code === "queue-overflow"
-            ? "queue_overflow"
-            : "protocol_failure",
-        )
+        yield* metric("protocol_failure")
         yield* channel.connection.close
       })
 
@@ -235,14 +230,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
                       code: "message",
                       phase: "receive",
                     })
-                  if (Queue.offerUnsafe(active.queue, message)) return undefined
-                  return yield* transportError("Session WebSocket inbound queue overflow", {
-                    url: exchange.connect.url,
-                    operation: "read",
-                    code: "queue-overflow",
-                    phase: "receive",
-                    delivery: "accepted",
-                  })
+                  Queue.offerUnsafe(active.queue, message)
                 }),
               ),
               Effect.catch((error) =>
@@ -255,9 +243,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
                         phase:
                           error.reason._tag === "Transport" && error.reason.phase === "close" ? "close" : "receive",
                         delivery:
-                          channel.active?.delivery === "provider-observed" ||
-                          channel.active?.delivery === "terminal" ||
-                          (error.reason._tag === "Transport" && error.reason.code === "queue-overflow")
+                          channel.active?.delivery === "provider-observed" || channel.active?.delivery === "terminal"
                             ? "accepted"
                             : error.reason._tag === "Transport" && error.reason.code === "1009"
                               ? "rejected"
@@ -370,7 +356,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
           mode: create.mode,
         })
         const active: Active = {
-          queue: yield* Queue.bounded<string, AIError>(INBOUND_CAPACITY),
+          queue: yield* Queue.unbounded<string, AIError>(),
           delivery: "send-attempted",
         }
         channel.active = active

packages/core/test/session-model-transport.test.ts

@@ -915,24 +915,22 @@ describe("SessionModelTransport", () => {
     )
   })
 
-  test("poisons instead of dropping data when the inbound queue overflows", async () => {
+  test("buffers a synchronous burst of frames larger than any fixed capacity", async () => {
+    // Bun dispatches every frame in a read buffer in one tick; a large tool call streams thousands
+    // of small argument deltas, so the exchange must absorb the whole burst before it can consume.
+    const burst = 1500
     const messages = queue<string | Uint8Array, AIError>()
-    const poisoned = Deferred.makeUnsafe<void>()
     let closed = 0
     const connector: WebSocketConnector = {
       open: () =>
         Effect.succeed({
           sendText: () =>
-            // Hold consumption at the send boundary until the reader fills and poisons the inbound queue.
             Effect.sync(() => {
-              for (let index = 0; index <= 129; index++) Queue.offerUnsafe(messages, `frame:${index}`)
-            }).pipe(Effect.andThen(Deferred.await(poisoned))),
-          messages: Stream.fromQueue(messages).pipe(Stream.tap(() => Effect.yieldNow)),
-          close: Effect.sync(() => closed++).pipe(
-            Effect.andThen(Deferred.succeed(poisoned, undefined)),
-            Effect.andThen(Queue.shutdown(messages)),
-            Effect.asVoid,
-          ),
+              for (let index = 0; index < burst; index++) Queue.offerUnsafe(messages, `frame:${index}`)
+              Queue.offerUnsafe(messages, "completed")
+            }),
+          messages: Stream.fromQueue(messages),
+          close: Effect.sync(() => closed++).pipe(Effect.andThen(Queue.shutdown(messages)), Effect.asVoid),
         }),
     }
 
@@ -941,20 +939,21 @@ describe("SessionModelTransport", () => {
       Effect.gen(function* () {
         const transport = yield* SessionModelTransport.Service
         const item = exchange("first")
-        const result = yield* Effect.result(
-          collect(transport.bind(session), {
-            ...item,
-            driver: {
-              create: item.driver.create,
-              observe: (_create, frame) => Effect.succeed({ type: "frame" as const, frame }),
-            },
-          }),
-        )
-        expect(result).toMatchObject({
-          _tag: "Failure",
-          failure: { reason: { _tag: "Transport", code: "queue-overflow", delivery: "accepted" } },
+        const frames = yield* collect(transport.bind(session), {
+          ...item,
+          driver: {
+            create: item.driver.create,
+            observe: (_create, frame) =>
+              Effect.succeed(
+                frame === "completed" ? { type: "completed" as const, frame } : { type: "frame" as const, frame },
+              ),
+          },
         })
-        expect(closed).toBe(1)
+        expect(frames).toHaveLength(burst + 1)
+        expect(frames.slice(0, 3)).toEqual(["frame:0", "frame:1", "frame:2"])
+        expect(frames.at(-2)).toBe(`frame:${burst - 1}`)
+        expect(frames.at(-1)).toBe("completed")
+        expect(closed).toBe(0)
       }),
     )
   })