From c9aef0e5958ab7c8a67c8bb8086daefc1e7d1f3c Mon Sep 17 00:00:00 2001 From: be0hhh Date: Fri, 24 Apr 2026 18:44:59 +0000 Subject: [PATCH] feat(codex): support GPT-5.5 responses websocket (#1573) Integrated into release/v3.7.0 --- bin/omniroute.mjs | 3 +- next.config.mjs | 5 + open-sse/config/providerRegistry.ts | 2 +- open-sse/executors/codex.ts | 228 ++++++- package.json | 1 + scripts/pack-artifact-policy.ts | 5 + scripts/prepublish.ts | 11 + scripts/responses-ws-proxy.mjs | 597 ++++++++++++++++++ scripts/run-next.mjs | 9 + scripts/standalone-server-ws.mjs | 70 ++ scripts/validate-pack-artifact.ts | 18 +- .../cli-tools/components/CodexToolCard.tsx | 8 +- .../api/internal/codex-responses-ws/route.ts | 185 ++++++ src/shared/validation/schemas.ts | 2 +- tests/unit/cli-model-config-schema.test.ts | 19 + tests/unit/executor-codex.test.ts | 37 ++ tests/unit/plan3-p0.test.ts | 12 + tests/unit/responses-ws-proxy.test.mjs | 159 +++++ 18 files changed, 1362 insertions(+), 9 deletions(-) create mode 100644 scripts/responses-ws-proxy.mjs create mode 100644 scripts/standalone-server-ws.mjs create mode 100644 src/app/api/internal/codex-responses-ws/route.ts create mode 100644 tests/unit/cli-model-config-schema.test.ts create mode 100644 tests/unit/responses-ws-proxy.test.mjs diff --git a/bin/omniroute.mjs b/bin/omniroute.mjs index a3f77500a2..2cc86f6554 100644 --- a/bin/omniroute.mjs +++ b/bin/omniroute.mjs @@ -169,7 +169,8 @@ if (!nodeSupport.nodeCompatible) { `); } -const serverJs = join(APP_DIR, "server.js"); +const serverWsJs = join(APP_DIR, "server-ws.mjs"); +const serverJs = existsSync(serverWsJs) ? serverWsJs : join(APP_DIR, "server.js"); if (!existsSync(serverJs)) { console.error("\x1b[31m✖ Server not found at:\x1b[0m", serverJs); diff --git a/next.config.mjs b/next.config.mjs index 4e66ad3db3..a359da869d 100644 --- a/next.config.mjs +++ b/next.config.mjs @@ -1,19 +1,24 @@ import createNextIntlPlugin from "next-intl/plugin"; +import { dirname } from "node:path"; +import { fileURLToPath } from "node:url"; const withNextIntl = createNextIntlPlugin("./src/i18n/request.ts"); const distDir = process.env.NEXT_DIST_DIR || ".next"; +const projectRoot = dirname(fileURLToPath(import.meta.url)); /** @type {import('next').NextConfig} */ const nextConfig = { distDir, // Turbopack config: redirect native modules to stubs at build time turbopack: { + root: projectRoot, resolveAlias: { // Point mitm/manager to a stub during build (native child_process/fs can't be bundled) "@/mitm/manager": "./src/mitm/manager.stub.ts", }, }, output: "standalone", + outputFileTracingRoot: projectRoot, outputFileTracingExcludes: { // Planning/task docs are not runtime assets and can break standalone copies // when broad fs/path tracing pulls the whole repository into the NFT graph. diff --git a/open-sse/config/providerRegistry.ts b/open-sse/config/providerRegistry.ts index cf9f826ff3..a8cfac03b0 100644 --- a/open-sse/config/providerRegistry.ts +++ b/open-sse/config/providerRegistry.ts @@ -480,7 +480,7 @@ export const REGISTRY: Record = { tokenUrl: "https://auth.openai.com/oauth/token", }, models: [ - { id: "codex-auto-review", name: "Codex Auto Review" }, + { id: "gpt-5.5", name: "GPT 5.5", targetFormat: "openai-responses" }, { id: "gpt-5.4", name: "GPT 5.4", targetFormat: "openai-responses" }, { id: "gpt-5.4-mini", name: "GPT 5.4 Mini", targetFormat: "openai-responses" }, { id: "gpt-5.5", name: "GPT 5.5", targetFormat: "openai-responses" }, diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index 5e33c67b9f..ab4fe28ef9 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -1,10 +1,16 @@ import { getCodexRequestDefaults } from "@/lib/providers/requestDefaults"; -import { BaseExecutor, setUserAgentHeader } from "./base.ts"; +import { + BaseExecutor, + mergeUpstreamExtraHeaders, + setUserAgentHeader, + type ExecuteInput, +} from "./base.ts"; import { CODEX_DEFAULT_INSTRUCTIONS } from "../config/codexInstructions.ts"; import { PROVIDERS } from "../config/constants.ts"; import { getCodexClientVersion, getCodexUserAgent } from "../config/codexClient.ts"; import { getAccessToken } from "../services/tokenRefresh.ts"; import { getThinkingBudgetConfig, ThinkingMode } from "../services/thinkingBudget.ts"; +import { websocket } from "wreq-js"; // ─── T09: Codex vs Spark Scope-Aware Rate Limiting ──────────────────────── // Codex has two independent quota pools: "codex" (standard) and "spark" (premium). @@ -163,6 +169,7 @@ export function getCodexDualWindowCooldownMs( const EFFORT_ORDER = ["none", "low", "medium", "high", "xhigh"] as const; type EffortLevel = (typeof EFFORT_ORDER)[number]; const CODEX_FAST_WIRE_VALUE = "priority"; +const CODEX_RESPONSES_WS_URL = "wss://chatgpt.com/backend-api/codex/responses"; function stringifyCodexInstructionContent(content: unknown): string { if (typeof content === "string") { @@ -484,6 +491,99 @@ function consumeResponsesStoreMarker(body: Record): unknown { return marker; } +function isCodexResponsesWebSocketRequired(model: string, credentials: unknown): boolean { + const normalizedModel = String(model || "") + .trim() + .toLowerCase(); + if (normalizedModel === "gpt-5.5") return true; + const providerSpecificData = + credentials && typeof credentials === "object" + ? (credentials as { providerSpecificData?: Record }).providerSpecificData + : null; + return providerSpecificData?.codexTransport === "websocket"; +} + +function toCodexResponseFailedEvent(parsed: Record): Record { + const upstreamError = + parsed.error && typeof parsed.error === "object" && !Array.isArray(parsed.error) + ? (parsed.error as Record) + : parsed; + const code = + typeof upstreamError.code === "string" + ? upstreamError.code + : typeof upstreamError.type === "string" + ? upstreamError.type + : "upstream_error"; + const message = + typeof upstreamError.message === "string" && upstreamError.message.trim() + ? upstreamError.message + : "Codex upstream error"; + const error: Record = { code, message }; + + if (typeof upstreamError.type === "string") error.type = upstreamError.type; + if (typeof parsed.status_code === "number") error.status_code = parsed.status_code; + if (typeof parsed.status === "number") error.status_code = parsed.status; + + return { + type: "response.failed", + response: { + id: null, + status: "failed", + error, + }, + }; +} + +export function encodeResponseSseEvent(raw: string): { sse: string; terminal: boolean } { + let eventType = "message"; + let payload = raw; + let terminal = false; + + try { + const parsed = JSON.parse(raw); + if (parsed && typeof parsed.type === "string" && parsed.type.trim()) { + eventType = parsed.type.trim(); + if (eventType === "error") { + const failed = toCodexResponseFailedEvent(parsed as Record); + payload = JSON.stringify(failed); + eventType = "response.failed"; + } + terminal = eventType === "response.completed" || eventType === "response.failed"; + } + } catch { + // Keep message as the generic SSE event for non-JSON upstream payloads. + } + + return { sse: `event: ${eventType}\ndata: ${payload}\n\n`, terminal }; +} + +function toWebSocketUrl(url: string): string { + if (url.startsWith("wss://") || url.startsWith("ws://")) return url; + if (url.startsWith("https://")) return `wss://${url.slice("https://".length)}`; + if (url.startsWith("http://")) return `ws://${url.slice("http://".length)}`; + return CODEX_RESPONSES_WS_URL; +} + +function normalizeCodexWsHeaders(headers: Record): Record { + const result: Record = {}; + for (const [key, value] of Object.entries(headers)) { + const lower = key.toLowerCase(); + if ( + lower === "host" || + lower === "connection" || + lower === "upgrade" || + lower === "sec-websocket-key" || + lower === "sec-websocket-version" || + lower === "sec-websocket-extensions" + ) { + continue; + } + result[key] = value; + } + result.Origin = "https://chatgpt.com"; + return result; +} + /** * Codex Executor - handles OpenAI Codex API (Responses API format) * Automatically injects default instructions if missing. @@ -494,6 +594,132 @@ export class CodexExecutor extends BaseExecutor { super("codex", PROVIDERS.codex); } + async execute(input: ExecuteInput) { + if (!isCodexResponsesWebSocketRequired(input.model, input.credentials)) { + return super.execute(input); + } + + const url = CODEX_RESPONSES_WS_URL; + const headers = normalizeCodexWsHeaders(this.buildHeaders(input.credentials, true)); + mergeUpstreamExtraHeaders(headers, input.upstreamExtraHeaders); + + const transformedBody = (await this.transformRequest( + input.model, + input.body, + true, + input.credentials + )) as Record; + transformedBody.model = input.model; + delete transformedBody.stream; + delete transformedBody.stream_options; + + const bodyString = JSON.stringify({ + type: "response.create", + ...transformedBody, + }); + + const stream = new ReadableStream({ + async start(controller) { + const encoder = new TextEncoder(); + let closed = false; + let ws: Awaited> | null = null; + + const closeController = () => { + if (closed) return; + closed = true; + try { + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + } catch { + // The downstream may already have gone away. + } + try { + controller.close(); + } catch { + // The controller may already be closed. + } + }; + + const failController = (code: string, message: string) => { + if (closed) return; + const payload = JSON.stringify({ + type: "response.failed", + response: { + id: null, + status: "failed", + error: { code, message }, + }, + }); + controller.enqueue(encoder.encode(`event: response.failed\ndata: ${payload}\n\n`)); + closeController(); + }; + + const abort = () => { + try { + ws?.close(1000, "client_aborted"); + } catch { + // ignore abort races + } + closeController(); + }; + + input.signal?.addEventListener("abort", abort, { once: true }); + + try { + ws = await websocket(toWebSocketUrl(url), { + browser: "chrome_142", + os: "windows", + headers, + }); + if (input.signal?.aborted) { + abort(); + return; + } + ws.onmessage = (event) => { + if (closed) return; + const raw = + typeof event.data === "string" + ? event.data + : Buffer.from(event.data as Buffer).toString("utf8"); + const sseEvent = encodeResponseSseEvent(raw); + controller.enqueue(encoder.encode(sseEvent.sse)); + if (sseEvent.terminal) { + closeController(); + } + }; + ws.onerror = (event) => { + failController( + "upstream_websocket_error", + event.message || "Codex upstream WebSocket error" + ); + }; + ws.onclose = () => { + closeController(); + }; + ws.send(bodyString); + } catch (error) { + failController( + "upstream_websocket_connect_failed", + error instanceof Error ? error.message : String(error) + ); + } + }, + }); + + return { + response: new Response(stream, { + status: 200, + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }, + }), + url, + headers, + transformedBody, + }; + } + buildUrl(model, stream, urlIndex = 0, credentials = null) { void model; void stream; diff --git a/package.json b/package.json index b96bec6cc6..24b0f77724 100644 --- a/package.json +++ b/package.json @@ -24,6 +24,7 @@ ".env.example", "scripts/postinstall.mjs", "scripts/postinstallSupport.mjs", + "scripts/responses-ws-proxy.mjs", "scripts/check-supported-node-runtime.ts", "scripts/sync-env.mjs", "scripts/native-binary-compat.mjs", diff --git a/scripts/pack-artifact-policy.ts b/scripts/pack-artifact-policy.ts index 7a2fc451ad..b2e55c0232 100644 --- a/scripts/pack-artifact-policy.ts +++ b/scripts/pack-artifact-policy.ts @@ -32,8 +32,10 @@ export const APP_STAGING_ALLOWED_EXACT_PATHS: string[] = [ "docs/openapi.yaml", "open-sse/mcp-server/server.js", "package.json", + "responses-ws-proxy.mjs", "scripts/sync-env.mjs", "server.js", + "server-ws.mjs", ]; export const APP_STAGING_ALLOWED_PATH_PREFIXES: string[] = [ @@ -74,6 +76,7 @@ export const PACK_ARTIFACT_ROOT_ALLOWED_EXACT_PATHS: string[] = [ "scripts/native-binary-compat.mjs", "scripts/postinstall.mjs", "scripts/postinstallSupport.mjs", + "scripts/responses-ws-proxy.mjs", "scripts/sync-env.mjs", "src/shared/utils/nodeRuntimeSupport.ts", ]; @@ -86,6 +89,8 @@ export const PACK_ARTIFACT_ROOT_ALLOWED_PATH_PREFIXES: string[] = [ export const PACK_ARTIFACT_REQUIRED_PATHS: string[] = [ "app/server.js", + "app/server-ws.mjs", + "app/responses-ws-proxy.mjs", "bin/mcp-server.mjs", "bin/nodeRuntimeSupport.mjs", "bin/omniroute.mjs", diff --git a/scripts/prepublish.ts b/scripts/prepublish.ts index 1e278ccb13..e389e0e4d4 100644 --- a/scripts/prepublish.ts +++ b/scripts/prepublish.ts @@ -195,6 +195,17 @@ console.log(" 📋 Copying standalone build to app/..."); mkdirSync(APP_DIR, { recursive: true }); cpSync(standaloneDir, APP_DIR, { recursive: true }); +const standaloneWsSrc = join(ROOT, "scripts", "standalone-server-ws.mjs"); +const responsesWsProxySrc = join(ROOT, "scripts", "responses-ws-proxy.mjs"); +if (existsSync(standaloneWsSrc) && existsSync(responsesWsProxySrc)) { + console.log(" 📋 Adding Responses WebSocket standalone wrapper..."); + cpSync(standaloneWsSrc, join(APP_DIR, "server-ws.mjs")); + writeFileSync( + join(APP_DIR, "responses-ws-proxy.mjs"), + 'export * from "../scripts/responses-ws-proxy.mjs";\n' + ); +} + // ── Next.js Turbopack Standalone Tracer Fix ─────────────── // Workaround for Next.js 15+ standalone mode missing Turbopack chunks const staticChunksSrc = join(ROOT, ".next", "server", "chunks"); diff --git a/scripts/responses-ws-proxy.mjs b/scripts/responses-ws-proxy.mjs new file mode 100644 index 0000000000..4e1d75f75e --- /dev/null +++ b/scripts/responses-ws-proxy.mjs @@ -0,0 +1,597 @@ +import { createHash, randomUUID } from "node:crypto"; +import { STATUS_CODES } from "node:http"; +import { websocket } from "wreq-js"; + +export const RESPONSES_WS_PUBLIC_PATHS = new Set([ + "/responses", + "/v1/responses", + "/api/v1/responses", +]); + +const INTERNAL_ROUTE = "/api/internal/codex-responses-ws"; +const WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; +const WS_QUERY_TOKEN_KEYS = ["api_key", "token", "access_token"]; +const textDecoder = new TextDecoder(); + +function isText(value) { + return typeof value === "string" && value.length > 0; +} + +function jsonStringifySafe(value) { + try { + return JSON.stringify(value); + } catch { + return JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { + code: "serialization_failed", + message: "Failed to serialize WebSocket payload", + }, + }, + }); + } +} + +export function isResponsesWsPath(pathname) { + return RESPONSES_WS_PUBLIC_PATHS.has(pathname); +} + +export function encodeWsFrame(opcode, payload = Buffer.alloc(0)) { + const payloadBuffer = Buffer.isBuffer(payload) ? payload : Buffer.from(payload); + const length = payloadBuffer.length; + + let header; + if (length < 126) { + header = Buffer.allocUnsafe(2); + header[1] = length; + } else if (length <= 0xffff) { + header = Buffer.allocUnsafe(4); + header[1] = 126; + header.writeUInt16BE(length, 2); + } else { + header = Buffer.allocUnsafe(10); + header[1] = 127; + header.writeBigUInt64BE(BigInt(length), 2); + } + + header[0] = 0x80 | (opcode & 0x0f); + return Buffer.concat([header, payloadBuffer]); +} + +export function decodeClientFrames(buffer) { + const frames = []; + let offset = 0; + + while (buffer.length - offset >= 2) { + const byte1 = buffer[offset]; + const byte2 = buffer[offset + 1]; + const fin = (byte1 & 0x80) !== 0; + const opcode = byte1 & 0x0f; + const masked = (byte2 & 0x80) !== 0; + let payloadLength = byte2 & 0x7f; + let headerLength = 2; + + if (!masked) { + throw new Error("Client WebSocket frames must be masked"); + } + + if (payloadLength === 126) { + if (buffer.length - offset < 4) break; + payloadLength = buffer.readUInt16BE(offset + 2); + headerLength = 4; + } else if (payloadLength === 127) { + if (buffer.length - offset < 10) break; + const bigLength = buffer.readBigUInt64BE(offset + 2); + if (bigLength > BigInt(Number.MAX_SAFE_INTEGER)) { + throw new Error("WebSocket payload too large"); + } + payloadLength = Number(bigLength); + headerLength = 10; + } + + const totalLength = headerLength + 4 + payloadLength; + if (buffer.length - offset < totalLength) break; + + const mask = buffer.subarray(offset + headerLength, offset + headerLength + 4); + const payload = Buffer.from(buffer.subarray(offset + headerLength + 4, offset + totalLength)); + for (let index = 0; index < payload.length; index += 1) { + payload[index] ^= mask[index % 4]; + } + + frames.push({ fin, opcode, payload }); + offset += totalLength; + } + + return { + frames, + remaining: buffer.subarray(offset), + }; +} + +function writeHttpError(socket, status, body, headers = {}) { + if (!socket.writable || socket.destroyed) return; + + const bodyBuffer = Buffer.from(body || "", "utf8"); + const statusText = STATUS_CODES[status] || "Error"; + const responseHeaders = { + Connection: "close", + "Content-Length": String(bodyBuffer.length), + "Content-Type": "application/json; charset=utf-8", + ...headers, + }; + + const head = [ + `HTTP/1.1 ${status} ${statusText}`, + ...Object.entries(responseHeaders).map(([name, value]) => `${name}: ${value}`), + "", + "", + ].join("\r\n"); + + socket.write(head); + socket.end(bodyBuffer); +} + +function getAuthHeaders(requestUrl, requestHeaders) { + const headers = {}; + if (isText(requestHeaders.authorization)) { + headers.authorization = requestHeaders.authorization; + } else { + const url = new URL(requestUrl, "http://omniroute.local"); + for (const key of WS_QUERY_TOKEN_KEYS) { + const value = url.searchParams.get(key); + if (isText(value)) { + headers.authorization = `Bearer ${value.trim()}`; + break; + } + } + } + + if (isText(requestHeaders.cookie)) headers.cookie = requestHeaders.cookie; + if (isText(requestHeaders.origin)) headers.origin = requestHeaders.origin; + if (isText(requestHeaders["x-forwarded-for"])) { + headers["x-forwarded-for"] = requestHeaders["x-forwarded-for"]; + } + return headers; +} + +function getResponseCreatePayload(message) { + if (!message || typeof message !== "object" || Array.isArray(message)) return null; + if (message.type !== "response.create") return null; + if ( + message.response && + typeof message.response === "object" && + !Array.isArray(message.response) + ) { + return message.response; + } + if (message.body && typeof message.body === "object" && !Array.isArray(message.body)) { + return message.body; + } + if (message.payload && typeof message.payload === "object" && !Array.isArray(message.payload)) { + return message.payload; + } + const { type, ...payload } = message; + return payload; +} + +function withPreparedResponseCreate(message, preparedBody) { + const next = { ...message }; + if ( + message.response && + typeof message.response === "object" && + !Array.isArray(message.response) + ) { + next.response = preparedBody; + } else if (message.body && typeof message.body === "object" && !Array.isArray(message.body)) { + next.body = preparedBody; + } else if ( + message.payload && + typeof message.payload === "object" && + !Array.isArray(message.payload) + ) { + next.payload = preparedBody; + } else { + return { type: "response.create", ...preparedBody }; + } + return next; +} + +async function callInternal(fetchImpl, baseUrl, bridgeSecret, action, payload) { + const response = await fetchImpl(new URL(INTERNAL_ROUTE, baseUrl), { + method: "POST", + headers: { + "content-type": "application/json", + "x-omniroute-ws-bridge-secret": bridgeSecret, + }, + body: JSON.stringify({ action, ...payload }), + }); + const text = await response.text(); + let json = null; + try { + json = text ? JSON.parse(text) : null; + } catch { + json = null; + } + return { ok: response.ok, status: response.status, text, json, headers: response.headers }; +} + +class ResponsesWsSession { + constructor({ + baseUrl, + bridgeSecret, + fetchImpl, + socket, + requestHeaders, + requestUrl, + wsFactory, + pingIntervalMs, + idleTimeoutMs, + }) { + this.baseUrl = baseUrl; + this.bridgeSecret = bridgeSecret; + this.fetchImpl = fetchImpl; + this.socket = socket; + this.requestHeaders = requestHeaders; + this.requestUrl = requestUrl; + this.wsFactory = wsFactory; + this.pingIntervalMs = pingIntervalMs; + this.idleTimeoutMs = idleTimeoutMs; + this.sessionId = randomUUID(); + this.closed = false; + this.buffer = Buffer.alloc(0); + this.fragmentOpcode = null; + this.fragmentParts = []; + this.upstream = null; + this.upstreamReady = null; + this.lastSeenAt = Date.now(); + + this.pingTimer = setInterval(() => { + if (this.closed) return; + const idleForMs = Date.now() - this.lastSeenAt; + if (idleForMs >= this.idleTimeoutMs) { + this.close(1001, "idle_timeout"); + return; + } + this.sendFrame(0x9); + }, this.pingIntervalMs); + + this.socket.setNoDelay(true); + this.socket.on("data", (chunk) => { + this.onData(chunk).catch((error) => { + this.sendFailure( + "frame_decode_failed", + error instanceof Error ? error.message : String(error) + ); + this.close(1011, "frame_decode_failed"); + }); + }); + this.socket.on("close", () => this.dispose()); + this.socket.on("end", () => this.dispose()); + this.socket.on("error", () => this.dispose()); + } + + sendFrame(opcode, payload) { + if (this.closed || this.socket.destroyed) return; + this.socket.write(encodeWsFrame(opcode, payload)); + } + + sendJson(payload) { + this.sendFrame(0x1, Buffer.from(jsonStringifySafe(payload), "utf8")); + } + + sendFailure(code, message) { + this.sendJson({ + type: "response.failed", + response: { + id: null, + status: "failed", + error: { + code, + message, + }, + }, + }); + } + + async onData(chunk) { + this.lastSeenAt = Date.now(); + this.buffer = Buffer.concat([this.buffer, chunk]); + const parsed = decodeClientFrames(this.buffer); + this.buffer = parsed.remaining; + + for (const frame of parsed.frames) { + await this.handleFrame(frame); + } + } + + async handleFrame(frame) { + switch (frame.opcode) { + case 0x0: + if (this.fragmentOpcode === null) { + this.sendFailure("unexpected_continuation", "Unexpected continuation frame"); + return; + } + this.fragmentParts.push(frame.payload); + if (frame.fin) { + const payload = Buffer.concat(this.fragmentParts); + const opcode = this.fragmentOpcode; + this.fragmentOpcode = null; + this.fragmentParts = []; + await this.handleDataFrame(opcode, payload); + } + return; + case 0x1: + case 0x2: + if (!frame.fin) { + this.fragmentOpcode = frame.opcode; + this.fragmentParts = [frame.payload]; + return; + } + await this.handleDataFrame(frame.opcode, frame.payload); + return; + case 0x8: + this.close(); + return; + case 0x9: + this.sendFrame(0xa, frame.payload); + return; + case 0xa: + this.lastSeenAt = Date.now(); + return; + default: + this.sendFailure("unsupported_opcode", `Unsupported opcode ${frame.opcode}`); + } + } + + async handleDataFrame(opcode, payload) { + if (opcode !== 0x1) { + this.sendFailure("unsupported_payload", "Only UTF-8 text messages are supported"); + return; + } + + const raw = textDecoder.decode(payload); + let message; + try { + message = JSON.parse(raw); + } catch { + this.sendFailure("invalid_json", "WebSocket message must be valid JSON"); + return; + } + + await this.forwardClientMessage(message); + } + + async ensureUpstream(firstMessage) { + if (this.upstreamReady) return this.upstreamReady; + + this.upstreamReady = (async () => { + const responseBody = getResponseCreatePayload(firstMessage); + if (responseBody === null) { + throw new Error("First Responses WebSocket message must be response.create"); + } + + const prepared = await callInternal( + this.fetchImpl, + this.baseUrl, + this.bridgeSecret, + "prepare", + { + requestUrl: this.requestUrl, + headers: getAuthHeaders(this.requestUrl, this.requestHeaders), + message: firstMessage, + response: responseBody, + } + ); + + if (!prepared.ok) { + const message = + prepared.json?.error?.message || + prepared.json?.message || + prepared.text || + "Codex WS prepare failed"; + const code = prepared.json?.error?.code || "codex_ws_prepare_failed"; + const error = new Error(message); + error.code = code; + throw error; + } + + const upstream = await this.wsFactory(prepared.json.upstreamUrl, { + browser: prepared.json.browser || "chrome_142", + os: prepared.json.os || "windows", + headers: prepared.json.headers || {}, + }); + + upstream.onmessage = (event) => { + if (this.closed) return; + const data = + typeof event.data === "string" ? event.data : Buffer.from(event.data).toString("utf8"); + this.sendFrame(0x1, Buffer.from(data, "utf8")); + }; + upstream.onerror = (event) => { + if (this.closed) return; + this.sendFailure( + "upstream_websocket_error", + event.message || "Codex upstream WebSocket error" + ); + }; + upstream.onclose = (event) => { + if (this.closed) return; + this.close(event.code || 1000, event.reason || "upstream_closed"); + }; + + this.upstream = upstream; + return { + upstream, + firstMessage: withPreparedResponseCreate(firstMessage, prepared.json.response), + }; + })(); + + return this.upstreamReady; + } + + async forwardClientMessage(message) { + try { + if (!this.upstream) { + const { upstream, firstMessage } = await this.ensureUpstream(message); + upstream.send(jsonStringifySafe(firstMessage)); + return; + } + this.upstream.send(jsonStringifySafe(message)); + } catch (error) { + const code = error?.code || "upstream_websocket_connect_failed"; + const messageText = error instanceof Error ? error.message : String(error); + this.sendFailure(code, messageText); + this.close(1011, "upstream_connect_failed"); + } + } + + close(code = 1000, reason = "normal_closure") { + if (this.closed) return; + this.closed = true; + + clearInterval(this.pingTimer); + try { + this.upstream?.close?.(code, reason); + } catch { + // ignore close races + } + + const reasonBuffer = Buffer.from(reason, "utf8"); + const payload = Buffer.allocUnsafe(2 + reasonBuffer.length); + payload.writeUInt16BE(code, 0); + reasonBuffer.copy(payload, 2); + this.sendFrame(0x8, payload); + this.socket.end(); + setTimeout(() => { + if (!this.socket.destroyed) { + this.socket.destroy(); + } + }, 50).unref?.(); + } + + dispose() { + if (this.closed) return; + this.closed = true; + clearInterval(this.pingTimer); + try { + this.upstream?.close?.(1000, "downstream_closed"); + } catch { + // ignore close races + } + } +} + +export function createResponsesWsProxy({ + baseUrl, + bridgeSecret, + fetchImpl = fetch, + wsFactory = websocket, + pingIntervalMs = 25000, + idleTimeoutMs = 90000, +} = {}) { + if (!isText(baseUrl)) { + throw new Error("createResponsesWsProxy requires a baseUrl"); + } + if (!isText(bridgeSecret)) { + throw new Error("createResponsesWsProxy requires a bridgeSecret"); + } + + return { + isResponsesWsPath, + async handleUpgrade(req, socket, head) { + const pathname = new URL(req.url || "/", baseUrl).pathname; + if (!isResponsesWsPath(pathname)) { + return false; + } + + const upgradeHeader = String(req.headers.upgrade || "").toLowerCase(); + if (upgradeHeader !== "websocket") { + writeHttpError( + socket, + 426, + JSON.stringify({ + error: { + message: "Upgrade Required", + code: "upgrade_required", + }, + }), + { Upgrade: "websocket" } + ); + return true; + } + + try { + const auth = await callInternal(fetchImpl, baseUrl, bridgeSecret, "authenticate", { + requestUrl: req.url || pathname, + headers: getAuthHeaders(req.url || pathname, req.headers), + }); + if (!auth.ok) { + writeHttpError( + socket, + auth.status, + auth.text || "{}", + Object.fromEntries(auth.headers.entries()) + ); + return true; + } + + const wsKey = req.headers["sec-websocket-key"]; + if (!isText(wsKey)) { + writeHttpError( + socket, + 400, + JSON.stringify({ + error: { + message: "Missing sec-websocket-key header", + code: "bad_websocket_handshake", + }, + }) + ); + return true; + } + + const acceptKey = createHash("sha1").update(`${wsKey}${WS_GUID}`).digest("base64"); + const headers = [ + "HTTP/1.1 101 Switching Protocols", + "Upgrade: websocket", + "Connection: Upgrade", + `Sec-WebSocket-Accept: ${acceptKey}`, + "", + "", + ].join("\r\n"); + + socket.write(headers); + if (head && head.length > 0) { + socket.unshift(head); + } + + new ResponsesWsSession({ + baseUrl, + bridgeSecret, + fetchImpl, + socket, + requestUrl: req.url || pathname, + requestHeaders: req.headers, + wsFactory, + pingIntervalMs, + idleTimeoutMs, + }); + return true; + } catch (error) { + writeHttpError( + socket, + 500, + JSON.stringify({ + error: { + message: error instanceof Error ? error.message : String(error), + code: "responses_websocket_proxy_failed", + }, + }) + ); + return true; + } + }, + }; +} diff --git a/scripts/run-next.mjs b/scripts/run-next.mjs index 15f31dbb0d..908d85be9d 100644 --- a/scripts/run-next.mjs +++ b/scripts/run-next.mjs @@ -7,6 +7,8 @@ import next from "next"; import { bootstrapEnv } from "./bootstrap-env.mjs"; import { resolveRuntimePorts, withRuntimePortEnv } from "./runtime-env.mjs"; import { createOmnirouteWsBridge } from "./v1-ws-bridge.mjs"; +import { createResponsesWsProxy } from "./responses-ws-proxy.mjs"; +import { randomUUID } from "node:crypto"; // Add check for conflicting app/ directory (Issue #1206) const rootAppDir = path.join(process.cwd(), "app"); @@ -35,6 +37,7 @@ for (const [key, value] of Object.entries(mergedEnv)) { const { dashboardPort } = runtimePorts; const hostname = process.env.HOST || "0.0.0.0"; const useTurbopack = dev && mergedEnv.OMNIROUTE_USE_TURBOPACK === "1"; +process.env.OMNIROUTE_WS_BRIDGE_SECRET ||= randomUUID(); const nextApp = next({ dev, @@ -50,6 +53,10 @@ async function start() { const requestHandler = nextApp.getRequestHandler(); const upgradeHandler = nextApp.getUpgradeHandler(); + const responsesWsProxy = createResponsesWsProxy({ + baseUrl: `http://127.0.0.1:${dashboardPort}`, + bridgeSecret: process.env.OMNIROUTE_WS_BRIDGE_SECRET, + }); const wsBridge = createOmnirouteWsBridge({ baseUrl: `http://127.0.0.1:${dashboardPort}`, }); @@ -57,6 +64,8 @@ async function start() { const server = http.createServer((req, res) => requestHandler(req, res)); server.on("upgrade", async (req, socket, head) => { try { + const responsesWsHandled = await responsesWsProxy.handleUpgrade(req, socket, head); + if (responsesWsHandled) return; const handled = await wsBridge.handleUpgrade(req, socket, head); if (handled) return; await upgradeHandler(req, socket, head); diff --git a/scripts/standalone-server-ws.mjs b/scripts/standalone-server-ws.mjs new file mode 100644 index 0000000000..12a20bf9eb --- /dev/null +++ b/scripts/standalone-server-ws.mjs @@ -0,0 +1,70 @@ +import http from "node:http"; +import { randomUUID } from "node:crypto"; +import { createResponsesWsProxy } from "./responses-ws-proxy.mjs"; + +const originalCreateServer = http.createServer.bind(http); +const proxiesByPort = new Map(); + +process.env.OMNIROUTE_WS_BRIDGE_SECRET ||= randomUUID(); + +function getPort(server) { + const address = server.address?.(); + if (address && typeof address === "object" && typeof address.port === "number") { + return address.port; + } + const rawPort = process.env.PORT || process.env.DASHBOARD_PORT || "3000"; + const parsed = Number.parseInt(rawPort, 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 3000; +} + +function getProxy(server) { + const port = getPort(server); + const existing = proxiesByPort.get(port); + if (existing) return existing; + + const proxy = createResponsesWsProxy({ + baseUrl: `http://127.0.0.1:${port}`, + bridgeSecret: process.env.OMNIROUTE_WS_BRIDGE_SECRET, + }); + proxiesByPort.set(port, proxy); + return proxy; +} + +function wrapUpgradeListener(server, listener) { + return async function responsesWsAwareUpgrade(req, socket, head) { + try { + const handled = await getProxy(server).handleUpgrade(req, socket, head); + if (handled) return; + return listener.call(this, req, socket, head); + } catch (error) { + if (!socket.destroyed) { + socket.destroy(error instanceof Error ? error : undefined); + } + console.error("[Responses WS] Upgrade handling failed:", error); + } + }; +} + +http.createServer = function createServerWithResponsesWs(...args) { + const server = originalCreateServer(...args); + const originalOn = server.on.bind(server); + const originalAddListener = server.addListener.bind(server); + + server.on = function patchedOn(eventName, listener) { + if (eventName === "upgrade" && typeof listener === "function") { + return originalOn(eventName, wrapUpgradeListener(server, listener)); + } + return originalOn(eventName, listener); + }; + + server.addListener = function patchedAddListener(eventName, listener) { + if (eventName === "upgrade" && typeof listener === "function") { + return originalAddListener(eventName, wrapUpgradeListener(server, listener)); + } + return originalAddListener(eventName, listener); + }; + + return server; +}; + +await import("./server.js"); diff --git a/scripts/validate-pack-artifact.ts b/scripts/validate-pack-artifact.ts index 2e6a5bf219..afaaa20761 100644 --- a/scripts/validate-pack-artifact.ts +++ b/scripts/validate-pack-artifact.ts @@ -18,13 +18,27 @@ const ROOT: string = join(__dirname, ".."); const npmCommand: string = process.platform === "win32" ? "npm.cmd" : "npm"; function runPackDryRun(): any { - const output = execFileSync(npmCommand, ["pack", "--dry-run", "--json", "--ignore-scripts"], { + const npmExecPath = process.env.npm_execpath; + const command = npmExecPath ? process.execPath : npmCommand; + const args = [ + ...(npmExecPath ? [npmExecPath] : []), + "pack", + "--dry-run", + "--json", + "--ignore-scripts", + ]; + + const output = execFileSync(command, args, { cwd: ROOT, encoding: "utf8", stdio: ["ignore", "pipe", "pipe"], }); - const parsed = JSON.parse(output); + const jsonStart = output.indexOf("["); + const jsonEnd = output.lastIndexOf("]"); + const jsonPayload = + jsonStart >= 0 && jsonEnd > jsonStart ? output.slice(jsonStart, jsonEnd + 1) : output; + const parsed = JSON.parse(jsonPayload); const packReport = Array.isArray(parsed) ? parsed[0] : null; if (!packReport || !Array.isArray(packReport.files)) { diff --git a/src/app/(dashboard)/dashboard/cli-tools/components/CodexToolCard.tsx b/src/app/(dashboard)/dashboard/cli-tools/components/CodexToolCard.tsx index 185dbe0687..689d18ee8b 100644 --- a/src/app/(dashboard)/dashboard/cli-tools/components/CodexToolCard.tsx +++ b/src/app/(dashboard)/dashboard/cli-tools/components/CodexToolCard.tsx @@ -25,8 +25,9 @@ export default function CodexToolCard({ const [message, setMessage] = useState(null); const [showInstallGuide, setShowInstallGuide] = useState(false); const [selectedApiKey, setSelectedApiKey] = useState(""); - const [selectedModel, setSelectedModel] = useState(""); + const [selectedModel, setSelectedModel] = useState("gpt-5.5"); const CODEX_DEFAULT_MODELS = [ + "gpt-5.5", "gpt-5.3-codex", "gpt-5.4", "gpt-5.2-codex", @@ -35,7 +36,7 @@ export default function CodexToolCard({ "gpt-5.1-codex-mini", ]; const [modelMappings, setModelMappings] = useState>({}); - const [reasoningEffort, setReasoningEffort] = useState("medium"); + const [reasoningEffort, setReasoningEffort] = useState("xhigh"); const [wireApi, setWireApi] = useState("chat"); const [modalOpen, setModalOpen] = useState(false); const [modalTarget, setModalTarget] = useState(null); // null = default model, string = mapping key @@ -584,7 +585,7 @@ openai_base_url = "${getEffectiveBaseUrl()}" type="text" value={selectedModel} onChange={(e) => setSelectedModel(e.target.value)} - placeholder="gpt-5.4" + placeholder="gpt-5.5" className="flex-1 px-2 py-1.5 bg-surface rounded border border-border text-xs focus:outline-none focus:ring-1 focus:ring-primary/50" /> {selectedModel && ( @@ -615,6 +616,7 @@ openai_base_url = "${getEffectiveBaseUrl()}" + diff --git a/src/app/api/internal/codex-responses-ws/route.ts b/src/app/api/internal/codex-responses-ws/route.ts new file mode 100644 index 0000000000..a85014c67b --- /dev/null +++ b/src/app/api/internal/codex-responses-ws/route.ts @@ -0,0 +1,185 @@ +import { NextResponse } from "next/server"; +import { CodexExecutor } from "@omniroute/open-sse/executors/codex.ts"; +import { getApiKeyMetadata } from "@/lib/db/apiKeys"; +import { authorizeWebSocketHandshake, extractWsTokenFromRequest } from "@/lib/ws/handshake"; +import { getModelInfo } from "@/sse/services/model"; +import { getProviderCredentialsWithQuotaPreflight } from "@/sse/services/auth"; +import { checkAndRefreshToken } from "@/sse/services/tokenRefresh"; + +const CODEX_RESPONSES_WS_URL = "wss://chatgpt.com/backend-api/codex/responses"; +const executor = new CodexExecutor(); + +type JsonRecord = Record; + +function isRecord(value: unknown): value is JsonRecord { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function getBridgeSecret(): string { + return process.env.OMNIROUTE_WS_BRIDGE_SECRET || ""; +} + +function getAuthRequest(body: JsonRecord): Request { + const requestUrl = typeof body.requestUrl === "string" ? body.requestUrl : "/api/v1/responses"; + const headers = isRecord(body.headers) ? body.headers : {}; + const url = new URL(requestUrl, "http://omniroute.local"); + const requestHeaders = new Headers(); + + for (const [key, value] of Object.entries(headers)) { + if (typeof value === "string") { + requestHeaders.set(key, value); + } + } + + return new Request(url, { headers: requestHeaders }); +} + +function jsonError(status: number, code: string, message: string) { + return NextResponse.json( + { + error: { + code, + message, + }, + }, + { status } + ); +} + +function normalizeUpstreamHeaders(headers: Record): Record { + const result: Record = {}; + for (const [key, value] of Object.entries(headers)) { + const lower = key.toLowerCase(); + if ( + lower === "host" || + lower === "connection" || + lower === "upgrade" || + lower === "sec-websocket-key" || + lower === "sec-websocket-version" || + lower === "sec-websocket-extensions" + ) { + continue; + } + result[key] = value; + } + result.Origin = "https://chatgpt.com"; + return result; +} + +async function authenticate(body: JsonRecord) { + const authRequest = getAuthRequest(body); + const auth = await authorizeWebSocketHandshake(authRequest); + if (!auth.authorized) { + return jsonError( + auth.hasCredential ? 403 : 401, + auth.hasCredential ? "ws_auth_invalid" : "ws_auth_required", + auth.hasCredential ? "Invalid WebSocket credential" : "WebSocket auth required" + ); + } + + return NextResponse.json({ + ok: true, + authenticated: auth.authenticated, + authType: auth.authType, + wsAuth: auth.wsAuth, + }); +} + +async function prepare(body: JsonRecord) { + const authResponse = await authenticate(body); + if (!authResponse.ok) return authResponse; + + const authRequest = getAuthRequest(body); + const apiKey = extractWsTokenFromRequest(authRequest); + const metadata = apiKey ? await getApiKeyMetadata(apiKey).catch(() => null) : null; + const allowedConnections = + metadata && Array.isArray(metadata.allowedConnections) && metadata.allowedConnections.length > 0 + ? metadata.allowedConnections + : null; + + const responseBody = isRecord(body.response) ? body.response : {}; + const requestedModel = + typeof responseBody.model === "string" && responseBody.model.trim() + ? responseBody.model.trim() + : "gpt-5.5"; + const modelInfo = await getModelInfo(requestedModel); + const provider = modelInfo.provider; + const model = modelInfo.model || requestedModel; + + if (provider !== "codex") { + return jsonError( + 400, + "codex_ws_provider_required", + `Responses WebSocket bridge only supports Codex models, got ${provider || "unknown"}` + ); + } + + const credentials = await getProviderCredentialsWithQuotaPreflight( + provider, + null, + allowedConnections, + model + ); + + if (!credentials || "allRateLimited" in credentials) { + return jsonError( + 503, + "codex_credentials_unavailable", + "No available Codex OAuth connection for Responses WebSocket" + ); + } + + const refreshedCredentials = await checkAndRefreshToken(provider, credentials); + if (!refreshedCredentials?.accessToken) { + return jsonError(401, "codex_oauth_token_missing", "Codex OAuth access token is missing"); + } + + const transformed = (await executor.transformRequest( + model, + responseBody, + true, + refreshedCredentials + )) as JsonRecord; + transformed.model = model; + delete transformed.stream; + delete transformed.stream_options; + + const headers = normalizeUpstreamHeaders(executor.buildHeaders(refreshedCredentials, true)); + + return NextResponse.json({ + ok: true, + upstreamUrl: CODEX_RESPONSES_WS_URL, + browser: "chrome_142", + os: "windows", + connectionId: refreshedCredentials.connectionId, + model, + headers, + response: transformed, + }); +} + +export async function POST(request: Request) { + const expectedSecret = getBridgeSecret(); + const receivedSecret = request.headers.get("x-omniroute-ws-bridge-secret") || ""; + if (!expectedSecret || receivedSecret !== expectedSecret) { + return jsonError(403, "internal_bridge_forbidden", "Forbidden"); + } + + let body: JsonRecord; + try { + const parsed = await request.json(); + body = isRecord(parsed) ? parsed : {}; + } catch { + return jsonError(400, "invalid_json", "Request body must be JSON"); + } + + const action = typeof body.action === "string" ? body.action : ""; + if (action === "authenticate") { + return authenticate(body); + } + if (action === "prepare") { + return prepare(body); + } + + return jsonError(400, "invalid_action", "Unsupported bridge action"); +} diff --git a/src/shared/validation/schemas.ts b/src/shared/validation/schemas.ts index ba31860acf..6049535101 100644 --- a/src/shared/validation/schemas.ts +++ b/src/shared/validation/schemas.ts @@ -1710,7 +1710,7 @@ export const cliModelConfigSchema = z.object({ baseUrl: z.string().trim().min(1, "baseUrl and model are required"), apiKey: z.string().optional(), model: z.string().trim().min(1, "baseUrl and model are required"), - reasoningEffort: z.enum(["none", "low", "medium", "high"]).optional(), + reasoningEffort: z.enum(["none", "low", "medium", "high", "xhigh"]).optional(), wireApi: z.enum(["chat", "responses"]).optional(), modelMappings: z.record(z.string().trim().min(1), z.string().trim().min(1)).optional(), }); diff --git a/tests/unit/cli-model-config-schema.test.ts b/tests/unit/cli-model-config-schema.test.ts new file mode 100644 index 0000000000..1c1b23bf75 --- /dev/null +++ b/tests/unit/cli-model-config-schema.test.ts @@ -0,0 +1,19 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { cliModelConfigSchema } from "../../src/shared/validation/schemas.ts"; + +test("cliModelConfigSchema accepts Codex xhigh reasoning effort", () => { + const result = cliModelConfigSchema.safeParse({ + baseUrl: "http://localhost:20128/api/v1", + apiKey: "sk_omniroute", + model: "gpt-5.5", + reasoningEffort: "xhigh", + wireApi: "responses", + }); + + assert.equal(result.success, true); + if (result.success) { + assert.equal(result.data.reasoningEffort, "xhigh"); + } +}); diff --git a/tests/unit/executor-codex.test.ts b/tests/unit/executor-codex.test.ts index 32d81f5d92..f125b28ab1 100644 --- a/tests/unit/executor-codex.test.ts +++ b/tests/unit/executor-codex.test.ts @@ -3,6 +3,7 @@ import assert from "node:assert/strict"; import { CodexExecutor, + encodeResponseSseEvent, getCodexModelScope, getCodexRateLimitKey, getCodexResetTime, @@ -272,6 +273,42 @@ test("CodexExecutor.transformRequest lets model suffix beat connection reasoning assert.equal(result.reasoning.effort, "high"); }); +test("CodexExecutor.transformRequest keeps gpt-5.5 as the model and applies xhigh reasoning", () => { + const executor = new CodexExecutor(); + const result = executor.transformRequest( + "gpt-5.5", + { model: "gpt-5.5", input: [], reasoning_effort: "xhigh" }, + false, + {} + ); + + assert.equal(result.model, "gpt-5.5"); + assert.equal(result.reasoning.effort, "xhigh"); +}); + +test("CodexExecutor maps Codex websocket error events to response.failed SSE", () => { + const raw = JSON.stringify({ + type: "error", + status_code: 429, + error: { + type: "usage_limit_reached", + message: "The usage limit has been reached", + }, + }); + + const result = encodeResponseSseEvent(raw); + assert.equal(result.terminal, true); + assert.match(result.sse, /^event: response\.failed/m); + + const dataLine = result.sse.split("\n").find((line) => line.startsWith("data: ")); + assert.ok(dataLine); + const payload = JSON.parse(dataLine.slice("data: ".length)); + assert.equal(payload.type, "response.failed"); + assert.equal(payload.response.status, "failed"); + assert.equal(payload.response.error.code, "usage_limit_reached"); + assert.equal(payload.response.error.status_code, 429); +}); + test("CodexExecutor.transformRequest does not apply connection reasoning defaults when Thinking Budget is not passthrough", () => { const executor = new CodexExecutor(); setThinkingBudgetConfig({ mode: ThinkingMode.AUTO }); diff --git a/tests/unit/plan3-p0.test.ts b/tests/unit/plan3-p0.test.ts index 3a722d288b..5cba2ea3ff 100644 --- a/tests/unit/plan3-p0.test.ts +++ b/tests/unit/plan3-p0.test.ts @@ -43,6 +43,18 @@ test("getModelInfoCore resolves codex-auto-review to codex", async () => { assert.equal(info.model, "codex-auto-review"); }); +test("getModelInfoCore resolves gpt-5.5 to codex", async () => { + const info = await getModelInfoCore("gpt-5.5", {}); + assert.equal(info.provider, "codex"); + assert.equal(info.model, "gpt-5.5"); +}); + +test("getModelInfoCore resolves explicit gpt-5.5 Codex model", async () => { + const info = await getModelInfoCore("cx/gpt-5.5", {}); + assert.equal(info.provider, "codex"); + assert.equal(info.model, "gpt-5.5"); +}); + test("getModelInfoCore returns explicit ambiguity metadata for ambiguous unprefixed model", async () => { const info = await getModelInfoCore("claude-haiku-4.5", {}); assert.equal(info.provider, null); diff --git a/tests/unit/responses-ws-proxy.test.mjs b/tests/unit/responses-ws-proxy.test.mjs new file mode 100644 index 0000000000..51bf5c3448 --- /dev/null +++ b/tests/unit/responses-ws-proxy.test.mjs @@ -0,0 +1,159 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; + +const { createResponsesWsProxy } = await import("../../scripts/responses-ws-proxy.mjs"); + +function listen(server) { + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + resolve(address.port); + }); + }); +} + +function close(server) { + return new Promise((resolve) => { + server.close(() => resolve()); + }); +} + +function readRequestBody(req) { + return new Promise((resolve, reject) => { + const chunks = []; + req.on("data", (chunk) => chunks.push(chunk)); + req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); + req.on("error", reject); + }); +} + +function waitFor(predicate, { timeoutMs = 3000, intervalMs = 10 } = {}) { + const startedAt = Date.now(); + return new Promise((resolve, reject) => { + const timer = setInterval(() => { + try { + const value = predicate(); + if (value) { + clearInterval(timer); + resolve(value); + return; + } + if (Date.now() - startedAt >= timeoutMs) { + clearInterval(timer); + reject(new Error("Timed out waiting for condition")); + } + } catch (error) { + clearInterval(timer); + reject(error); + } + }, intervalMs); + }); +} + +test("responses ws proxy prepares and forwards OpenAI Responses websocket events", async () => { + const internalRequests = []; + const upstreamSends = []; + const downstreamMessages = []; + + const server = http.createServer(async (req, res) => { + const url = new URL(req.url || "/", `http://${req.headers.host}`); + if (url.pathname === "/api/internal/codex-responses-ws") { + const body = JSON.parse((await readRequestBody(req)) || "{}"); + internalRequests.push(body); + + if (body.action === "authenticate") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true, authenticated: true, authType: "api_key" })); + return; + } + + if (body.action === "prepare") { + res.writeHead(200, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + ok: true, + upstreamUrl: "wss://chatgpt.com/backend-api/codex/responses", + headers: { Authorization: "Bearer upstream-token" }, + response: { + ...body.response, + model: "gpt-5.5", + stream: undefined, + }, + }) + ); + return; + } + } + + res.writeHead(404, { "content-type": "application/json" }); + res.end(JSON.stringify({ error: "not_found" })); + }); + + const fakeUpstream = { + send(data) { + upstreamSends.push(JSON.parse(data)); + setTimeout(() => { + fakeUpstream.onmessage?.({ + data: JSON.stringify({ + type: "response.completed", + response: { id: "resp_1", status: "completed" }, + }), + }); + }, 10); + }, + close() {}, + onmessage: null, + onerror: null, + onclose: null, + }; + + const port = await listen(server); + const baseUrl = `http://127.0.0.1:${port}`; + const proxy = createResponsesWsProxy({ + baseUrl, + bridgeSecret: "bridge-secret", + pingIntervalMs: 1000, + idleTimeoutMs: 10000, + wsFactory: async (url, options) => { + assert.equal(url, "wss://chatgpt.com/backend-api/codex/responses"); + assert.equal(options.headers.Authorization, "Bearer upstream-token"); + return fakeUpstream; + }, + }); + + server.on("upgrade", async (req, socket, head) => { + const handled = await proxy.handleUpgrade(req, socket, head); + if (!handled && !socket.destroyed) { + socket.destroy(); + } + }); + + const ws = new WebSocket(`ws://127.0.0.1:${port}/api/v1/responses?api_key=local-token`); + ws.addEventListener("message", (event) => { + downstreamMessages.push(JSON.parse(String(event.data))); + }); + + await new Promise((resolve) => ws.addEventListener("open", resolve, { once: true })); + ws.send( + JSON.stringify({ + type: "response.create", + model: "gpt-5.5", + input: [{ role: "user", content: "hello" }], + reasoning: { effort: "xhigh" }, + }) + ); + + await waitFor(() => downstreamMessages.find((entry) => entry.type === "response.completed")); + + assert.equal(internalRequests[0].action, "authenticate"); + assert.equal(internalRequests[1].action, "prepare"); + assert.equal(upstreamSends.length, 1); + assert.equal(upstreamSends[0].type, "response.create"); + assert.equal(upstreamSends[0].model, "gpt-5.5"); + assert.equal(upstreamSends[0].reasoning.effort, "xhigh"); + assert.equal("stream" in upstreamSends[0], false); + + ws.close(); + await close(server); +});