test(ai): harden websocket lifecycle coverage (#42600)

packages/ai/src/protocols/openai-responses-channel.ts

@@ -1,6 +1,6 @@
 import { AIError, TransportReason } from "../schema/index.js"
 import type { ChannelCheckpoint, ChannelObservation, WebSocketChannelDriver } from "../route/transport/index.js"
-import { Effect, Schema } from "effect"
+import { Effect, Option, Schema } from "effect"
 import * as ProviderShared from "./shared.js"
 import { OpenResponses } from "./open-responses.js"
 
@@ -47,6 +47,11 @@ const canonical = (value: unknown): string => {
     .join(",")}}`
 }
 
+const json = (value: unknown) => {
+  if (typeof value !== "string") return value
+  return Option.getOrElse(Schema.decodeUnknownOption(ProviderShared.Json)(value), () => value)
+}
+
 const comparable = (value: unknown) => {
   if (!ProviderShared.isRecord(value)) return value
   if (value.type === "message" && value.role === "assistant")
@@ -60,12 +65,11 @@ const comparable = (value: unknown) => {
       type: value.type,
       call_id: value.call_id,
       name: value.name,
-      arguments: value.arguments,
+      arguments: json(value.arguments),
     }
   if (value.type === "reasoning")
     return {
       type: value.type,
-      ...(value.id === undefined ? {} : { id: value.id }),
       summary: value.summary,
       encrypted_content: value.encrypted_content,
     }
@@ -87,7 +91,8 @@ const incremental = (
   if (canonical(invariant(request)) !== canonical(invariant(checkpoint.request))) return undefined
   const baseline = [...previousInput, ...checkpoint.output]
   if (input.length <= baseline.length) return undefined
-  if (!baseline.every((item, index) => canonical(comparable(item)) === canonical(input[index]))) return undefined
+  if (!baseline.every((item, index) => canonical(comparable(item)) === canonical(comparable(input[index]))))
+    return undefined
   return input.slice(baseline.length)
 }
 

packages/ai/test/provider/openai-responses.test.ts

@@ -413,7 +413,7 @@ describe("OpenAI Responses route", () => {
             status: "completed",
             call_id: "call_1",
             name: "weather",
-            arguments: '{"city":"Paris"}',
+            arguments: '{ "city": "Paris" }',
           },
         }),
       )
@@ -484,6 +484,53 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("continues store-false reasoning without replaying the output-only item ID", () =>
+    Effect.gen(function* () {
+      const firstInput = [{ role: "user", content: [{ type: "input_text", text: "Think" }] }]
+      const request = { type: "response.create", model: "gpt-5.2", store: false, input: firstInput }
+      const first = continuationDriver(request)
+      const create = yield* first.create(undefined)
+      yield* first.observe(
+        create,
+        ProviderShared.encodeJson({
+          type: "response.output_item.done",
+          item: {
+            type: "reasoning",
+            id: "rs_1",
+            summary: [{ type: "summary_text", text: "Thought" }],
+            encrypted_content: "encrypted",
+          },
+        }),
+      )
+      const saved = checkpoint(
+        yield* first.observe(
+          create,
+          ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_1" } }),
+        ),
+      )
+      const next = continuationDriver({
+        ...request,
+        input: [
+          ...firstInput,
+          {
+            type: "reasoning",
+            summary: [{ type: "summary_text", text: "Thought" }],
+            encrypted_content: "encrypted",
+          },
+          { role: "user", content: [{ type: "input_text", text: "Continue" }] },
+        ],
+      })
+
+      const continued = yield* next.create(saved)
+
+      expect(continued.mode).toBe("incremental")
+      expect(ProviderShared.decodeJson(continued.message)).toMatchObject({
+        previous_response_id: "resp_1",
+        input: [{ role: "user", content: [{ type: "input_text", text: "Continue" }] }],
+      })
+    }),
+  )
+
   it.effect("uses a full request when any non-input invariant changes", () =>
     Effect.gen(function* () {
       const request = {

packages/core/src/session/model-transport.ts

@@ -12,7 +12,7 @@ import {
 } from "@opencode-ai/ai/route"
 import { AIError, TransportReason, type TransportOperation } from "@opencode-ai/ai"
 import { Hash } from "@opencode-ai/util/hash"
-import { Cause, Clock, Context, Effect, Fiber, Layer, Queue, Scope, Semaphore, Stream } from "effect"
+import { Cause, Clock, Context, Effect, Fiber, Layer, Metric, Queue, Scope, Semaphore, Stream } from "effect"
 import { Socket } from "effect/unstable/socket"
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
 import { SessionSchema } from "./schema.js"
@@ -21,6 +21,12 @@ import { webSocketConstructor } from "../effect/app-node-platform.js"
 const ROTATE_AFTER_MS = 55 * 60 * 1000
 const INBOUND_CAPACITY = 128
 const IDLE_TIMEOUT = "5 minutes"
+const events = Metric.counter("opencode_session_websocket_events_total", {
+  description: "Session WebSocket lifecycle events",
+  incremental: true,
+})
+const metric = (event: string, attributes: Record<string, string> = {}) =>
+  Metric.update(events.pipe(Metric.withAttributes({ event, ...attributes })), 1)
 
 type Delivery = "queued" | "connecting" | "ready" | "send-attempted" | "provider-observed" | "terminal"
 
@@ -144,6 +150,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
               }),
             ),
           )
+        yield* metric("close")
       })
 
       const poison = Effect.fn("SessionModelTransport.poison")(function* (
@@ -156,6 +163,11 @@ export const makeLayer = (connector: WebSocketConnector) =>
         if (channel.closing) return
         channel.closing = true
         if (channel.active) Queue.failCauseUnsafe(channel.active.queue, Cause.fail(error))
+        yield* metric(
+          error.reason._tag === "Transport" && error.reason.code === "queue-overflow"
+            ? "queue_overflow"
+            : "protocol_failure",
+        )
         yield* channel.connection.close
       })
 
@@ -166,7 +178,9 @@ export const makeLayer = (connector: WebSocketConnector) =>
       ) {
         return yield* Effect.uninterruptibleMask((restore) =>
           Effect.gen(function* () {
-            const connection = yield* restore(connector.open(exchange.connect))
+            const connection = yield* restore(
+              connector.open(exchange.connect).pipe(Effect.withSpan("SessionModelTransport.connect")),
+            )
             if (owner.closed) {
               yield* connection.close
               return yield* transportError("open", "Session WebSocket owner closed while connecting", {
@@ -237,6 +251,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
               sessionTransport: "websocket",
               phase: "connect",
             })
+            yield* metric("connect")
             return channel
           }),
         )
@@ -278,6 +293,8 @@ export const makeLayer = (connector: WebSocketConnector) =>
             phase: "connect",
             reason: rotation,
           })
+          yield* metric("rotation", { reason: rotation })
+          yield* metric("reconnect")
           yield* closeChannel(owner, current)
         }
 
@@ -287,6 +304,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
             sessionTransport: "websocket",
             phase: "connect",
           })
+        if (owner.channel) yield* metric("reuse")
         const channel = owner.channel
           ? owner.channel
           : yield* open(owner, exchange, key).pipe(
@@ -298,7 +316,11 @@ export const makeLayer = (connector: WebSocketConnector) =>
                       phase: "connect",
                       delivery: "not-sent",
                       code: error.reason._tag === "Transport" ? error.reason.code : error.reason._tag,
-                    }).pipe(Effect.andThen(Effect.succeed(undefined))),
+                    }).pipe(
+                      Effect.andThen(metric("connect_failure")),
+                      Effect.andThen(metric("fallback")),
+                      Effect.andThen(Effect.succeed(undefined)),
+                    ),
               ),
             )
         if (!channel) return fallback(exchange)
@@ -318,6 +340,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
         channel.active = active
         lifecycle.delivery = "send-attempted"
         const sent = yield* channel.connection.sendText(create.message).pipe(
+          Effect.withSpan("SessionModelTransport.send"),
           Effect.onInterrupt(() => closeChannel(owner, channel)),
           Effect.result,
         )
@@ -325,9 +348,14 @@ export const makeLayer = (connector: WebSocketConnector) =>
           const failure = sent.failure
           const notSent = failure.reason._tag === "Transport" && failure.reason.delivery === "not-sent"
           yield* closeChannel(owner, channel)
-          if (notSent) return fallback(exchange)
+          if (notSent) {
+            yield* metric("fallback")
+            return fallback(exchange)
+          }
+          yield* metric("ambiguous_delivery")
           return yield* annotate(failure, { phase: "send", delivery: "ambiguous" })
         }
+        yield* metric("send")
 
         let terminal: ChannelObservation | undefined
         const token = {}
@@ -353,6 +381,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
               terminal = observation
               lifecycle.delivery = "terminal"
               staged = observation.type === "completed" ? observation.checkpoint : undefined
+              if (staged) channel.pending = { token, checkpoint: staged }
               if (observation.type !== "completed" || !staged) channel.checkpoint = undefined
             }),
           ),
@@ -364,11 +393,13 @@ export const makeLayer = (connector: WebSocketConnector) =>
               const pending = yield* Queue.size(active.queue)
               yield* Queue.shutdown(active.queue)
               if (terminal && pending === 0) {
-                if (staged) channel.pending = { token, checkpoint: staged }
+                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)
                 return
               }
+              yield* metric("cancellation")
               channel.checkpoint = undefined
               channel.pending = undefined
               const error = terminal

packages/core/test/lib/websocket-server.ts

@@ -0,0 +1,71 @@
+import { Buffer } from "node:buffer"
+import { Effect } from "effect"
+
+interface ConnectionData {
+  readonly id: number
+}
+
+export interface WebSocketServerState {
+  readonly headers: Array<Record<string, string>>
+  readonly messages: string[]
+  opens: number
+  closes: number
+  pongs: number
+}
+
+export interface WebSocketServerFixture {
+  readonly url: string
+  readonly state: WebSocketServerState
+}
+
+export interface WebSocketServerOptions {
+  readonly upgrade?: (request: Request) => boolean
+  readonly open?: (socket: Bun.ServerWebSocket<ConnectionData>) => void
+  readonly message?: (socket: Bun.ServerWebSocket<ConnectionData>, message: string | Buffer) => void
+}
+
+export const makeWebSocketServer = (options: WebSocketServerOptions = {}) =>
+  Effect.acquireRelease(
+    Effect.sync(() => {
+      const state: WebSocketServerState = { headers: [], messages: [], opens: 0, closes: 0, pongs: 0 }
+      let connection = 0
+      const server = Bun.serve<ConnectionData>({
+        hostname: "127.0.0.1",
+        port: 0,
+        fetch(request, server) {
+          state.headers.push(Object.fromEntries(request.headers.entries()))
+          if ((options.upgrade?.(request) ?? true) && server.upgrade(request, { data: { id: connection++ } }))
+            return undefined
+          return new Response("WebSocket upgrade required", {
+            status: 426,
+            headers: { "x-upgrade-rejected": "true" },
+          })
+        },
+        websocket: {
+          open(socket) {
+            state.opens++
+            options.open?.(socket)
+          },
+          message(socket, message) {
+            const text = typeof message === "string" ? message : message.toString()
+            state.messages.push(text)
+            options.message?.(socket, message)
+          },
+          close() {
+            state.closes++
+          },
+          pong() {
+            state.pongs++
+          },
+        },
+      })
+      return {
+        server,
+        fixture: {
+          url: `${server.url.toString().replace(/^http/, "ws")}responses`,
+          state,
+        } satisfies WebSocketServerFixture,
+      }
+    }),
+    ({ server }) => Effect.promise(() => server.stop(true)),
+  ).pipe(Effect.map((item) => item.fixture))

packages/core/test/session-model-transport-live.test.ts

@@ -0,0 +1,318 @@
+import { describe, expect, test } from "bun:test"
+import { NodeSocket } from "@effect/platform-node"
+import { AIError, LLM, Message } from "@opencode-ai/ai"
+import {
+  LLMClient,
+  RequestExecutor,
+  WebSocketTransport,
+  type ChannelObservation,
+  type WebSocketChannelExchange,
+} from "@opencode-ai/ai/route"
+import { configure } from "@opencode-ai/ai/providers/openai"
+import { SessionModelTransport } from "@opencode-ai/core/session/model-transport"
+import { Session } from "@opencode-ai/schema/session"
+import { Effect, Fiber, Layer, Stream } from "effect"
+import { Headers } from "effect/unstable/http"
+import { Socket } from "effect/unstable/socket"
+import { makeWebSocketServer, type WebSocketServerFixture, type WebSocketServerOptions } from "./lib/websocket-server"
+
+const sessionID = Session.ID.make("ses_live_websocket")
+
+const exchange = (server: WebSocketServerFixture, id: string): WebSocketChannelExchange => ({
+  id,
+  connect: {
+    url: server.url,
+    headers: Headers.fromInput({ authorization: "Bearer local-secret", "x-handshake": "visible" }),
+  },
+  fallback: () => Stream.die("Unexpected HTTP fallback"),
+  driver: {
+    create: () => Effect.succeed({ message: id, mode: "full" }),
+    observe: (_create, frame): Effect.Effect<ChannelObservation, AIError> =>
+      Effect.succeed({ type: "completed", frame }),
+  },
+})
+
+const withServer = <A>(
+  options: WebSocketServerOptions,
+  effect: (server: WebSocketServerFixture) => Effect.Effect<A, unknown, SessionModelTransport.Service>,
+) =>
+  Effect.runPromise(
+    Effect.gen(function* () {
+      const constructor = yield* Socket.WebSocketConstructor
+      const server = yield* makeWebSocketServer(options)
+      return yield* effect(server).pipe(
+        Effect.provide(
+          SessionModelTransport.makeLayer({
+            open: (input) =>
+              WebSocketTransport.open(input).pipe(Effect.provideService(Socket.WebSocketConstructor, constructor)),
+          }),
+        ),
+      )
+    }).pipe(Effect.scoped, Effect.provide(NodeSocket.layerWebSocketConstructorWS)),
+  )
+
+const collect = (transport: SessionModelTransport.Interface, item: WebSocketChannelExchange) =>
+  Effect.gen(function* () {
+    const execution = yield* transport.bind(sessionID).execute(item)
+    return Array.from(yield* Stream.runCollect(execution.frames.pipe(Stream.onEnd(execution.complete))))
+  }).pipe(Effect.scoped)
+
+const waitFor = (predicate: () => boolean, remaining = 100): Effect.Effect<void> => {
+  if (predicate()) return Effect.void
+  if (remaining === 0) return Effect.die("Timed out waiting for local WebSocket server")
+  return Effect.sleep("5 millis").pipe(Effect.andThen(Effect.suspend(() => waitFor(predicate, remaining - 1))))
+}
+
+describe("SessionModelTransport local WebSocket server", () => {
+  test("continues a real Responses connection with only the appended input", async () => {
+    const requests: Array<Record<string, unknown>> = []
+    await withServer(
+      {
+        message: (socket, message) => {
+          const request = JSON.parse(message.toString())
+          requests.push(request)
+          const index = requests.length
+          const id = `msg_${index}`
+          const text = index === 1 ? "Hello" : "Brief"
+          socket.send(JSON.stringify({ type: "response.created", response: { id: `resp_${index}` } }))
+          socket.send(JSON.stringify({ type: "response.output_item.added", item: { type: "message", id } }))
+          socket.send(JSON.stringify({ type: "response.output_text.delta", item_id: id, delta: text }))
+          socket.send(JSON.stringify({ type: "response.output_text.done", item_id: id, text }))
+          socket.send(
+            JSON.stringify({
+              type: "response.output_item.done",
+              item: {
+                type: "message",
+                id,
+                status: "completed",
+                role: "assistant",
+                content: [{ type: "output_text", text }],
+              },
+            }),
+          )
+          socket.send(JSON.stringify({ type: "response.completed", response: { id: `resp_${index}` } }))
+        },
+      },
+      (server) =>
+        Effect.gen(function* () {
+          const transport = yield* SessionModelTransport.Service
+          const executor = transport.bind(sessionID)
+          const model = configure({
+            baseURL: server.url.replace(/^ws/, "http").replace(/responses$/, ""),
+            apiKey: "local",
+          }).responses("gpt-5.2")
+          const client = LLMClient.Service
+          const layer = LLMClient.layer.pipe(
+            Layer.provide(
+              Layer.succeed(
+                RequestExecutor.Service,
+                RequestExecutor.Service.of({ execute: () => Effect.die("Unexpected HTTP request") }),
+              ),
+            ),
+          )
+
+          const first = yield* client
+            .use((llm) => llm.generate(LLM.request({ model, prompt: "First" }), { webSocket: executor }))
+            .pipe(Effect.provide(layer))
+          const second = yield* client
+            .use((llm) =>
+              llm.generate(
+                LLM.request({
+                  model,
+                  messages: [Message.user("First"), Message.assistant("Hello"), Message.user("Be brief")],
+                }),
+                { webSocket: executor },
+              ),
+            )
+            .pipe(Effect.provide(layer))
+
+          expect(first.text).toBe("Hello")
+          expect(second.text).toBe("Brief")
+          expect(server.state.opens).toBe(1)
+          expect(requests[1]).toMatchObject({
+            previous_response_id: "resp_1",
+            input: [{ role: "user", content: [{ type: "input_text", text: "Be brief" }] }],
+          })
+        }),
+    )
+  })
+
+  test("clears a rejected continuation and keeps one provider request per attempt", async () => {
+    const requests: Array<Record<string, unknown>> = []
+    await withServer(
+      {
+        message: (socket, message) => {
+          requests.push(JSON.parse(message.toString()))
+          const index = requests.length
+          if (index === 2) {
+            socket.send(
+              JSON.stringify({
+                type: "error",
+                error: { code: "previous_response_not_found", message: "Missing response" },
+              }),
+            )
+            return
+          }
+          const id = `msg_${index}`
+          const text = index === 1 ? "Hello" : "Recovered"
+          socket.send(JSON.stringify({ type: "response.created", response: { id: `resp_${index}` } }))
+          socket.send(JSON.stringify({ type: "response.output_item.added", item: { type: "message", id } }))
+          socket.send(JSON.stringify({ type: "response.output_text.delta", item_id: id, delta: text }))
+          socket.send(JSON.stringify({ type: "response.output_text.done", item_id: id, text }))
+          socket.send(
+            JSON.stringify({
+              type: "response.output_item.done",
+              item: {
+                type: "message",
+                id,
+                role: "assistant",
+                content: [{ type: "output_text", text }],
+              },
+            }),
+          )
+          socket.send(JSON.stringify({ type: "response.completed", response: { id: `resp_${index}` } }))
+        },
+      },
+      (server) =>
+        Effect.gen(function* () {
+          const transport = yield* SessionModelTransport.Service
+          const executor = transport.bind(sessionID)
+          const model = configure({
+            baseURL: server.url.replace(/^ws/, "http").replace(/responses$/, ""),
+            apiKey: "local",
+          }).responses("gpt-5.2")
+          const layer = LLMClient.layer.pipe(
+            Layer.provide(
+              Layer.succeed(
+                RequestExecutor.Service,
+                RequestExecutor.Service.of({ execute: () => Effect.die("Unexpected HTTP request") }),
+              ),
+            ),
+          )
+          const request = LLM.request({
+            model,
+            messages: [Message.user("First"), Message.assistant("Hello"), Message.user("Continue")],
+          })
+
+          yield* LLMClient.Service.use((llm) =>
+            llm.generate(LLM.request({ model, prompt: "First" }), { webSocket: executor }),
+          ).pipe(Effect.provide(layer))
+          const rejected = yield* LLMClient.Service.use((llm) => llm.generate(request, { webSocket: executor })).pipe(
+            Effect.provide(layer),
+            Effect.flip,
+          )
+          const recovered = yield* LLMClient.Service.use((llm) => llm.generate(request, { webSocket: executor })).pipe(
+            Effect.provide(layer),
+          )
+
+          expect(rejected.reason).toMatchObject({
+            _tag: "Transport",
+            delivery: "rejected",
+            recovery: "retry-full",
+          })
+          expect(recovered.text).toBe("Recovered")
+          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)
+        }),
+    )
+  })
+
+  // The browser-compatible client surface cannot originate ping frames, so the server sends one and verifies pong.
+  test("reuses one real connection with handshake headers and ping/pong", async () => {
+    await withServer(
+      {
+        open: (socket) => socket.ping("health"),
+        message: (socket, message) => socket.send(`completed:${message.toString()}`),
+      },
+      (server) =>
+        Effect.gen(function* () {
+          const transport = yield* SessionModelTransport.Service
+
+          expect(yield* collect(transport, exchange(server, "first"))).toEqual(["completed:first"])
+          expect(yield* collect(transport, exchange(server, "second"))).toEqual(["completed:second"])
+          yield* waitFor(() => server.state.pongs === 1)
+
+          expect(server.state.opens).toBe(1)
+          expect(server.state.messages).toEqual(["first", "second"])
+          expect(server.state.headers[0]).toMatchObject({
+            authorization: "Bearer local-secret",
+            "x-handshake": "visible",
+          })
+        }),
+    )
+  })
+
+  test("closes a real active connection on cancellation", async () => {
+    await withServer({ message: () => {} }, (server) =>
+      Effect.gen(function* () {
+        const transport = yield* SessionModelTransport.Service
+        const running = yield* collect(transport, exchange(server, "blocked")).pipe(
+          Effect.forkChild({ startImmediately: true }),
+        )
+        yield* waitFor(() => server.state.messages.length === 1)
+
+        yield* Fiber.interrupt(running)
+        yield* waitFor(() => server.state.closes === 1)
+
+        expect(server.state.messages).toEqual(["blocked"])
+      }),
+    )
+  })
+
+  test("poisons a real connection after an unsupported binary frame", async () => {
+    await withServer({ message: (socket) => socket.sendBinary(new Uint8Array([1, 2, 3])) }, (server) =>
+      Effect.gen(function* () {
+        const transport = yield* SessionModelTransport.Service
+
+        const result = yield* Effect.result(collect(transport, exchange(server, "binary")))
+        yield* waitFor(() => server.state.closes === 1)
+
+        expect(result).toMatchObject({
+          _tag: "Failure",
+          failure: { reason: { _tag: "Transport", code: "message", delivery: "accepted" } },
+        })
+      }),
+    )
+  })
+
+  test("closes a real connection after an oversized frame", async () => {
+    await withServer({ message: (socket) => socket.send("x".repeat(16 * 1024 * 1024 + 1)) }, (server) =>
+      Effect.gen(function* () {
+        const transport = yield* SessionModelTransport.Service
+
+        const result = yield* Effect.result(collect(transport, exchange(server, "oversized")))
+        yield* waitFor(() => server.state.closes === 1)
+
+        expect(result).toMatchObject({
+          _tag: "Failure",
+          failure: { reason: { _tag: "Transport", code: "message-too-large", delivery: "ambiguous" } },
+        })
+      }),
+    )
+  })
+
+  // Effect's browser-compatible constructor does not expose upgrade response bodies or headers.
+  // The real 426 fixture therefore pins the observable contract: a not-sent connect failure and one HTTP fallback.
+  test("falls back once after a real rejected upgrade", async () => {
+    let fallbacks = 0
+    await withServer({ upgrade: () => false }, (server) =>
+      Effect.gen(function* () {
+        const transport = yield* SessionModelTransport.Service
+        const item = exchange(server, "fallback")
+        const result = yield* collect(transport, {
+          ...item,
+          fallback: () => {
+            fallbacks++
+            return Stream.make("http")
+          },
+        })
+
+        expect(result).toEqual(["http"])
+        expect(fallbacks).toBe(1)
+        expect(server.state.opens).toBe(0)
+      }),
+    )
+  })
+})

packages/core/test/session-model-transport.test.ts

@@ -8,7 +8,7 @@ import type {
 } from "@opencode-ai/ai/route"
 import { SessionModelTransport } from "@opencode-ai/core/session/model-transport"
 import { Session } from "@opencode-ai/schema/session"
-import { Deferred, Effect, Fiber, Queue, Stream } from "effect"
+import { Deferred, Effect, Fiber, Metric, Queue, Stream } from "effect"
 import { TestClock } from "effect/testing"
 import { Headers } from "effect/unstable/http"
 
@@ -717,4 +717,28 @@ describe("SessionModelTransport", () => {
 
     expect(fixture.connections[0]?.closed).toBe(1)
   })
+
+  test("records metadata-only lifecycle metrics", async () => {
+    const fixture = automatic()
+
+    await run(
+      fixture.connector,
+      Effect.gen(function* () {
+        const transport = yield* SessionModelTransport.Service
+        const executor = transport.bind(session)
+        yield* collect(executor, exchange("first", { headers: { authorization: "secret-one" } }))
+        yield* collect(executor, exchange("second", { headers: { authorization: "secret-one" } }))
+        yield* collect(executor, exchange("third", { headers: { authorization: "secret-two" } }))
+
+        const snapshots = yield* Metric.snapshot
+        const lifecycle = snapshots.filter((item) => item.id === "opencode_session_websocket_events_total")
+        const names = new Set(lifecycle.map((item) => item.attributes?.event))
+        expect(Array.from(names)).toEqual(
+          expect.arrayContaining(["connect", "reuse", "rotation", "reconnect", "send", "terminal"]),
+        )
+        expect(JSON.stringify(lifecycle)).not.toContain("secret-one")
+        expect(JSON.stringify(lifecycle)).not.toContain("secret-two")
+      }),
+    )
+  })
 })

packages/core/test/session-runner.test.ts

@@ -32,8 +32,11 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
 import { Session } from "@opencode-ai/core/session"
 import { Snapshot } from "@opencode-ai/core/snapshot"
 import { SessionEvent } from "@opencode-ai/core/session/event"
+import { SessionContext } from "@opencode-ai/core/session/context"
 import { SessionInbox } from "@opencode-ai/core/session/inbox"
 import { SessionMessage } from "@opencode-ai/core/session/message"
+import { SessionModelRequest } from "@opencode-ai/core/session/model-request"
+import { SessionModelTransport } from "@opencode-ai/core/session/model-transport"
 import { Money } from "@opencode-ai/schema/money"
 import { SessionProjector } from "@opencode-ai/core/session/projector"
 import { SessionExecution } from "@opencode-ai/core/session/execution"
@@ -60,6 +63,7 @@ import {
   SessionTable,
 } from "@opencode-ai/core/session/sql"
 import { InstructionEntry } from "@opencode-ai/core/session/instruction-entry"
+import { InstructionState } from "@opencode-ai/core/session/instruction-state"
 import { SessionStore } from "@opencode-ai/core/session/store"
 import { Instructions } from "@opencode-ai/core/instructions/index"
 import { InstructionBuiltIns } from "@opencode-ai/core/instructions/builtins"
@@ -72,6 +76,7 @@ import { Location } from "@opencode-ai/core/location"
 import { Provider } from "@opencode-ai/core/provider"
 import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Scope, Stream } from "effect"
 import { TestClock } from "effect/testing"
+import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
 import { asc, desc, eq } from "drizzle-orm"
 import { testEffect } from "./lib/effect"
 import { permissionLayer } from "./lib/permission"
@@ -132,6 +137,15 @@ const testLLM = TestLLM.layer({
     }),
 })
 const client = TestLLM.clientLayer
+const closedTransports: Session.ID[] = []
+const modelTransport = Layer.succeed(
+  SessionModelTransport.Service,
+  SessionModelTransport.Service.of({
+    bind: () => ({ execute: () => Effect.die("Unexpected WebSocket execution") }),
+    close: (sessionID) => Effect.sync(() => closedTransports.push(sessionID)),
+    closeAll: Effect.void,
+  }),
+)
 const model = LanguageModel.make({ id: "fake-model", provider: "fake", route: OpenAIChat.route })
 const defaultSystem = PROMPT_DEFAULT
 const replacementModel = LanguageModel.make({ id: "replacement", provider: "fake", route: OpenAIChat.route })
@@ -373,6 +387,7 @@ const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
   [Config.node, config],
   [McpInstructions.node, mcpInstructions],
   [PluginSupervisor.node, pluginSupervisor],
+  [SessionModelTransport.node, modelTransport],
 ])
 const execution = Layer.effect(
   SessionExecution.Service,
@@ -426,6 +441,8 @@ const it = testEffect(
       ReferenceInstructions.node,
       Config.node,
       Snapshot.node,
+      SessionContext.node,
+      SessionModelRequest.node,
       SessionRunnerLLM.node,
       SessionExecution.node,
       Session.node,
@@ -445,6 +462,7 @@ const it = testEffect(
       [SessionExecution.node, execution],
       [Config.node, config],
       [PluginSupervisor.node, pluginSupervisor],
+      [SessionModelTransport.node, modelTransport],
     ],
   ).pipe(Layer.provideMerge(testLLM)),
 )
@@ -491,6 +509,7 @@ const setup = Effect.gen(function* () {
   requests = (yield* TestLLM.Service).requests
   authorizations.length = 0
   executions.length = 0
+  closedTransports.length = 0
   systemBaseline = "Initial context"
   systemRemoved = false
   systemUnavailable = false
@@ -961,6 +980,48 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("forces HTTP and triggers active request and response hooks once", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const hooks = yield* PluginHooks.Service
+      let requestTriggers = 0
+      let responseTriggers = 0
+      yield* hooks.register("session", "http.request", (event) =>
+        Effect.sync(() => {
+          requestTriggers++
+          event.request.headers.set("x-request-hook", "active")
+        }),
+      )
+      yield* hooks.register("session", "http.response", (event) =>
+        Effect.sync(() => {
+          responseTriggers++
+          event.response.headers.set("x-response-hook", "active")
+        }),
+      )
+      const context = yield* SessionContext.Service
+      const modelRequests = yield* SessionModelRequest.Service
+      const selected = yield* context.select(sessionID)
+      const database = yield* Database.Service
+      const bus = yield* Bus.Service
+      yield* InstructionState.prepare(database.db, bus, selected.instructions, sessionID)
+      const prepared = yield* modelRequests.prepare({
+        context: yield* context.load(selected),
+        step: 1,
+      })
+      const http = prepared.options.http ?? (yield* Effect.die("Expected Session HTTP middleware"))
+
+      const response = yield* http(HttpClientRequest.post("https://provider.test/responses"), (request) => {
+        expect(request.headers["x-request-hook"]).toBe("active")
+        return Effect.succeed(HttpClientResponse.fromWeb(request, new Response("network")))
+      })
+
+      expect(prepared.webSocketEligible).toBe(false)
+      expect(response.headers["x-response-hook"]).toBe("active")
+      expect(requestTriggers).toBe(1)
+      expect(responseTriggers).toBe(1)
+    }),
+  )
+
   it.effect("executes a tool renamed by a session context hook", () =>
     Effect.gen(function* () {
       const session = yield* setup
@@ -1274,6 +1335,7 @@ describe("SessionRunnerLLM", () => {
       expect((yield* session.get(sessionID)).location.directory).toBe(AbsolutePath.make("/moved"))
       expect(yield* session.inbox(sessionID)).toEqual([])
       expect(requests).toEqual([])
+      expect(closedTransports).toEqual([sessionID])
       expect(
         (yield* db
           .select({ type: EventTable.type })

packages/www/content/docs/(Configure)/providers.mdx

@@ -95,6 +95,26 @@ models, and connection continue to apply:
 
 `settings` is package-specific. A field only has an effect when the selected package supports it.
 
+### Experimental OpenAI Responses WebSocket
+
+OpenAI Responses can reuse one Session-bound WebSocket connection across sequential model calls. This transport is
+experimental and disabled by default. To test it, set the flag before starting or restarting the OpenCode service:
+
+```bash
+export OPENCODE_EXPERIMENTAL_OPENAI_RESPONSES_WEBSOCKET=true
+opencode2 service restart
+```
+
+The flag applies only to the `openai` provider using the native `openai-responses` route. Azure, xAI,
+OpenAI-compatible providers, and other protocols continue to use HTTP. A Session HTTP plugin hook also forces HTTP so
+the hook is never bypassed.
+
+OpenCode reuses a connection only while its endpoint and effective handshake headers remain unchanged. It rotates
+connections before the provider lifetime limit and never replays a request after ambiguous delivery. HTTP fallback is
+limited to failures proven to occur before the provider received the request.
+
+Unset the variable or set it to `false`, then restart the service, to return to HTTP-only execution.
+
 ### Headers and body
 
 Headers and body fields can be set at provider, model, or variant scope: