From affa57e40f44fbd294c18199ecab5879c0ae0d70 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Wed, 23 Sep 2026 20:22:41 +0530 Subject: [PATCH] fix(core): relax websocket idle timeout and honor chunkTimeout (#50914) --- packages/core/src/model-resolver.ts | 3 ++ packages/core/src/session/model-request.ts | 40 ++++++++++--------- packages/core/src/session/model-transport.ts | 20 +++++++--- .../core/test/session-model-transport.test.ts | 2 +- 4 files changed, 41 insertions(+), 24 deletions(-) diff --git a/packages/core/src/model-resolver.ts b/packages/core/src/model-resolver.ts index 93721e0a5ca..56428b0cb5d 100644 --- a/packages/core/src/model-resolver.ts +++ b/packages/core/src/model-resolver.ts @@ -122,6 +122,8 @@ export interface Resolved { readonly compaction?: Provider.Compaction /** Provider transport policy; omitted means HTTP. */ readonly transport?: Provider.Transport + /** Milliseconds without streamed data before a WebSocket exchange fails. */ + readonly chunkTimeout?: number } export interface Interface { @@ -388,6 +390,7 @@ export const layer = Layer.effect( limit: selected.limit, compaction: runtimeInfo.settings?.compaction, transport: provider?.settings?.transport, + chunkTimeout: provider?.settings?.chunkTimeout, } }) return Service.of({ diff --git a/packages/core/src/session/model-request.ts b/packages/core/src/session/model-request.ts index a5e464d0d24..d2277ce8ac7 100644 --- a/packages/core/src/session/model-request.ts +++ b/packages/core/src/session/model-request.ts @@ -325,24 +325,28 @@ export const layer = Layer.effect( // which transport actually carries the request, so both hook families are always offered. const webSocket = input.webSocket === "session" && model.transport === "websocket" - ? transport.bind(session.id, { - handshake: (connect) => - hooks - .trigger("session", "experimental.ws.handshake", { - ...scope, - url: connect.url, - headers: connect.headers, - }) - .pipe(Effect.map((event) => ({ url: event.url, headers: event.headers }))), - send: (frame) => - hooks - .trigger("session", "experimental.ws.send", { ...scope, frame }) - .pipe(Effect.map((event) => event.frame)), - receive: (frame) => - hooks - .trigger("session", "experimental.ws.receive", { ...scope, frame }) - .pipe(Effect.map((event) => event.frame)), - }) + ? transport.bind( + session.id, + { + handshake: (connect) => + hooks + .trigger("session", "experimental.ws.handshake", { + ...scope, + url: connect.url, + headers: connect.headers, + }) + .pipe(Effect.map((event) => ({ url: event.url, headers: event.headers }))), + send: (frame) => + hooks + .trigger("session", "experimental.ws.send", { ...scope, frame }) + .pipe(Effect.map((event) => event.frame)), + receive: (frame) => + hooks + .trigger("session", "experimental.ws.receive", { ...scope, frame }) + .pipe(Effect.map((event) => event.frame)), + }, + model.chunkTimeout, + ) : undefined return { diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index e19f31513d3..638ef66af4d 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -21,7 +21,8 @@ import { webSocketConstructor } from "../effect/app-node-platform.js" const ROTATE_AFTER_MS = 55 * 60 * 1000 const CONNECT_TIMEOUT = "15 seconds" -const IDLE_TIMEOUT = "5 minutes" +// Reasoning models can stream nothing for several minutes while still working. +const IDLE_TIMEOUT = "30 minutes" /** Consecutive exchanges lost to the socket before the Session stays on HTTP. */ const MAX_STREAM_FAILURES = 5 const events = Metric.counter("opencode_session_websocket_events_total", { @@ -72,7 +73,11 @@ export interface Interceptor { } export interface Interface { - readonly bind: (sessionID: SessionSchema.ID, interceptor?: Interceptor) => WebSocketChannelExecutor + readonly bind: ( + sessionID: SessionSchema.ID, + interceptor?: Interceptor, + idleTimeout?: number, + ) => WebSocketChannelExecutor readonly close: (sessionID: SessionSchema.ID) => Effect.Effect readonly closeAll: Effect.Effect } @@ -291,6 +296,7 @@ export const makeLayer = (connector: WebSocketConnector) => owner: State, input: WebSocketChannelExchange, interceptor?: Interceptor, + idleTimeout?: number, ) { if (owner.closed) return yield* transportError("Session WebSocket owner is closed", { @@ -409,7 +415,7 @@ export const makeLayer = (connector: WebSocketConnector) => const token = {} const frames = Stream.fromQueue(active.queue).pipe( Stream.timeoutOrElse({ - duration: IDLE_TIMEOUT, + duration: idleTimeout ?? IDLE_TIMEOUT, orElse: () => Stream.fail( transportError("Timed out waiting for WebSocket data", { @@ -500,7 +506,11 @@ export const makeLayer = (connector: WebSocketConnector) => return { frames, complete, http: channel.connection.http } }) - const bind = (sessionID: SessionSchema.ID, interceptor?: Interceptor): WebSocketChannelExecutor => ({ + const bind = ( + sessionID: SessionSchema.ID, + interceptor?: Interceptor, + idleTimeout?: number, + ): WebSocketChannelExecutor => ({ execute: (exchange) => { const owner = state(sessionID) let execution: WebSocketChannelExecution | undefined @@ -510,7 +520,7 @@ export const makeLayer = (connector: WebSocketConnector) => }, frames: Stream.unwrap( Effect.acquireRelease(owner.lock.take(1), () => owner.lock.release(1), { interruptible: true }).pipe( - Effect.andThen(start(owner, exchange, interceptor)), + Effect.andThen(start(owner, exchange, interceptor, idleTimeout)), Effect.tap((started) => Effect.sync(() => { execution = started diff --git a/packages/core/test/session-model-transport.test.ts b/packages/core/test/session-model-transport.test.ts index 1e8d6f99f8f..e38c3fda60e 100644 --- a/packages/core/test/session-model-transport.test.ts +++ b/packages/core/test/session-model-transport.test.ts @@ -622,7 +622,7 @@ describe("SessionModelTransport", () => { yield* Deferred.await(started) yield* Effect.yieldNow - yield* TestClock.adjust("5 minutes") + yield* TestClock.adjust("30 minutes") const result = yield* Effect.result(Fiber.join(running)) expect(result).toMatchObject({