From c92d6e1dad1db2bcc9000775d13c194679df1154 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Tue, 8 Sep 2026 18:48:21 +0530 Subject: [PATCH 1/3] chore(core): log websocket send mode --- packages/core/src/session/model-transport.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index 5877443d24b9..5d59af88470d 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -316,6 +316,11 @@ export const makeLayer = (connector: WebSocketConnector) => Effect.onInterrupt(() => closeChannel(owner, channel)), ) if (create.mode === "full") channel.checkpoint = undefined + yield* Effect.logDebug("session websocket sending", { + sessionTransport: "websocket", + phase: "send", + mode: create.mode, + }) const active: Active = { queue: yield* Queue.bounded(INBOUND_CAPACITY), delivery: "send-attempted", From 753a9e795f617bc7a0f717faff2792406a12b268 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Tue, 8 Sep 2026 19:25:57 +0530 Subject: [PATCH 2/3] fix(core): close websocket after provider error frame --- packages/core/src/session/model-transport.ts | 13 +++++++- .../core/test/session-model-transport.test.ts | 32 ++++++++++++++++++- 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index 5d59af88470d..41bee61fed79 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -145,6 +145,12 @@ export const makeLayer = (connector: WebSocketConnector) => if (owner.channel === channel) owner.channel = undefined if (channel.closing) return channel.closing = true + yield* Effect.logDebug("session websocket poisoned", { + sessionTransport: "websocket", + phase: error.reason._tag === "Transport" && error.reason.phase === "close" ? "close" : "receive", + 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)) yield* metric( error.reason._tag === "Transport" && error.reason.code === "queue-overflow" @@ -388,7 +394,12 @@ export const makeLayer = (connector: WebSocketConnector) => if (terminal && pending === 0) { yield* metric("terminal", { type: terminal.type }) if (terminal.type === "rejected") yield* metric("rejection", { recovery: terminal.recovery }) - if (terminal.type === "rejected" && terminal.recovery === "rotate-and-retry-full") + // Providers close the socket after an error frame. Drop it now so the next exchange + // reconnects instead of racing that close and failing with an ambiguous delivery. + if ( + terminal.type === "provider-failure" || + (terminal.type === "rejected" && terminal.recovery === "rotate-and-retry-full") + ) yield* closeChannel(owner, channel) return } diff --git a/packages/core/test/session-model-transport.test.ts b/packages/core/test/session-model-transport.test.ts index b86d267e42a5..5f1d8ce68aa8 100644 --- a/packages/core/test/session-model-transport.test.ts +++ b/packages/core/test/session-model-transport.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test" -import { AIError, HttpContext, TransportError } from "@opencode/ai" +import { AIError, HttpContext, InvalidRequestError, TransportError } from "@opencode/ai" import type { ChannelObservation, WebSocketChannelExchange, @@ -286,6 +286,36 @@ describe("SessionModelTransport", () => { ) }) + test("closes the connection after a provider error frame so the next call reconnects", async () => { + const fixture = automatic() + const failed: WebSocketChannelExchange = { + ...exchange("failed"), + driver: { + create: () => Effect.succeed({ message: "failed", mode: "full" }), + observe: () => + Effect.succeed({ + type: "provider-failure", + error: new AIError({ reason: new InvalidRequestError({ message: "unsupported model" }) }), + }), + }, + } + + await run( + fixture.connector, + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const executor = transport.bind(session) + const result = yield* Effect.result(collect(executor, failed)) + expect(result._tag).toBe("Failure") + expect(yield* collect(executor, exchange("next"))).toEqual(["completed:next"]) + + // The provider closes the socket after an error frame; a reused connection would race that close. + expect(fixture.connections).toHaveLength(2) + expect(fixture.connections[0]?.closed).toBe(1) + }), + ) + }) + test("reuses one physical connection for sequential Session calls", async () => { const fixture = automatic() From e028bf9b6680c8ca341a691e92162b129059cff2 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Tue, 8 Sep 2026 21:21:33 +0530 Subject: [PATCH 3/3] refactor(core): drop the websocket after every error frame --- packages/core/src/session/model-transport.ts | 12 ++++-------- .../core/test/session-model-transport-live.test.ts | 2 +- packages/core/test/session-model-transport.test.ts | 5 +++-- 3 files changed, 8 insertions(+), 11 deletions(-) diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index 41bee61fed79..b8d5ab1c289e 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -147,7 +147,6 @@ export const makeLayer = (connector: WebSocketConnector) => channel.closing = true yield* Effect.logDebug("session websocket poisoned", { sessionTransport: "websocket", - phase: error.reason._tag === "Transport" && error.reason.phase === "close" ? "close" : "receive", code: error.reason._tag === "Transport" ? error.reason.code : error.reason._tag, active: channel.active !== undefined, }) @@ -394,13 +393,10 @@ export const makeLayer = (connector: WebSocketConnector) => if (terminal && pending === 0) { yield* metric("terminal", { type: terminal.type }) if (terminal.type === "rejected") yield* metric("rejection", { recovery: terminal.recovery }) - // Providers close the socket after an error frame. Drop it now so the next exchange - // reconnects instead of racing that close and failing with an ambiguous delivery. - if ( - terminal.type === "provider-failure" || - (terminal.type === "rejected" && terminal.recovery === "rotate-and-retry-full") - ) - yield* closeChannel(owner, channel) + // The Codex backend stops serving a connection after any error frame: the next request is + // never answered and the socket dies with 1006. api.openai.com keeps it open, so reconnecting + // costs one handshake there. Drop the socket after every error so retries never race that. + if (terminal.type !== "completed" && terminal.type !== "incomplete") yield* closeChannel(owner, channel) return } yield* metric("cancellation") diff --git a/packages/core/test/session-model-transport-live.test.ts b/packages/core/test/session-model-transport-live.test.ts index 4946408e32cf..3ca601baceb6 100644 --- a/packages/core/test/session-model-transport-live.test.ts +++ b/packages/core/test/session-model-transport-live.test.ts @@ -225,7 +225,7 @@ describe("SessionModelTransport local WebSocket server", () => { expect(requests).toHaveLength(3) expect(requests[1]).toHaveProperty("previous_response_id", "resp_1") expect(requests[2]).not.toHaveProperty("previous_response_id") - expect(server.state.opens).toBe(1) + expect(server.state.opens).toBe(2) }), ) }) diff --git a/packages/core/test/session-model-transport.test.ts b/packages/core/test/session-model-transport.test.ts index 5f1d8ce68aa8..8ee4b420d6e8 100644 --- a/packages/core/test/session-model-transport.test.ts +++ b/packages/core/test/session-model-transport.test.ts @@ -243,7 +243,9 @@ describe("SessionModelTransport", () => { yield* collect(executor, item("retry")) expect(checkpoints).toEqual([undefined, candidate, undefined]) - expect(fixture.connections).toHaveLength(1) + // Error frames end the connection on some backends, so the full retry uses a fresh one. + expect(fixture.connections).toHaveLength(2) + expect(fixture.connections[0]?.closed).toBe(1) }), ) }) @@ -309,7 +311,6 @@ describe("SessionModelTransport", () => { expect(result._tag).toBe("Failure") expect(yield* collect(executor, exchange("next"))).toEqual(["completed:next"]) - // The provider closes the socket after an error frame; a reused connection would race that close. expect(fixture.connections).toHaveLength(2) expect(fixture.connections[0]?.closed).toBe(1) }),