fix(core): relax websocket idle timeout and honor chunkTimeout (#50914)

This commit is contained in:
Shoubhit Dash
2026-09-23 20:22:41 +05:30
committed by GitHub
Unverified
parent 681ea07a92
commit affa57e40f
4 changed files with 41 additions and 24 deletions
+3
View File
@@ -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({
+22 -18
View File
@@ -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 {
+15 -5
View File
@@ -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<void>
readonly closeAll: Effect.Effect<void>
}
@@ -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
@@ -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({