From f0f930f4453405b9f7224829bb93e5326830a279 Mon Sep 17 00:00:00 2001 From: miles_tian Date: Sun, 2 Aug 2026 20:35:56 +0800 Subject: [PATCH] fix DeepSeek Responses over Codex WebSocket --- src/providers/registry.ts | 23 +++++++++++ src/server/index.ts | 1 + src/server/responses/core.ts | 28 ++++++++++++-- structure/04_transports-and-sidecars.md | 5 +++ tests/deepseek-inbound-wire.test.ts | 51 +++++++++++++++++-------- 5 files changed, 90 insertions(+), 18 deletions(-) diff --git a/src/providers/registry.ts b/src/providers/registry.ts index e336c2f39..e64491566 100644 --- a/src/providers/registry.ts +++ b/src/providers/registry.ts @@ -148,6 +148,14 @@ export interface ProviderRegistryEntry { * of paying a translation hop. */ modelWireDefaults?: Record; + /** + * 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; /** * 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 & Partial>, + 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 diff --git a/src/server/index.ts b/src/server/index.ts index c8d7968c1..0b1e323d8 100644 --- a/src/server/index.ts +++ b/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), diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index e714b3d81..aa002c94d 100644 --- a/src/server/responses/core.ts +++ b/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 { - 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 "/" namespace. if (route.modelId !== parsed.modelId) { @@ -786,6 +789,10 @@ 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); @@ -793,6 +800,13 @@ async function applyFinalRouteRequestNormalization(args: { 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).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); diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 5a39c59e2..3334b6d0f 100644 --- a/structure/04_transports-and-sidecars.md +++ b/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. diff --git a/tests/deepseek-inbound-wire.test.ts b/tests/deepseek-inbound-wire.test.ts index b8403f025..2bf898549 100644 --- a/tests/deepseek-inbound-wire.test.ts +++ b/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 }> { + const requests: Array<{ url: string; body: Record }> = []; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + requests.push({ + url: String(input), + body: JSON.parse(String(init?.body ?? "{}")) as Record, + }); + 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 { - const urls = captureUpstreamUrl(); + async function drive( + inboundWire?: "responses" | "chat" | "anthropic", + inboundTransport?: "websocket", + ): Promise<{ url: string; body: Record }> { + 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); }); });