diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index ac88a28dbc2..e19f31513d3 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -20,8 +20,10 @@ import { SessionSchema } from "./schema.js" import { webSocketConstructor } from "../effect/app-node-platform.js" const ROTATE_AFTER_MS = 55 * 60 * 1000 -const CONNECT_TIMEOUT = "10 seconds" +const CONNECT_TIMEOUT = "15 seconds" const IDLE_TIMEOUT = "5 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", { description: "Session WebSocket lifecycle events", incremental: true, @@ -49,6 +51,7 @@ interface State { readonly lock: Semaphore.Semaphore closed: boolean httpFallback: boolean + streamFailures: number channel?: Channel } @@ -125,7 +128,7 @@ export const makeLayer = (connector: WebSocketConnector) => const state = (sessionID: SessionSchema.ID) => { const current = states.get(sessionID) if (current) return current - const created = { lock: Semaphore.makeUnsafe(1), closed: false, httpFallback: false } + const created = { lock: Semaphore.makeUnsafe(1), closed: false, httpFallback: false, streamFailures: 0 } states.set(sessionID, created) return created } @@ -167,11 +170,27 @@ export const makeLayer = (connector: WebSocketConnector) => code: error.reason._tag === "Transport" ? error.reason.code : error.reason._tag, active: channel.active !== undefined, }) - if (channel.active) Queue.failCauseUnsafe(channel.active.queue, Cause.fail(error)) + if (channel.active) { + Queue.failCauseUnsafe(channel.active.queue, Cause.fail(error)) + yield* streamFailure(owner) + } yield* metric("protocol_failure") yield* channel.connection.close }) + // A socket that keeps dying mid-exchange costs a retry every step; after enough consecutive + // losses the Session stays on HTTP. + const streamFailure = Effect.fn("SessionModelTransport.streamFailure")(function* (owner: State) { + owner.streamFailures++ + if (owner.streamFailures < MAX_STREAM_FAILURES) return + owner.httpFallback = true + yield* Effect.logWarning("session websocket failed repeatedly; using http", { + sessionTransport: "websocket", + failures: owner.streamFailures, + }) + yield* metric("fallback", { reason: "stream_failures" }) + }) + const open = Effect.fn("SessionModelTransport.open")(function* ( owner: State, exchange: WebSocketChannelExchange, @@ -381,6 +400,7 @@ export const makeLayer = (connector: WebSocketConnector) => return fallback(exchange) } yield* metric("ambiguous_delivery") + yield* streamFailure(owner) return yield* annotate(failure, { phase: "send", delivery: "ambiguous" }) } yield* metric("send") @@ -421,6 +441,7 @@ export const makeLayer = (connector: WebSocketConnector) => const pending = yield* Queue.size(active.queue) yield* Queue.shutdown(active.queue) if (terminal && pending === 0) { + owner.streamFailures = 0 yield* metric("terminal", { type: terminal.type }) if (terminal.type === "rejected") yield* metric("rejection", { recovery: terminal.recovery }) // The Codex backend stops serving a connection after any error frame: the next request is diff --git a/packages/core/test/session-model-transport.test.ts b/packages/core/test/session-model-transport.test.ts index 182e9a9e264..1e8d6f99f8f 100644 --- a/packages/core/test/session-model-transport.test.ts +++ b/packages/core/test/session-model-transport.test.ts @@ -593,7 +593,7 @@ describe("SessionModelTransport", () => { Effect.forkChild({ startImmediately: true }), ) yield* Effect.yieldNow - yield* TestClock.adjust("10 seconds") + yield* TestClock.adjust("15 seconds") expect(yield* Fiber.join(running)).toEqual(["fallback:slow"]) }), ) @@ -731,6 +731,40 @@ describe("SessionModelTransport", () => { ) }) + test("keeps the Session on HTTP after repeated mid-stream socket losses", async () => { + let opens = 0 + const connector: WebSocketConnector = { + open: () => + Effect.gen(function* () { + opens++ + const messages = yield* Queue.unbounded() + return { + sendText: () => + Effect.sync(() => { + Queue.failCauseUnsafe(messages, Cause.fail(error("socket dropped"))) + }), + messages: Stream.fromQueue(messages), + close: Queue.shutdown(messages).pipe(Effect.asVoid), + } + }), + } + + await run( + connector, + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const executor = transport.bind(session) + for (let attempt = 0; attempt < 5; attempt++) { + const result = yield* Effect.result(collect(executor, exchange(`attempt-${attempt}`))) + expect(result._tag).toBe("Failure") + } + expect(opens).toBe(5) + expect(yield* collect(executor, exchange("sixth"))).toEqual(["fallback:sixth"]) + expect(opens).toBe(5) + }), + ) + }) + test("falls back to HTTP after close code 1009 and keeps the Session on HTTP", async () => { const messages = queue() let opened = 0