fix(core): pin the session to http after repeated websocket stream losses (#50031)

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

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<string | Uint8Array, AIError>()
+          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<string | Uint8Array, AIError>()
     let opened = 0