diff --git a/packages/ai/src/route/transport/websocket.ts b/packages/ai/src/route/transport/websocket.ts index bce2d28a6bc4..647c259008bc 100644 --- a/packages/ai/src/route/transport/websocket.ts +++ b/packages/ai/src/route/transport/websocket.ts @@ -215,7 +215,9 @@ export const fromWebSocket = ( ): Effect.Effect => Effect.gen(function* () { yield* waitOpen(ws, input) - const messages = yield* Queue.bounded>(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>() 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) => { diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index 590cabf3c157..ac88a28dbc2c 100644 --- a/packages/core/src/session/model-transport.ts +++ b/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(INBOUND_CAPACITY), + queue: yield* Queue.unbounded(), delivery: "send-attempted", } channel.active = active diff --git a/packages/core/test/session-model-transport.test.ts b/packages/core/test/session-model-transport.test.ts index 6933d49957f8..182e9a9e264c 100644 --- a/packages/core/test/session-model-transport.test.ts +++ b/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() - const poisoned = Deferred.makeUnsafe() 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) }), ) })