Fix DeepSeek Responses over Codex WebSocket

src/providers/registry.ts

@@ -148,6 +148,14 @@ export interface ProviderRegistryEntry {
    * of paying a translation hop.
    */
   modelWireDefaults?: Record<string, ModelWireDefault>;
+  /**
+   * Registry-only per-model override for the upstream request shape used behind a
+   * Codex Responses WebSocket turn. `false` keeps the client-facing WebSocket but
+   * asks the upstream Responses endpoint for bounded JSON, which the bridge then
+   * reframes as Responses events. Use only for upstreams whose streaming response
+   * can omit or indefinitely delay the terminal event.
+   */
+  modelWebsocketUpstreamStreaming?: Record<string, boolean>;
   /**
    * Responses-API resource path for providers whose route is not `/v1/responses`.
    * Unlike `modelWireDefaults` above, this IS seeded into saved config: it describes
@@ -960,6 +968,10 @@ export const PROVIDER_REGISTRY: readonly ProviderRegistryEntry[] = [
       // for no gain.
       "deepseek-v4-flash": { wire: "openai-responses", inbound: ["responses"] },
     },
+    // DeepSeek's Codex Responses stream can deliver output without closing on the
+    // terminal event. Keep Codex on WebSocket, but use the provider's bounded JSON
+    // response upstream so the bridge can synthesize a complete WS event sequence.
+    modelWebsocketUpstreamStreaming: { "deepseek-v4-flash": false },
     // DeepSeek's Responses route is `POST /responses` with no `/v1` segment. Without
     // this the passthrough adapter falls back to its legacy `/v1/responses`
     // construction and the wire above can never route.
@@ -1573,6 +1585,17 @@ export function providerModelWireDefault(
   return wire !== undefined && allowedWires.has(wire) ? wire : undefined;
 }
 
+/** Resolve a registry-only upstream-streaming compatibility hint for WS turns. */
+export function providerModelWebsocketUpstreamStreaming(
+  id: string,
+  provider: Pick<OcxProviderConfig, "baseUrl" | "adapter"> & Partial<Pick<OcxProviderConfig, "authMode">>,
+  modelId: string,
+): boolean | undefined {
+  const entry = getProviderRegistryEntry(id);
+  if (!entry?.modelWebsocketUpstreamStreaming || !providerMatchesRegistryTransport(id, provider)) return undefined;
+  return entry.modelWebsocketUpstreamStreaming[modelId.trim().toLowerCase()];
+}
+
 /**
  * Effective Codex account mode for a provider. For canonical `openai`, a valid persisted
  * `codexAccountMode` on the provider config wins and a missing/invalid value defaults to

src/server/index.ts

@@ -1030,6 +1030,7 @@ export function startServer(port?: number) {
             let terminalRecorder: ((status: ResponsesTerminalStatus, httpStatusOverride?: number) => void) | undefined;
             const response = await handleResponses(req, config, logCtx, {
               forceEmptyResponseId: true,
+              inboundTransport: "websocket",
               abortSignal: turnAbort.signal,
               turnAdmissionLease,
               onFirstOutput: () => recordFirstOutput(logCtx, start),

src/server/responses/core.ts

@@ -95,7 +95,7 @@ import { applyOpenAiVirtualModel, resolveOpenAiCompactModel } from "../../provid
 import { isUsageDebugEnabled } from "../../usage/debug";
 import { readJsonRequestBody, DecompressedBodyTooLargeError, UnsupportedContentEncodingError } from "../request-decompress";
 import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve";
-import type { InboundWire } from "../../providers/registry";
+import { providerModelWebsocketUpstreamStreaming, type InboundWire } from "../../providers/registry";
 import { hasKeyPoolFailover, rotateProviderTransportOn429 } from "../../providers/key-failover";
 import { shouldAttemptImageTierRetry } from "../image-retry";
 import { resolveProviderTransport } from "../../providers/xai-transport";
@@ -537,6 +537,8 @@ export interface HandleResponsesOptions {
    * it. Omitted means a genuine Responses inbound.
    */
   inboundWire?: InboundWire;
+  /** Internal transport identity for route-scoped upstream compatibility policy. */
+  inboundTransport?: "websocket";
   /** Internal recursion guard; callers outside this module must not set it. */
   comboAttempt?: boolean;
   /** Internal combo handoff: allow a later same-provider model after a reset-derived 429/402. */
@@ -776,8 +778,9 @@ async function applyFinalRouteRequestNormalization(args: {
   req: Request;
   logCtx: RequestLogContext;
   inboundWire: InboundWire;
+  inboundTransport?: "websocket";
 }): Promise<void> {
-  const { parsed, route, config, req, logCtx, inboundWire } = args;
+  const { parsed, route, config, req, logCtx, inboundWire, inboundTransport } = args;
 
   // Apply the routed model id upstream: routing may strip a "<provider>/" namespace.
   if (route.modelId !== parsed.modelId) {
@@ -786,13 +789,24 @@ async function applyFinalRouteRequestNormalization(args: {
     }
     parsed.modelId = route.modelId;
   }
+  const websocketUpstreamStreaming = inboundTransport === "websocket"
+    ? providerModelWebsocketUpstreamStreaming(route.providerName, route.provider, route.modelId)
+    : undefined;
+
   // Settle the wire once so logging, fast-mode, auth, and sidecars read the adapter
   // this request will actually use (#404).
   route.provider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire);
   logCtx.model = route.modelId;
   logCtx.provider = route.providerName;
   logCtx.providerAdapter = route.provider.adapter;
 
+  if (websocketUpstreamStreaming === false) {
+    parsed.stream = false;
+    if (parsed._rawBody && typeof parsed._rawBody === "object") {
+      (parsed._rawBody as Record<string, unknown>).stream = false;
+    }
+  }
+
   // Final selected model before virtual wire-model rewriting (Pro aliases).
   const finalSelectedModelId = route.modelId;
 
@@ -1347,7 +1361,15 @@ async function handleResponsesInner(
     );
   }
 
-  await applyFinalRouteRequestNormalization({ parsed, route, config, req, logCtx, inboundWire });
+  await applyFinalRouteRequestNormalization({
+    parsed,
+    route,
+    config,
+    req,
+    logCtx,
+    inboundWire,
+    inboundTransport: options.inboundTransport,
+  });
 
   {
     const finalAuth = await resolveResponsesCodexAuth(req, config, route, options);

structure/04_transports-and-sidecars.md

@@ -177,6 +177,11 @@ the upgrade with 426 so Codex falls back to HTTP cleanly.
 The endpoint handles `response.create`, ignores `response.processed`, supports warmup
 `generate: false`, and feeds the same request pipeline as HTTP/SSE.
 
+Registry-declared per-model compatibility hints may keep the client-facing WebSocket while asking
+the upstream Responses endpoint for bounded JSON. The bridge reframes that JSON into the same
+Responses event sequence. DeepSeek V4 Flash uses this path because its Codex streaming response can
+deliver output without closing on a terminal event; ordinary HTTP/SSE calls remain streaming.
+
 `ws-bridge.ts` preserves upstream `failed` and `incomplete` status values in the final WebSocket
 frame rather than always emitting `response.completed`. If the response status is `failed`, a
 `response.failed` frame is sent; otherwise `response.completed` carries through the original status.

tests/deepseek-inbound-wire.test.ts

@@ -73,20 +73,28 @@ describe("the inbound scope survives the handleResponses replay", () => {
   const originalFetch = globalThis.fetch;
   afterEach(() => { globalThis.fetch = originalFetch; });
 
-  function captureUpstreamUrl(): string[] {
-    const urls: string[] = [];
-    globalThis.fetch = (async (input: RequestInfo | URL) => {
-      urls.push(String(input));
-      return new Response("data: [DONE]\n\n", {
-        status: 200,
-        headers: { "content-type": "text/event-stream" },
+  function captureUpstreamRequests(): Array<{ url: string; body: Record<string, unknown> }> {
+    const requests: Array<{ url: string; body: Record<string, unknown> }> = [];
+    globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
+      requests.push({
+        url: String(input),
+        body: JSON.parse(String(init?.body ?? "{}")) as Record<string, unknown>,
+      });
+      return Response.json({
+        id: "resp_deepseek",
+        object: "response",
+        status: "completed",
+        output: [],
       });
     }) as typeof fetch;
-    return urls;
+    return requests;
   }
 
-  async function drive(inboundWire?: "responses" | "chat" | "anthropic"): Promise<string> {
-    const urls = captureUpstreamUrl();
+  async function drive(
+    inboundWire?: "responses" | "chat" | "anthropic",
+    inboundTransport?: "websocket",
+  ): Promise<{ url: string; body: Record<string, unknown> }> {
+    const requests = captureUpstreamRequests();
     const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig;
     await handleResponses(
       new Request("http://localhost/v1/responses", {
@@ -96,23 +104,36 @@ describe("the inbound scope survives the handleResponses replay", () => {
       }),
       config,
       { model: "", provider: "" },
-      inboundWire === undefined ? {} : { inboundWire },
+      {
+        ...(inboundWire === undefined ? {} : { inboundWire }),
+        ...(inboundTransport === undefined ? {} : { inboundTransport }),
+      },
     );
-    return urls[0] ?? "";
+    return requests[0] ?? { url: "", body: {} };
   }
 
   test("a native Responses request reaches the documented /responses route", async () => {
-    expect(await drive("responses")).toBe("https://api.deepseek.com/responses");
+    expect((await drive("responses")).url).toBe("https://api.deepseek.com/responses");
   });
 
   test("an Anthropic replay reaches /chat/completions, not /responses", async () => {
     // Regression guard for the audit's critical finding: editing only the pre-flight
     // resolution in claude-messages.ts left this URL on /responses.
-    expect(await drive("anthropic")).toBe("https://api.deepseek.com/chat/completions");
+    expect((await drive("anthropic")).url).toBe("https://api.deepseek.com/chat/completions");
   });
 
   test("a Chat replay reaches /chat/completions, not /responses", async () => {
-    expect(await drive("chat")).toBe("https://api.deepseek.com/chat/completions");
+    expect((await drive("chat")).url).toBe("https://api.deepseek.com/chat/completions");
+  });
+
+  test("a Codex WebSocket turn asks DeepSeek for bounded JSON upstream", async () => {
+    const request = await drive("responses", "websocket");
+    expect(request.url).toBe("https://api.deepseek.com/responses");
+    expect(request.body.stream).toBe(false);
+  });
+
+  test("ordinary HTTP Responses requests keep streaming upstream", async () => {
+    expect((await drive("responses")).body.stream).toBe(true);
   });
 });