diff --git a/src/providers/registry.ts b/src/providers/registry.ts index 87a83b7d8..3b9f203e6 100644 --- a/src/providers/registry.ts +++ b/src/providers/registry.ts @@ -154,11 +154,12 @@ export interface ProviderRegistryEntry { */ 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. + * Registry-only per-model override for the upstream Responses request shape. + * `false` asks the upstream for a bounded JSON body instead of an open-ended + * event stream, then the proxy reframes that JSON into a complete Responses + * event sequence for the client (WebSocket frames or HTTP SSE). Use only for + * upstreams whose streaming response can omit or indefinitely delay the + * terminal event. */ modelWebsocketUpstreamStreaming?: Record; /** @@ -1141,8 +1142,8 @@ export const PROVIDER_REGISTRY: readonly ProviderRegistryEntry[] = [ "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. + // terminal event. Prefer the provider's bounded JSON response upstream so the + // proxy can synthesize a complete client event sequence (HTTP SSE or WS). 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` @@ -1815,7 +1816,10 @@ export function providerModelWireDefault( return wire !== undefined && allowedWires.has(wire) ? wire : undefined; } -/** Resolve a registry-only upstream-streaming compatibility hint for WS turns. */ +/** + * Resolve a registry-only upstream-streaming compatibility hint for Responses turns. + * Historically named for the WebSocket path; callers may now apply it to HTTP too. + */ export function providerModelWebsocketUpstreamStreaming( id: string, provider: Pick & Partial>, diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 619a2e7f0..4b8876f46 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -825,6 +825,42 @@ async function resolveResponsesCodexAuth( * Apply every route-dependent request mutation against the final selected route. * Must run only after subagent fallback has settled the model/provider. */ + +/** + * Reframe a completed Responses JSON body into a minimal SSE sequence that always + * ends on a terminal event. Used when registry policy forces bounded upstream JSON + * while the client still requested stream=true (HTTP/SSE Codex path). + */ +function responsesJsonToClientSse(response: Record): string { + const output = Array.isArray(response.output) ? response.output : []; + const frames: Array> = [ + { + type: "response.created", + response: { ...response, status: "in_progress", output: [] }, + }, + ]; + output.forEach((item, outputIndex) => { + frames.push({ + type: "response.output_item.done", + output_index: outputIndex, + item, + }); + }); + const finalStatus = response.status === "failed" || response.status === "incomplete" + ? response.status + : "completed"; + frames.push({ + type: `response.${finalStatus}`, + response: { ...response, status: finalStatus }, + }); + return frames + .map(frame => { + const type = typeof frame.type === "string" ? frame.type : "message"; + return `event: ${type}\ndata: ${JSON.stringify(frame)}\n\n`; + }) + .join(""); +} + async function applyFinalRouteRequestNormalization(args: { parsed: OcxParsedRequest; route: RouteResult; @@ -834,7 +870,9 @@ async function applyFinalRouteRequestNormalization(args: { inboundWire: InboundWire; inboundTransport?: "websocket"; }): Promise { - const { parsed, route, config, req, logCtx, inboundWire, inboundTransport } = args; + // inboundTransport remains part of the call shape for downstream client framing, + // but the upstream registry compatibility policy below is no longer WS-only. + const { parsed, route, config, req, logCtx, inboundWire } = args; // Apply the routed model id upstream: routing may strip a "/" namespace. if (route.modelId !== parsed.modelId) { @@ -843,9 +881,11 @@ async function applyFinalRouteRequestNormalization(args: { } parsed.modelId = route.modelId; } - const websocketUpstreamStreaming = inboundTransport === "websocket" - ? providerModelWebsocketUpstreamStreaming(route.providerName, route.provider, route.modelId) - : undefined; + // Capture the client-facing stream preference before any registry compatibility + // rewrite. HTTP clients still expect SSE even when upstream JSON is forced. + if (parsed._clientRequestedStream === undefined) { + parsed._clientRequestedStream = parsed.stream; + } // Settle the wire once so logging, fast-mode, auth, and sidecars read the adapter // this request will actually use (#404). @@ -854,7 +894,18 @@ async function applyFinalRouteRequestNormalization(args: { logCtx.provider = route.providerName; logCtx.providerAdapter = route.provider.adapter; - if (websocketUpstreamStreaming === false) { + // Some Responses upstreams can emit output without a terminal event. Apply the + // bounded-JSON compatibility policy only after the final wire is known, so Chat + // and Anthropic replays that remain on openai-chat keep their streaming contract. + const responsesUpstreamStreaming = route.provider.adapter === "openai-responses" + ? providerModelWebsocketUpstreamStreaming( + route.providerName, + route.provider, + route.modelId, + ) + : undefined; + + if (responsesUpstreamStreaming === false) { parsed.stream = false; if (parsed._rawBody && typeof parsed._rawBody === "object") { (parsed._rawBody as Record).stream = false; @@ -2125,9 +2176,9 @@ async function handleResponsesInner( // Bounded whole-body read: a non-streaming upstream JSON body is fully materialized // here (and again by the request-log finalizer and the WebSocket bridge's reframing), // so an unbounded .text() would let a hostile or stuck upstream grow proxy memory - // without limit. This path is no longer rare — WebSocket turns for models whose - // streaming terminal event is unreliable are deliberately answered with bounded JSON. - // Oversize and stall deadlines both fail closed; a partial body is never parsed. + // without limit. This path is no longer rare — models whose streaming terminal event + // is unreliable are deliberately answered with bounded JSON, then reframed for the + // client. Oversize and stall deadlines both fail closed; a partial body is never parsed. const bounded = await readBoundedResponseBody(upstreamResponse, { maxBytes: MAX_UPSTREAM_JSON_BODY_BYTES, totalTimeoutMs: UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS, @@ -2146,7 +2197,33 @@ async function handleResponsesInner( rememberPassthroughResponse(JSON.parse(text) as { id?: unknown; output?: unknown; status?: unknown }); } catch { /* non-JSON despite content-type; recording is best-effort */ } } - return new Response(restoreImageGenCallsInJson(text, imageGenCallAliases), { + const restoredText = restoreImageGenCallsInJson(text, imageGenCallAliases); + // HTTP/SSE clients requested a stream; synthesize a complete event sequence so + // Codex never waits on an upstream that can omit the terminal frame. WebSocket + // turns keep the bounded JSON response intact for sendResponsesJsonAsEvents(). + if (parsed._clientRequestedStream === true && options.inboundTransport !== "websocket") { + let responseJson: Record; + try { + const parsedJson: unknown = JSON.parse(restoredText); + if (typeof parsedJson !== "object" || parsedJson === null || Array.isArray(parsedJson)) { + return formatErrorResponse(502, "upstream_error", "upstream returned malformed JSON"); + } + responseJson = parsedJson as Record; + } catch { + return formatErrorResponse(502, "upstream_error", "upstream returned malformed JSON"); + } + const sseHeaders = new Headers(headers); + sseHeaders.delete("content-length"); + sseHeaders.delete("content-encoding"); + sseHeaders.set("content-type", "text/event-stream; charset=utf-8"); + sseHeaders.set("cache-control", "no-store"); + return new Response(responsesJsonToClientSse(responseJson), { + status: upstreamResponse.status, + statusText: upstreamResponse.statusText, + headers: sseHeaders, + }); + } + return new Response(restoredText, { status: upstreamResponse.status, statusText: upstreamResponse.statusText, headers, diff --git a/src/types.ts b/src/types.ts index ae84aa674..0b69e9014 100644 --- a/src/types.ts +++ b/src/types.ts @@ -7,6 +7,11 @@ export interface OcxParsedRequest { previousResponseId?: string; context: OcxContext; stream: boolean; + /** + * Client-facing stream preference captured before registry compatibility policy + * rewrites `stream` for unreliable upstream event streams. + */ + _clientRequestedStream?: boolean; options: OcxRequestOptions; _rawBody?: unknown; /** Number of leading raw input items restored from local previous_response_id state. */ diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 7653876ee..47870ea63 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -206,10 +206,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. +Registry-declared per-model compatibility hints may ask the upstream Responses endpoint for bounded +JSON when that upstream's streaming terminal event is unreliable. The proxy reframes that JSON into +a complete Responses event sequence for the client (WebSocket frames or HTTP SSE). DeepSeek V4 Flash +uses this path on both transports because its Codex streaming response can deliver output without +closing on a terminal event. `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 diff --git a/tests/deepseek-inbound-wire.test.ts b/tests/deepseek-inbound-wire.test.ts index 4e7693c7b..0f360a25f 100644 --- a/tests/deepseek-inbound-wire.test.ts +++ b/tests/deepseek-inbound-wire.test.ts @@ -112,6 +112,40 @@ describe("the inbound scope survives the handleResponses replay", () => { return requests[0] ?? { url: "", body: {} }; } + async function respondWithUpstreamJson( + payload: unknown, + options: { + clientStream?: boolean; + inboundTransport?: "websocket"; + upstreamHeaders?: HeadersInit; + } = {}, + ): Promise { + const upstreamHeaders = new Headers(options.upstreamHeaders); + upstreamHeaders.set("content-type", "application/json"); + globalThis.fetch = (async () => new Response(JSON.stringify(payload), { + status: 200, + headers: upstreamHeaders, + })) as typeof fetch; + const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; + return handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: MODEL, + input: "ping", + stream: options.clientStream ?? true, + }), + }), + config, + { model: "", provider: "" }, + { + inboundWire: "responses", + ...(options.inboundTransport === undefined ? {} : { inboundTransport: options.inboundTransport }), + }, + ); + } + test("a native Responses request reaches the documented /responses route", async () => { expect((await drive("responses")).url).toBe("https://api.deepseek.com/responses"); }); @@ -119,11 +153,15 @@ describe("the inbound scope survives the handleResponses replay", () => { 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")).url).toBe("https://api.deepseek.com/chat/completions"); + const request = await drive("anthropic"); + expect(request.url).toBe("https://api.deepseek.com/chat/completions"); + expect(request.body.stream).toBe(true); }); test("a Chat replay reaches /chat/completions, not /responses", async () => { - expect((await drive("chat")).url).toBe("https://api.deepseek.com/chat/completions"); + const request = await drive("chat"); + expect(request.url).toBe("https://api.deepseek.com/chat/completions"); + expect(request.body.stream).toBe(true); }); test("a Codex WebSocket turn asks DeepSeek for bounded JSON upstream", async () => { @@ -132,8 +170,101 @@ describe("the inbound scope survives the handleResponses replay", () => { expect(request.body.stream).toBe(false); }); - test("ordinary HTTP Responses requests keep streaming upstream", async () => { - expect((await drive("responses")).body.stream).toBe(true); + test("ordinary HTTP Responses turns also force bounded JSON for DeepSeek Flash", async () => { + // Codex Desktop defaults to HTTP/SSE while websockets stay opt-in. Flash still + // needs the terminal-safe upstream path on that transport. + const request = await drive("responses"); + expect(request.url).toBe("https://api.deepseek.com/responses"); + expect(request.body.stream).toBe(false); + }); + + test("HTTP stream clients receive a terminal SSE sequence from bounded JSON", async () => { + const response = await respondWithUpstreamJson({ + id: "resp_deepseek", + object: "response", + status: "completed", + output: [{ + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "shell", + arguments: "{\"command\":\"pwd\"}", + status: "completed", + }], + }, { + upstreamHeaders: { + "content-length": "999", + "content-encoding": "gzip", + }, + }); + expect(response.status).toBe(200); + expect(response.headers.get("content-type") ?? "").toContain("text/event-stream"); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(response.headers.get("content-length")).toBeNull(); + expect(response.headers.get("content-encoding")).toBeNull(); + const body = await response.text(); + expect(body).toContain("event: response.created"); + expect(body).toContain("event: response.output_item.done"); + expect(body).toContain("event: response.completed"); + expect(body).toContain("function_call"); + }); + + test("HTTP stream clients preserve failed terminal status from bounded JSON", async () => { + const response = await respondWithUpstreamJson({ + id: "resp_failed", + object: "response", + status: "failed", + output: [], + error: { code: "server_error", message: "upstream failed" }, + }); + const body = await response.text(); + expect(body).toContain("event: response.failed"); + expect(body).not.toContain("event: response.completed"); + expect(body).toContain("upstream failed"); + }); + + test("HTTP stream clients preserve incomplete terminal status from bounded JSON", async () => { + const response = await respondWithUpstreamJson({ + id: "resp_incomplete", + object: "response", + status: "incomplete", + output: [], + incomplete_details: { reason: "upstream_stall_timeout" }, + }); + const body = await response.text(); + expect(body).toContain("event: response.incomplete"); + expect(body).not.toContain("event: response.completed"); + expect(body).toContain("upstream_stall_timeout"); + }); + + test("HTTP stream clients reject null bounded JSON with a typed upstream error", async () => { + const response = await respondWithUpstreamJson(null); + expect(response.status).toBe(502); + const payload = (await response.json()) as { error?: { code?: string; message?: string } }; + expect(payload.error?.code).toBe("upstream_server_error"); + expect(payload.error?.message).toContain("malformed JSON"); + }); + + test("non-streaming HTTP clients keep the bounded JSON response", async () => { + const response = await respondWithUpstreamJson({ + id: "resp_json", + object: "response", + status: "completed", + output: [], + }, { clientStream: false }); + expect(response.headers.get("content-type") ?? "").toContain("application/json"); + expect(await response.json()).toMatchObject({ id: "resp_json", status: "completed" }); + }); + + test("WebSocket turns keep bounded JSON for the existing WS re-framer", async () => { + const response = await respondWithUpstreamJson({ + id: "resp_ws", + object: "response", + status: "completed", + output: [{ type: "message", role: "assistant", status: "completed", content: [] }], + }, { inboundTransport: "websocket" }); + expect(response.headers.get("content-type") ?? "").toContain("application/json"); + expect(await response.json()).toMatchObject({ id: "resp_ws", status: "completed" }); }); test("an oversized upstream JSON body fails closed instead of buffering without limit", async () => {