Merge branch 'dev' into websocket-support

bun.lock

@@ -478,6 +478,7 @@
         "@solid-primitives/event-bus": "1.1.2",
         "@solid-primitives/scheduled": "1.5.2",
         "@standard-schema/spec": "1.0.0",
+        "@types/ws": "8.18.1",
         "@zip.js/zip.js": "2.7.62",
         "ai": "catalog:",
         "ai-gateway-provider": "3.1.2",
@@ -519,6 +520,7 @@
         "vscode-jsonrpc": "8.2.1",
         "web-tree-sitter": "0.25.10",
         "which": "6.0.1",
+        "ws": "8.21.0",
         "xdg-basedir": "5.1.0",
         "yargs": "18.0.0",
         "zod": "catalog:",
@@ -5138,7 +5140,7 @@
 
     "write-file-atomic": ["write-file-atomic@7.0.1", "", { "dependencies": { "signal-exit": "^4.0.1" } }, "sha512-OTIk8iR8/aCRWBqvxrzxR0hgxWpnYBblY1S5hDWBQfk/VFmJwzmJgQFN3WsoUKHISv2eAwe+PpbUzyL1CKTLXg=="],
 
-    "ws": ["ws@8.18.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw=="],
+    "ws": ["ws@8.21.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g=="],
 
     "wsl-utils": ["wsl-utils@0.3.1", "", { "dependencies": { "is-wsl": "^3.1.0", "powershell-utils": "^0.1.0" } }, "sha512-g/eziiSUNBSsdDJtCLB8bdYEUMj4jR7AGeUo96p/3dTafgjHhpF4RiCFPiRILwjQoDXx5MqkBr4fwWtR3Ky4Wg=="],
 
@@ -5496,6 +5498,8 @@
 
     "@cloudflare/kv-asset-handler/mime": ["mime@3.0.0", "", { "bin": { "mime": "cli.js" } }, "sha512-jSCU7/VB1loIWBZe14aEYHU/+1UMEHoaO7qxCOVJOw9GgH72VAWppxNcjU+x9a2k3GSIBXNKxXQFqRvvZ7vr3A=="],
 
+    "@cloudflare/vite-plugin/ws": ["ws@8.18.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw=="],
+
     "@cspotcode/source-map-support/@jridgewell/trace-mapping": ["@jridgewell/trace-mapping@0.3.9", "", { "dependencies": { "@jridgewell/resolve-uri": "^3.0.3", "@jridgewell/sourcemap-codec": "^1.4.10" } }, "sha512-3Belt6tdc8bPgAtbcmdtNJlirVoTmEb5e2gC94PnkwEW9jI6CAHUeoG85tjWP5WquqfavoMtMwiG4P926ZKKuQ=="],
 
     "@develar/schema-utils/ajv": ["ajv@6.14.0", "", { "dependencies": { "fast-deep-equal": "^3.1.1", "fast-json-stable-stringify": "^2.0.0", "json-schema-traverse": "^0.4.1", "uri-js": "^4.2.2" } }, "sha512-IWrosm/yrn43eiKqkfkHis7QioDleaXQHdDVPKg0FSwwd/DuvyX79TZnFOnYpB7dcsFAMmtFztZuXPDvSePkFw=="],
@@ -5958,6 +5962,8 @@
 
     "miniflare/undici": ["undici@7.14.0", "", {}, "sha512-Vqs8HTzjpQXZeXdpsfChQTlafcMQaaIwnGwLam1wudSSjlJeQ3bw1j+TLPePgrCnCpUXx7Ba5Pdpf5OBih62NQ=="],
 
+    "miniflare/ws": ["ws@8.18.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw=="],
+
     "miniflare/zod": ["zod@3.22.3", "", {}, "sha512-EjIevzuJRiRPbVH4mGc8nApb/lVLKVpmUhAaR5R5doKGfAnGJ6Gr3CViAVjP+4FWSxCsybeWQdcgCtbX+7oZug=="],
 
     "minipass-flush/minipass": ["minipass@3.3.6", "", { "dependencies": { "yallist": "^4.0.0" } }, "sha512-DxiNidxSEK+tHG6zOIklvNOwm3hvCrbUrdtzY74U6HKTJxvIDfOUL5W5P2Ghd3DTkhhKPYGqeNUIh5qcM4YBfw=="],
@@ -6074,6 +6080,8 @@
 
     "storybook/open": ["open@10.2.0", "", { "dependencies": { "default-browser": "^5.2.1", "define-lazy-prop": "^3.0.0", "is-inside-container": "^1.0.0", "wsl-utils": "^0.1.0" } }, "sha512-YgBpdJHPyQ2UE5x+hlSXcnejzAvD0b22U2OuAP+8OnlJT+PjWPxtgmGqKKc+RgTM63U9gN0YzrYc71R2WT/hTA=="],
 
+    "storybook/ws": ["ws@8.18.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw=="],
+
     "storybook-solidjs-vite/vite-plugin-solid": ["vite-plugin-solid@2.11.12", "", { "dependencies": { "@babel/core": "^7.23.3", "@types/babel__core": "^7.20.4", "babel-preset-solid": "^1.8.4", "merge-anything": "^5.1.7", "solid-refresh": "^0.6.3", "vitefu": "^1.0.4" }, "peerDependencies": { "@testing-library/jest-dom": "^5.16.6 || ^5.17.0 || ^6.*", "solid-js": "^1.7.2", "vite": "^3.0.0 || ^4.0.0 || ^5.0.0 || ^6.0.0 || ^7.0.0 || ^8.0.0" }, "optionalPeers": ["@testing-library/jest-dom"] }, "sha512-FgjPcx2OwX9h6f28jli7A4bG7PP3te8uyakE5iqsmpq3Jqi1TWLgSroC9N6cMfGRU2zXsl4Q6ISvTr2VL0QHpA=="],
 
     "string-width-cjs/emoji-regex": ["emoji-regex@8.0.0", "", {}, "sha512-MSjYzcWNOA0ewAHpz0MxpYFvwg6yjy1NG3xteoqz644VCo/RPgnr1/GGt+ic3iJTzQ8Eu3TdM14SawnVUmGE6A=="],

packages/opencode/package.json

@@ -122,6 +122,7 @@
     "@solid-primitives/event-bus": "1.1.2",
     "@solid-primitives/scheduled": "1.5.2",
     "@standard-schema/spec": "1.0.0",
+    "@types/ws": "8.18.1",
     "@zip.js/zip.js": "2.7.62",
     "ai": "catalog:",
     "ai-gateway-provider": "3.1.2",
@@ -163,6 +164,7 @@
     "vscode-jsonrpc": "8.2.1",
     "web-tree-sitter": "0.25.10",
     "which": "6.0.1",
+    "ws": "8.21.0",
     "xdg-basedir": "5.1.0",
     "yargs": "18.0.0",
     "zod": "catalog:"

packages/opencode/src/plugin/index.ts

@@ -10,7 +10,7 @@ import { Bus } from "../bus"
 import * as Log from "@opencode-ai/core/util/log"
 import { createOpencodeClient } from "@opencode-ai/sdk"
 import { ServerAuth } from "@/server/auth"
-import { CodexAuthPlugin } from "./codex"
+import { CodexAuthPlugin } from "./openai/codex"
 import { Session } from "@/session/session"
 import { NamedError } from "@opencode-ai/core/util/error"
 import { CopilotAuthPlugin } from "./github-copilot/copilot"

packages/opencode/src/plugin/openai/codex.ts

@@ -1,10 +1,11 @@
 import type { Hooks, PluginInput } from "@opencode-ai/plugin"
 import * as Log from "@opencode-ai/core/util/log"
 import { InstallationVersion } from "@opencode-ai/core/installation/version"
-import { OAUTH_DUMMY_KEY } from "../auth"
+import { OAUTH_DUMMY_KEY } from "../../auth"
 import os from "os"
 import { setTimeout as sleep } from "node:timers/promises"
 import { createServer } from "http"
+import { OpenAIWebSocketPool } from "./ws-pool"
 
 const log = Log.create({ service: "plugin.codex" })
 
@@ -371,8 +372,14 @@ function waitForOAuthCallback(pkce: PkceCodes, state: string): Promise<TokenResp
 export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPluginOptions = {}): Promise<Hooks> {
   const issuer = options.issuer ?? ISSUER
   const codexApiEndpoint = options.codexApiEndpoint ?? CODEX_API_ENDPOINT
+  let websocketFetchInstalled = false
+  const websocketFetches: Array<ReturnType<typeof OpenAIWebSocketPool.createWebSocketFetch>> = []
 
   return {
+    async dispose() {
+      for (const websocketFetch of websocketFetches) websocketFetch.close()
+      websocketFetches.length = 0
+    },
     provider: {
       id: "openai",
       async models(provider, ctx) {
@@ -410,7 +417,10 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug
       provider: "openai",
       async loader(getAuth) {
         const auth = await getAuth()
-        if (auth.type !== "oauth") return {}
+        const websocketFetch = OpenAIWebSocketPool.createWebSocketFetch({ httpFetch: fetch })
+        websocketFetches.push(websocketFetch)
+        websocketFetchInstalled = true
+        if (auth.type !== "oauth") return { fetch: websocketFetch }
 
         let refreshPromise:
           | Promise<{
@@ -436,7 +446,7 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug
             }
 
             const currentAuth = await getAuth()
-            if (currentAuth.type !== "oauth") return fetch(requestInput, init)
+            if (currentAuth.type !== "oauth") return websocketFetch(requestInput, init)
 
             // Cast to include accountId field
             const authWithAccount = currentAuth as typeof currentAuth & { accountId?: string }
@@ -507,10 +517,12 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug
                 ? new URL(codexApiEndpoint)
                 : parsed
 
-            return fetch(url, {
+            const requestInit = {
               ...init,
               headers,
-            })
+            }
+            if (parsed.pathname.includes("/v1/responses")) return websocketFetch(url, requestInit)
+            return fetch(url, OpenAIWebSocketPool.withoutInternalHeaders(requestInit))
           },
         }
       },
@@ -639,6 +651,10 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug
       output.headers.originator = "opencode"
       output.headers["User-Agent"] = `opencode/${InstallationVersion} (${os.platform()} ${os.release()}; ${os.arch()})`
       output.headers["session-id"] = input.sessionID
+      // Temporary fetch-layer hack: title generation currently shares the conversation
+      // session ID, so the OpenAI plugin marks it for HTTP fallback until transport
+      // context can be passed directly instead of smuggled through headers.
+      if (websocketFetchInstalled && input.agent === "title") output.headers[OpenAIWebSocketPool.TITLE_HEADER] = "true"
     },
     "chat.params": async (input, output) => {
       if (input.model.providerID !== "openai") return

packages/opencode/src/plugin/openai/ws-pool.ts

@@ -0,0 +1,226 @@
+import WebSocket from "ws"
+import * as Log from "@opencode-ai/core/util/log"
+import { OpenAIWebSocket } from "./ws"
+
+export const TITLE_HEADER = "x-opencode-title"
+
+const log = Log.create({ service: "plugin.openai.ws" })
+
+export interface CreateWebSocketFetchOptions {
+  httpFetch?: typeof globalThis.fetch
+  url?: string
+  connectTimeout?: number
+  idleTimeout?: number
+  maxConnectionAge?: number
+}
+
+interface PoolEntry {
+  socket?: WebSocket
+  connectedAt?: number
+  lastUsedAt: number
+  busy: boolean
+  fallback: boolean
+}
+
+const DEFAULT_CONNECT_TIMEOUT = 15_000
+const DEFAULT_IDLE_TIMEOUT = 5 * 60 * 1000
+const DEFAULT_MAX_CONNECTION_AGE = 55 * 60 * 1000
+
+export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
+  const httpFetch = options?.httpFetch ?? globalThis.fetch
+  const pool = new Map<string, PoolEntry>()
+  const connectTimeout = options?.connectTimeout ?? DEFAULT_CONNECT_TIMEOUT
+  const idleTimeout = options?.idleTimeout ?? DEFAULT_IDLE_TIMEOUT
+  const maxConnectionAge = options?.maxConnectionAge ?? DEFAULT_MAX_CONNECTION_AGE
+  const pruneTimer = setInterval(() => prune(), Math.min(idleTimeout, 60_000))
+  if (typeof pruneTimer === "object" && "unref" in pruneTimer && typeof pruneTimer.unref === "function") {
+    pruneTimer.unref()
+  }
+
+  async function websocketFetch(input: RequestInfo | URL, init?: RequestInit): Promise<Response> {
+    const url = requestUrl(input)
+    const internalHeaders = OpenAIWebSocket.normalizeHeaders(init?.headers)
+    const httpInit = withoutInternalHeaders(init)
+
+    if (init?.method !== "POST" || !new URL(url).pathname.endsWith("/responses")) {
+      return httpFetch(input, httpInit)
+    }
+
+    const body = parseBody(init?.body)
+    if (!body?.stream) return httpFetch(input, httpInit)
+    if (internalHeaders[TITLE_HEADER] === "true") {
+      log.debug("http fallback", { reason: "title" })
+      return httpFetch(input, httpInit)
+    }
+
+    const key = poolKey(internalHeaders)
+    if (!key) {
+      log.debug("http fallback", { reason: "missing_session" })
+      return httpFetch(input, httpInit)
+    }
+
+    const entry = pool.get(key) ?? { lastUsedAt: Date.now(), busy: false, fallback: false }
+    pool.set(key, entry)
+
+    if (entry.fallback) {
+      log.debug("http fallback", { key, reason: "fallback_active" })
+      return httpFetch(input, httpInit)
+    }
+    if (entry.busy) {
+      log.debug("http fallback", { key, reason: "busy" })
+      return httpFetch(input, httpInit)
+    }
+
+    try {
+      entry.socket = await socket(
+        entry,
+        options?.url ?? url,
+        OpenAIWebSocket.normalizeHeaders(httpInit?.headers),
+        connectTimeout,
+        maxConnectionAge,
+        init?.signal,
+      )
+      entry.busy = true
+      entry.lastUsedAt = Date.now()
+      return OpenAIWebSocket.streamResponsesWebSocket({
+        socket: entry.socket,
+        body,
+        signal: init?.signal ?? undefined,
+        onTerminal: (event) => {
+          entry.busy = false
+          entry.lastUsedAt = Date.now()
+          if (event.type !== "response.completed" && event.type !== "response.done") {
+            log.warn("websocket terminal failure", { key, type: event.type })
+            invalidate(entry)
+          }
+        },
+        onConnectionInvalid: (error) => {
+          log.warn("websocket invalidated", { key, error: errorMessage(error) })
+          entry.busy = false
+          entry.fallback = true
+          invalidate(entry)
+        },
+        onAbort: () => {
+          log.debug("websocket aborted", { key })
+          entry.busy = false
+          entry.lastUsedAt = Date.now()
+          invalidate(entry)
+        },
+      })
+    } catch (error) {
+      if (OpenAIWebSocket.isAbortError(error)) {
+        invalidate(entry)
+        throw error
+      }
+
+      entry.fallback = true
+      log.warn("websocket setup failed", { key, error: errorMessage(error), fallback: "http" })
+      invalidate(entry)
+      return httpFetch(input, httpInit)
+    }
+  }
+
+  function prune() {
+    const now = Date.now()
+    for (const [key, entry] of pool) {
+      if (entry.busy) continue
+      if (now - entry.lastUsedAt < idleTimeout) continue
+      log.debug("websocket idle prune", { key })
+      invalidate(entry)
+      pool.delete(key)
+    }
+  }
+
+  function close() {
+    log.debug("websocket pool close", { count: pool.size })
+    clearInterval(pruneTimer)
+    for (const entry of pool.values()) invalidate(entry)
+    pool.clear()
+  }
+
+  return Object.assign(websocketFetch, { close })
+}
+
+async function socket(
+  entry: PoolEntry,
+  url: string,
+  headers: Record<string, string>,
+  connectTimeout: number,
+  maxConnectionAge: number,
+  signal?: AbortSignal | null,
+) {
+  if (entry.socket?.readyState === WebSocket.OPEN && entry.connectedAt && Date.now() - entry.connectedAt < maxConnectionAge) {
+    return entry.socket
+  }
+
+  invalidate(entry)
+  const next = await OpenAIWebSocket.connectResponsesWebSocket({
+    url: OpenAIWebSocket.toWebSocketUrl(url),
+    headers,
+    timeout: connectTimeout,
+    signal: signal ?? undefined,
+  })
+  entry.connectedAt = Date.now()
+  return next
+}
+
+function invalidate(entry: PoolEntry) {
+  if (entry.socket) {
+    entry.socket.terminate()
+    entry.socket = undefined
+  }
+  entry.connectedAt = undefined
+}
+
+function poolKey(headers: Record<string, string>) {
+  const sessionID = headers["x-session-affinity"] ?? headers["session-id"]
+  if (!sessionID) return undefined
+
+  return [
+    sessionID,
+    headers[TITLE_HEADER] === "true" ? "title" : "conversation",
+  ].join(":")
+}
+
+export function withoutInternalHeaders<T extends { headers?: HeadersInit }>(init: T | undefined): T | undefined {
+  if (!init?.headers) return init
+  return {
+    ...init,
+    headers: stripInternalHeaders(init.headers),
+  }
+}
+
+function stripInternalHeaders(headers: HeadersInit): HeadersInit {
+  if (headers instanceof Headers) {
+    const next = new Headers(headers)
+    next.delete(TITLE_HEADER)
+    return next
+  }
+
+  if (Array.isArray(headers)) return headers.filter((item) => !isInternalHeader(item[0]))
+  return Object.fromEntries(Object.entries(headers).filter(([key]) => !isInternalHeader(key)))
+}
+
+function isInternalHeader(key: string) {
+  return key.toLowerCase() === TITLE_HEADER
+}
+
+function requestUrl(input: RequestInfo | URL) {
+  return input instanceof URL ? input.toString() : typeof input === "string" ? input : input.url
+}
+
+function parseBody(body: BodyInit | null | undefined): Record<string, unknown> | undefined {
+  try {
+    if (typeof body !== "string") return undefined
+    const parsed = JSON.parse(body)
+    return typeof parsed === "object" && parsed !== null ? parsed : undefined
+  } catch {
+    return undefined
+  }
+}
+
+function errorMessage(error: unknown) {
+  return error instanceof Error ? error.message : String(error)
+}
+
+export * as OpenAIWebSocketPool from "./ws-pool"

packages/opencode/src/plugin/openai/ws.ts

@@ -0,0 +1,234 @@
+// Low-level OpenAI Responses WebSocket protocol helpers. Session pooling,
+// fallback, and continuation state intentionally live above this file.
+
+import WebSocket from "ws"
+
+export const PROTOCOL_HEADER = "responses_websockets=2026-02-06"
+
+export interface ConnectResponsesWebSocketOptions {
+  url: string
+  headers: Record<string, string>
+  timeout?: number
+  signal?: AbortSignal
+}
+
+export interface StreamResponsesWebSocketOptions {
+  socket: WebSocket
+  body: Record<string, unknown>
+  signal?: AbortSignal
+  onComplete?: (event: Record<string, unknown>) => void
+  onTerminal?: (event: Record<string, unknown>) => void
+  onConnectionInvalid?: (error: Error) => void
+  onAbort?: (error: Error) => void
+}
+
+export function toWebSocketUrl(url: string) {
+  return url.replace(/^http/, "ws")
+}
+
+export function normalizeHeaders(headers: HeadersInit | undefined): Record<string, string> {
+  const result: Record<string, string> = {}
+  if (!headers) return result
+
+  if (headers instanceof Headers) {
+    headers.forEach((value, key) => {
+      result[key.toLowerCase()] = value
+    })
+    return result
+  }
+
+  if (Array.isArray(headers)) {
+    for (const [key, value] of headers) {
+      result[key.toLowerCase()] = value
+    }
+    return result
+  }
+
+  for (const [key, value] of Object.entries(headers)) {
+    if (value != null) result[key.toLowerCase()] = value
+  }
+  return result
+}
+
+export function isAbortError(error: unknown) {
+  return error instanceof DOMException && error.name === "AbortError"
+}
+
+export function connectResponsesWebSocket(options: ConnectResponsesWebSocketOptions) {
+  return new Promise<WebSocket>((resolve, reject) => {
+    if (options.signal?.aborted) {
+      reject(abortError(options.signal))
+      return
+    }
+
+    const headers: Record<string, string> = {
+      ...options.headers,
+      "openai-beta": options.headers["openai-beta"] ?? PROTOCOL_HEADER,
+    }
+    delete headers["content-length"]
+
+    const socket = new WebSocket(options.url, { headers })
+    const timeout = options.timeout
+      ? setTimeout(() => {
+          cleanup()
+          socket.terminate()
+          reject(new Error("WebSocket connect timed out"))
+        }, options.timeout)
+      : undefined
+
+    function cleanup() {
+      if (timeout) clearTimeout(timeout)
+      socket.off("open", onOpen)
+      socket.off("error", onError)
+      socket.off("close", onClose)
+      options.signal?.removeEventListener("abort", onAbort)
+    }
+
+    function onOpen() {
+      cleanup()
+      resolve(socket)
+    }
+
+    function onError(error: Error) {
+      cleanup()
+      reject(error)
+    }
+
+    function onClose(code: number, reason: Buffer) {
+      cleanup()
+      reject(closeError("WebSocket closed before open", code, reason))
+    }
+
+    function onAbort() {
+      cleanup()
+      socket.terminate()
+      reject(abortError(options.signal))
+    }
+
+    socket.once("open", onOpen)
+    socket.once("error", onError)
+    socket.once("close", onClose)
+    options.signal?.addEventListener("abort", onAbort, { once: true })
+  })
+}
+
+export function streamResponsesWebSocket(options: StreamResponsesWebSocketOptions) {
+  const encoder = new TextEncoder()
+  let completed = false
+
+  return new Response(
+    new ReadableStream<Uint8Array>({
+      start(controller) {
+        function cleanup() {
+          options.socket.off("message", onMessage)
+          options.socket.off("error", onError)
+          options.socket.off("close", onClose)
+          options.signal?.removeEventListener("abort", onAbort)
+        }
+
+        function closeCompleted() {
+          cleanup()
+          controller.enqueue(encoder.encode("data: [DONE]\n\n"))
+          controller.close()
+        }
+
+        function invalidate(error: Error) {
+          cleanup()
+          options.onConnectionInvalid?.(error)
+          controller.error(error)
+        }
+
+        function onMessage(data: WebSocket.RawData, isBinary: boolean) {
+          if (isBinary) {
+            invalidate(new Error("Unexpected binary WebSocket frame"))
+            return
+          }
+
+          const text = data.toString()
+          controller.enqueue(encoder.encode(`${text.split(/\r?\n/).map((line) => `data: ${line}`).join("\n")}\n\n`))
+
+          const event = parseEvent(text)
+          if (!event) return
+
+          if (event.type === "response.completed" || event.type === "response.done") {
+            completed = true
+            options.onComplete?.(event)
+            options.onTerminal?.(event)
+            closeCompleted()
+            return
+          }
+
+          if (event.type === "response.failed" || event.type === "response.incomplete" || event.type === "error") {
+            completed = true
+            options.onTerminal?.(event)
+            closeCompleted()
+          }
+        }
+
+        function onError(error: Error) {
+          invalidate(error)
+        }
+
+        function onClose(code: number, reason: Buffer) {
+          if (completed) return
+          invalidate(closeError("WebSocket closed before response.completed", code, reason))
+        }
+
+        function onAbort() {
+          const error = abortError(options.signal)
+          cleanup()
+          options.onAbort?.(error)
+          controller.error(error)
+        }
+
+        options.socket.on("message", onMessage)
+        options.socket.once("error", onError)
+        options.socket.once("close", onClose)
+        options.signal?.addEventListener("abort", onAbort, { once: true })
+
+        if (options.signal?.aborted) {
+          onAbort()
+          return
+        }
+
+        options.socket.send(JSON.stringify(responseCreate(options.body)), (error) => {
+          if (!error) return
+          invalidate(error)
+        })
+      },
+    }),
+    {
+      status: 200,
+      headers: { "content-type": "text/event-stream" },
+    },
+  )
+}
+
+function responseCreate(body: Record<string, unknown>) {
+  const { stream: _stream, background: _background, ...payload } = body
+  return { type: "response.create", ...payload }
+}
+
+function parseEvent(text: string): Record<string, unknown> | undefined {
+  try {
+    const event = JSON.parse(text)
+    return typeof event === "object" && event !== null ? event : undefined
+  } catch {
+    return undefined
+  }
+}
+
+function abortError(signal: AbortSignal | undefined) {
+  const reason = signal?.reason
+  if (isAbortError(reason)) return reason
+  return new DOMException(reason instanceof Error ? reason.message : "Aborted", "AbortError")
+}
+
+function closeError(message: string, code: number, reason: Buffer) {
+  const details = [`code ${code}`]
+  if (code === 1009) details.push("message too big")
+  if (reason.length > 0) details.push(reason.toString())
+  return new Error(`${message} (${details.join(": ")})`)
+}
+
+export * as OpenAIWebSocket from "./ws"

packages/opencode/test/plugin/codex.test.ts

@@ -5,7 +5,7 @@ import {
   extractAccountIdFromClaims,
   extractAccountId,
   type IdTokenClaims,
-} from "../../src/plugin/codex"
+} from "../../src/plugin/openai/codex"
 
 function createTestJwt(payload: object): string {
   const header = Buffer.from(JSON.stringify({ alg: "none" })).toString("base64url")