feat: support opt-in upstream Responses WebSockets

docs-site/src/content/docs/reference/configuration/providers.md

@@ -68,6 +68,7 @@ differing backup and rewrites known legacy namespaced selected ids to bare ids.
 | `baseUrl` | `string` | Upstream API base URL. Most built-in fixed endpoints ignore a mismatch; collision-safe key presets preserve an older same-named custom destination. |
 | `requestPacing?` | `{ enabled, requestsPerMinute?, minIntervalMs?, models? }` | Optional client-side outbound request-start pacing, separate from upstream usage, billing, and rate-limit indicators. RPM is converted to an even interval; `minIntervalMs` may impose a longer interval. Provider limits apply across all models, while `models` entries use exact upstream model IDs (for example `nvidia/llama-3.1-nemotron-ultra-253b-v1`) and can only add delay. Queue waits do not consume the upstream response-header timeout. HTTP, Responses WebSocket, and explicit adapter `fetchResponse`/`runTurn` dispatches are covered. |
 | `upstreamHttpVersion?` | `"auto" \| "http1.1" \| "h1" \| "http2" \| "h2"` | Pin the HTTP version used for upstream requests to this provider. Defaults to `auto`, which lets Bun negotiate. An explicit pin requires an HTTPS target and fails locally when it cannot be honored. Set `http1.1` when a provider's HTTP/2 SSE stream stalls instead of delivering events — the symptom is a long-running streaming request that produces nothing and eventually times out. For Cursor, `http1.1`/`h1` selects its `RunSSE` + `BidiAppend` compatibility transport for inference and also pins live model discovery. Management `POST`/`PATCH` accept `null` to clear it back to `auto`. |
+| `upstreamWebsocket?` | `boolean` | Opt-in upstream Responses WebSocket transport for this provider (default false). When the upstream supports the Responses WebSocket protocol, streaming POST requests to the configured Responses path (default `/v1/responses`) are dialed as WSS over an HTTPS base URL and re-encoded to SSE for the usual pipeline, mirroring the canonical ChatGPT backend optimization for OpenAI-compatible gateways (for example sub2api) whose WebSocket ingress is measurably faster than its SSE queue. Plain HTTP remains on SSE; only Responses endpoints qualify, while other paths stay HTTP. |
 | `responsesPath?` | `string` | Relative resource path for key-auth `openai-responses` requests. It must start with `/` and contain no scheme, query, or fragment. |
 | `supportsServiceTier?` | `boolean` | Tri-state canonical Fast capability fallback. `true` publishes Fast in the catalog, satisfies service-tier routing requirements, contributes a supported fingerprint, and lets fast mode inject the provider's canonical wire value on a compatible final adapter. `false` strips the field and never injects, and exact model declarations cannot reopen it. Absent leaves the provider unclassified: fast mode does not inject or normalize a canonical caller value, and caller values obey the final wire's forwarding permission (`chatServiceTier` on Chat; passthrough on Responses). The registry classifies canonical OpenAI (`true`), DeepSeek, and Volcengine Ark (`false`); set it explicitly only for custom gateways that genuinely support tiers. |
 | `modelSupportsServiceTier?` | `Record<string, boolean>` | Exact upstream model capability overrides. Exact `true` enables canonical Fast for that model; exact `false` narrows provider defaults. An explicit provider-level `supportsServiceTier: false` remains fail-closed and cannot be reopened. Exact `true` does not authorize foreign caller-tier forwarding on Chat. Undeclared models fall back to provider-wide behavior. Management `PATCH /api/providers` merges entries and accepts `null` to clear one. |

src/config.ts

@@ -513,6 +513,10 @@ const providerConfigSchema = z.object({
   upstreamHttpVersion: z.enum(UPSTREAM_HTTP_VERSION_VALUES)
     .nullish()
     .transform(value => value ?? undefined),
+  // Opt-in upstream Responses WebSocket for OpenAI-compatible providers (e.g.
+  // aggregators whose WebSocket ingress is measurably faster than SSE). The
+  // canonical ChatGPT backend WS selection is independent of this flag.
+  upstreamWebsocket: z.boolean().optional(),
   directGeminiWireRenames: z.boolean().optional(),
   noStructuredOutputModels: z.array(z.string().min(1))
     .transform(normalizeNonBlankStringArray)

src/server/management/provider-routes.ts

@@ -228,6 +228,11 @@ function applyProviderPatchFields(
     }
     touched = true;
   }
+  if (Object.hasOwn(rawBody, "upstreamWebsocket")) {
+    if (typeof rawBody.upstreamWebsocket !== "boolean") return { error: "upstreamWebsocket must be a boolean" };
+    next.upstreamWebsocket = rawBody.upstreamWebsocket;
+    touched = true;
+  }
   // The Models page edits the catalog hints in place; keep them on the existing
   // provider mutation path so validation, cache invalidation, and convergence stay unified (#1073).
   if (Object.hasOwn(rawBody, "contextWindow")) {
@@ -455,6 +460,7 @@ export async function handleProviderRoutes(ctx: ManagementContext): Promise<Resp
       modelSupportsServiceTier: p.modelSupportsServiceTier,
       noStructuredOutputModels: p.noStructuredOutputModels,
       upstreamHttpVersion: p.upstreamHttpVersion,
+      upstreamWebsocket: p.upstreamWebsocket === true,
       authMode: p.authMode,
       apiKeyTransport: p.apiKeyTransport,
       disabled: p.disabled === true,
@@ -545,6 +551,10 @@ export async function handleProviderRoutes(ctx: ManagementContext): Promise<Resp
     const name = typeof body.name === "string" ? body.name.trim() : "";
     const providerError = providerManagementConfigError(name, body.provider);
     if (providerError) return jsonResponse({ error: providerError }, 400);
+    const rawProvider = body.provider as Record<string, unknown>;
+    if (rawProvider.upstreamWebsocket !== undefined && typeof rawProvider.upstreamWebsocket !== "boolean") {
+      return jsonResponse({ error: "upstreamWebsocket must be a boolean" }, 400);
+    }
     const serviceTierError = providerServiceTierConfigError(name, body.provider);
     if (serviceTierError) return jsonResponse({ error: serviceTierError }, 400);
     const prov = body.provider ? stripCodexRuntimeProviderFields(body.provider as OcxProviderConfig) : undefined;

src/server/responses/fetch-helpers.ts

@@ -76,7 +76,8 @@ export function providerFetch(
   // transport (measured ~3s faster TTFT than the SSE POST queue); everything
   // else keeps the provider's HTTP fetch. See ws-upstream.ts for the details.
   const unpaced = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
-    if (typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime)) {
+    const upstreamWebsocket = provider.upstreamWebsocket === true;
+    if (typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime, upstreamWebsocket)) {
       // The fallback has to be the same HTTP fetch the non-WS branch would have
       // used, protocol pin included: a WS turn that falls back is serving the
       // request over HTTP, and dropping the provider's `upstreamHttpVersion`

src/server/responses/ws-upstream.ts

@@ -18,6 +18,36 @@ import { compareBunVersions } from "../../lib/bun-stream-caps";
 const CODEX_RESPONSES_HTTP_URL = "https://chatgpt.com/backend-api/codex/responses";
 const CODEX_RESPONSES_WS_URL = "wss://chatgpt.com/backend-api/codex/responses";
 const WS_BETA = "responses_websockets=2026-02-06";
+
+/**
+ * Dial URL for a request URL. The canonical ChatGPT backend keeps its constant;
+ * an operator-opted OpenAI-compatible upstream swaps https for wss on the same
+ * path so gateways that serve the Responses WebSocket protocol on their
+ * /v1/responses path get the same fast lane. Plain HTTP remains on SSE because
+ * a provider WS handshake would otherwise send credentials and request data
+ * without transport encryption.
+ */
+function wsUpstreamUrlFor(httpUrl: string): string {
+  if (httpUrl === CODEX_RESPONSES_HTTP_URL) return CODEX_RESPONSES_WS_URL;
+  return httpUrl.replace(/^http(s?):/, "ws$1:");
+}
+
+/**
+ * An operator-opted OpenAI-compatible upstream only joins the WS lane for
+ * Responses endpoints: the WebSocket path speaks the Responses event protocol,
+ * and every downstream consumer (adapter parsers, usage sniffing, SSE relay)
+ * assumes that wire. Other paths (chat completions, images, search) stay HTTP.
+ */
+function isResponsesWebsocketEligibleUrl(url: string): boolean {
+  let parsed: URL;
+  try {
+    parsed = new URL(url);
+  } catch {
+    return false;
+  }
+  return parsed.protocol === "https:"
+    && parsed.pathname.endsWith("/responses");
+}
 // If the 101 never arrives (network black hole), give SSE a chance well before
 // the caller's connect timeout (default 200s) would fire.
 const UPGRADE_DEADLINE_MS = 10_000;
@@ -102,9 +132,11 @@ export function shouldUseCodexWsUpstream(
   url: string,
   init?: RequestInit,
   runtime: BunRuntimeGateInput = currentBunRuntimeIdentity(),
+  upstreamWebsocketConfigured = false,
 ): boolean {
   if (!bunSupportsBoundedCodexWsRelay(runtime)) return false;
-  if (url !== CODEX_RESPONSES_HTTP_URL) return false;
+  if (url !== CODEX_RESPONSES_HTTP_URL && !upstreamWebsocketConfigured) return false;
+  if (upstreamWebsocketConfigured && !isResponsesWebsocketEligibleUrl(url)) return false;
   if ((init?.method ?? "GET").toUpperCase() !== "POST") return false;
   const body = init?.body;
   if (typeof body !== "string") return false;
@@ -123,6 +155,41 @@ export function shouldUseCodexWsUpstream(
 
 const CLOSED_BEFORE_TERMINAL = "codex websocket closed before a Responses terminal event";
 
+type ResponsesWsRelayEvent = {
+  type: string;
+  text: string;
+};
+
+/**
+ * Responses WebSocket uses `response.done` as its terminal event, while the
+ * SSE Responses surface uses status-specific terminal events. Normalize the
+ * WS-only discriminator before relaying so the existing SSE consumers can
+ * settle the turn and the socket close cannot be mistaken for a drop.
+ */
+function normalizeResponsesWsRelayEvent(text: string): ResponsesWsRelayEvent | null {
+  let payload: unknown;
+  try {
+    payload = JSON.parse(text);
+  } catch {
+    return null;
+  }
+  if (!payload || typeof payload !== "object" || Array.isArray(payload)) return null;
+  const record = payload as Record<string, unknown>;
+  if (typeof record.type !== "string") return null;
+  if (record.type !== "response.done") return { type: record.type, text };
+
+  const response = record.response;
+  const status = response && typeof response === "object" && !Array.isArray(response)
+    ? (response as Record<string, unknown>).status
+    : undefined;
+  const type = status === "failed"
+    ? "response.failed"
+    : status === "incomplete" || status === "cancelled"
+      ? "response.incomplete"
+      : "response.completed";
+  return { type, text: JSON.stringify({ ...record, type }) };
+}
+
 /**
  * The close code is the only thing that separates "the backend refused this
  * payload" from "the network dropped", and both used to reach the caller as the
@@ -221,7 +288,7 @@ export function codexWsUpstreamFetch(
     let ws: WebSocket;
     try {
       // Bun accepts per-handshake headers; the DOM lib types only list protocol arrays.
-      ws = new WebSocket(CODEX_RESPONSES_WS_URL, { headers } as unknown as string[]);
+      ws = new WebSocket(wsUpstreamUrlFor(url), { headers } as unknown as string[]);
     } catch {
       resolve(sseFallback(url, init));
       return;
@@ -311,14 +378,19 @@ export function codexWsUpstreamFetch(
         failStream("codex websocket frame exceeds the response size limit");
         return;
       }
-      const encodedText = encoder.encode(text);
+      const rawEncodedText = encoder.encode(text);
+      if (rawEncodedText.byteLength > MAX_CODEX_WS_FRAME_BYTES) {
+        failStream("codex websocket frame exceeds the response size limit");
+        return;
+      }
+      const normalized = normalizeResponsesWsRelayEvent(text);
+      if (!normalized) return;
+      const { type } = normalized;
+      const encodedText = normalized.text === text ? rawEncodedText : encoder.encode(normalized.text);
       if (encodedText.byteLength > MAX_CODEX_WS_FRAME_BYTES) {
         failStream("codex websocket frame exceeds the response size limit");
         return;
       }
-      let type: unknown;
-      try { type = (JSON.parse(text) as { type?: unknown }).type; } catch { return; }
-      if (typeof type !== "string") return;
       // Relay only the event surface the SSE path produces today. WS-only
       // frames (codex.rate_limits, responsesapi.websocket_timing) are dropped
       // so downstream clients see exactly the stream shape they always got.

src/types/provider.ts

@@ -246,6 +246,17 @@ export interface OcxProviderConfig {
    * (current behavior unchanged). Only meaningful for https: base URLs.
   */
   upstreamHttpVersion?: UpstreamHttpVersion;
+  /**
+   * Opt-in upstream Responses WebSocket transport for this provider. When true,
+   * streaming POST turns to this provider's /v1/responses are dialed over
+   * wss for HTTPS providers and re-encoded to SSE; HTTP providers continue
+   * using SSE. This mirrors the canonical ChatGPT backend
+   * optimization for any OpenAI-compatible gateway that speaks the Responses
+   * WebSocket protocol (for example an aggregator like sub2api whose WS ingress
+   * is measurably faster than its SSE queue). Default false. Canonical ChatGPT
+   * backend WS selection is independent of this flag.
+   */
+  upstreamWebsocket?: boolean;
   /**
    * Google only. When `false`, the AI Studio (direct) path sends Gemini Flash ids
    * unchanged to the wire instead of applying the `-tiered` suffix (`gemini-3.7-flash`

tests/management-provider-validation.test.ts

@@ -3417,7 +3417,7 @@ describe("provider management validation", () => {
   });
 });
 
-describe("provider upstreamHttpVersion management contract (#1668)", () => {
+describe("provider transport option management contract (#1668, #2816)", () => {
   function makeConfig(): OcxConfig {
     return {
       port: 0,
@@ -3640,4 +3640,67 @@ describe("provider upstreamHttpVersion management contract (#1668)", () => {
       upstreamHttpVersion: 42,
     })).toContain("upstreamHttpVersion");
   });
+
+  test("upstreamWebsocket round-trips through POST, GET, and PATCH", async () => {
+    if (existsSync(TEST_DIR)) rmSync(TEST_DIR, { recursive: true });
+    mkdirSync(TEST_DIR, { recursive: true });
+    process.env.OPENCODEX_HOME = TEST_DIR;
+    const liveConfig = makeConfig();
+    saveConfig(liveConfig);
+    await withRequest(liveConfig, async (request) => {
+      const created = await request("/api/providers", {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({
+          name: "ws-provider",
+          provider: {
+            adapter: "openai-responses",
+            baseUrl: "https://api.example.test/v1",
+            upstreamWebsocket: true,
+          },
+        }),
+      });
+      expect(created?.status).toBe(200);
+      expect(liveConfig.providers["ws-provider"]?.upstreamWebsocket).toBe(true);
+      expect(loadConfig().providers["ws-provider"]?.upstreamWebsocket).toBe(true);
+
+      const list = await request("/api/providers");
+      expect(await list?.json()).toContainEqual(expect.objectContaining({
+        name: "ws-provider",
+        upstreamWebsocket: true,
+      }));
+
+      const invalid = await request("/api/providers?name=ws-provider", {
+        method: "PATCH",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ upstreamWebsocket: "true" }),
+      });
+      expect(invalid?.status).toBe(400);
+      expect(liveConfig.providers["ws-provider"]?.upstreamWebsocket).toBe(true);
+
+      const cleared = await request("/api/providers?name=ws-provider", {
+        method: "PATCH",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ upstreamWebsocket: false }),
+      });
+      expect(cleared?.status).toBe(200);
+      expect(liveConfig.providers["ws-provider"]?.upstreamWebsocket).toBe(false);
+      expect(loadConfig().providers["ws-provider"]?.upstreamWebsocket).toBe(false);
+
+      const invalidPost = await request("/api/providers", {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({
+          name: "invalid-ws-provider",
+          provider: {
+            adapter: "openai-responses",
+            baseUrl: "https://api.example.test/v1",
+            upstreamWebsocket: "true",
+          },
+        }),
+      });
+      expect(invalidPost?.status).toBe(400);
+      expect(liveConfig.providers["invalid-ws-provider"]).toBeUndefined();
+    });
+  });
 });