diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index 5877443d24b9..b8d5ab1c289e 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -145,6 +145,11 @@ 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", + 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" @@ -316,6 +321,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", @@ -383,8 +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 }) - if (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 b86d267e42a5..8ee4b420d6e8 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, @@ -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) }), ) }) @@ -286,6 +288,35 @@ 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"]) + + 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()