diff --git a/src/providers/registry.ts b/src/providers/registry.ts index 87a83b7d8b..0042f223c1 100644 --- a/src/providers/registry.ts +++ b/src/providers/registry.ts @@ -160,7 +160,7 @@ export interface ProviderRegistryEntry { * reframes as Responses events. Use only for upstreams whose streaming response * can omit or indefinitely delay the terminal event. */ - modelWebsocketUpstreamStreaming?: Record; + modelResponsesUpstreamStreaming?: Record; /** * Responses-API resource path for providers whose route is not `/v1/responses`. * Unlike `modelWireDefaults` above, this IS seeded into saved config: it describes @@ -1143,7 +1143,7 @@ export const PROVIDER_REGISTRY: readonly ProviderRegistryEntry[] = [ // 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 }, + modelResponsesUpstreamStreaming: { "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. @@ -1815,15 +1815,15 @@ 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( +/** Resolve a registry-only upstream-streaming compatibility hint for Responses turns. */ +export function providerModelResponsesUpstreamStreaming( 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()]; + if (!entry?.modelResponsesUpstreamStreaming || !providerMatchesRegistryTransport(id, provider)) return undefined; + return entry.modelResponsesUpstreamStreaming[modelId.trim().toLowerCase()]; } /** diff --git a/src/server/responses-item-id-repair.ts b/src/server/responses-item-id-repair.ts index 1f0397e8e6..1e399154bc 100644 --- a/src/server/responses-item-id-repair.ts +++ b/src/server/responses-item-id-repair.ts @@ -222,3 +222,25 @@ export function hasResponsesItemIdRepair(config: ResponsesItemIdRepairConfig | u || (config?.message?.length ?? 0) > 0 || (config?.reasoning?.length ?? 0) > 0; } + +/** + * Client-facing id normalization for a WHOLE bounded-JSON Responses object. + * + * The bounded-JSON policy (#875) answers a streaming client by synthesizing SSE + * from a completed JSON body, and reframes the same body into events for WS + * turns. Neither path goes through the SSE relay, so neither picks up the SSE + * item-id rewrite — a provider that needs id repair would get it on a streaming + * response and silently lose it the moment the reliability policy switched the + * upstream to bounded JSON. This applies the same rewrite to the object so all + * three paths agree. Raw recorded state is untouched: recording happens before + * any normalization. + */ +export function repairResponsesJsonItemIds( + response: Record, + config: ResponsesItemIdRepairConfig, + budget?: TranslatorBudget, +): Record { + const state = createRepairState(config, budget); + const rewritten = rewriteResponseSnapshot(state, response); + return rewritten.changed ? rewritten.response : response; +} diff --git a/src/server/responses-json-events.ts b/src/server/responses-json-events.ts new file mode 100644 index 0000000000..9910d17d54 --- /dev/null +++ b/src/server/responses-json-events.ts @@ -0,0 +1,52 @@ +/** + * Shared bounded-JSON → Responses event sequence (#875): the same pure event + * list used by the WebSocket bridge (sendResponsesJsonAsEvents) and by the + * HTTP SSE synthesis for models whose reliability policy forces a bounded + * JSON upstream. One algorithm, two serializations — no duplicated drift. + */ + +export type ResponsesJsonEventFrame = Record; + +/** + * The canonical minimal sequence Codex commits: response.created (empty + * output, in_progress) → one response.output_item.done per output item → a + * status-preserving terminal (completed / failed / incomplete). + */ +export function responsesJsonEventSequence( + response: Record, + rewritePayload?: (payload: Record) => Record, +): ResponsesJsonEventFrame[] { + const rewrite = rewritePayload ?? ((payload: Record) => payload); + const output = Array.isArray(response.output) ? response.output : []; + const finalStatus = response.status === "failed" || response.status === "incomplete" + ? response.status + : "completed"; + return [ + rewrite({ + type: "response.created", + response: { ...response, status: "in_progress", output: [] }, + }), + ...output.map((item, outputIndex) => rewrite({ + type: "response.output_item.done", + output_index: outputIndex, + item, + })), + rewrite({ + type: `response.${finalStatus}`, + response: { ...response, status: finalStatus }, + }), + ]; +} + +/** + * Serialize the event sequence as one SSE body with exactly one + * `data: [DONE]\n\n` trailer. + */ +export function responsesJsonToSseBody( + response: Record, + rewritePayload?: (payload: Record) => Record, +): string { + const frames = responsesJsonEventSequence(response, rewritePayload) + .map(frame => `data: ${JSON.stringify(frame)}\n\n`); + return `${frames.join("")}data: [DONE]\n\n`; +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 383c24b504..f1e20a6ba8 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -102,7 +102,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 { providerModelWebsocketUpstreamStreaming, type InboundWire } from "../../providers/registry"; +import { providerModelResponsesUpstreamStreaming, type InboundWire } from "../../providers/registry"; import type { AdapterRequest } from "../../adapters/base"; import { hasKeyPoolFailover, @@ -163,6 +163,7 @@ import { cancelBodyOnAbort } from "../../lib/abort"; import { createResponsesItemIdPayloadRewrite, hasResponsesItemIdRepair, + repairResponsesJsonItemIds, } from "../responses-item-id-repair"; import { createImageGenCallRestoreRewrite, @@ -187,6 +188,7 @@ import { payloadRewriteAsBlockRewrite, relaySseWithBlockRewrite, } from "../sse-payload-rewrite"; +import { responsesJsonToSseBody } from "../responses-json-events"; import { guardTerminalEventStream } from "./terminal-guard"; /** @@ -860,9 +862,13 @@ async function applyFinalRouteRequestNormalization(args: { } parsed.modelId = route.modelId; } - const websocketUpstreamStreaming = inboundTransport === "websocket" - ? providerModelWebsocketUpstreamStreaming(route.providerName, route.provider, route.modelId) - : undefined; + // Transport-neutral reliability policy (#875): applies to any Responses + // upstream whose final adapter is openai-responses, not only WS turns. + const responsesUpstreamStreaming = providerModelResponsesUpstreamStreaming( + route.providerName, + route.provider, + route.modelId, + ); // Settle the wire once so logging, fast-mode, auth, and sidecars read the adapter // this request will actually use (#404). @@ -872,7 +878,7 @@ async function applyFinalRouteRequestNormalization(args: { logCtx.providerAdapter = route.provider.adapter; logCtx.routeDecision = route.routeDecision; - if (websocketUpstreamStreaming === false) { + if (responsesUpstreamStreaming === false && route.provider.adapter === "openai-responses") { parsed.stream = false; if (parsed._rawBody && typeof parsed._rawBody === "object") { (parsed._rawBody as Record).stream = false; @@ -1501,6 +1507,11 @@ async function handleResponsesInner( ); } + // Captured before normalization: whether the CLIENT asked for SSE. The + // transport-neutral upstream-streaming policy below may force a bounded JSON + // upstream for reliability (#875); the answer must then be reframed to SSE + // for streaming clients. + const clientRequestedStream = parsed.stream; await applyFinalRouteRequestNormalization({ parsed, route, @@ -2235,7 +2246,54 @@ async function handleResponsesInner( } return repairResponsesSnapshotJson(restored, outbound); })(); - return new Response(clientJson, { + // #875: the transport-neutral reliability policy forced a bounded JSON + // upstream for a client that asked for SSE. Reframe the completed JSON + // as the canonical terminal SSE sequence (created → output_item.done → + // terminal → [DONE]) so Codex commits the turn instead of hanging on a + // stream that never closes. Non-streaming clients keep the plain JSON. + if (clientRequestedStream === true + && options.inboundTransport !== "websocket" + && providerModelResponsesUpstreamStreaming(route.providerName, route.provider, route.modelId) === false + && route.provider.adapter === "openai-responses") { + try { + let completed = JSON.parse(clientJson) as Record; + // The bounded-JSON answer bypasses the SSE relay, so it also bypasses + // the SSE item-id rewrite. Apply the same client-facing normalization + // here or this policy would silently disable id repair for the very + // providers that need it (raw record already happened above). + if (hasResponsesItemIdRepair(route.provider.responsesItemIdRepair)) { + completed = repairResponsesJsonItemIds(completed, route.provider.responsesItemIdRepair!, translatorBudget); + } + const sseHeaders = sanitizePassthroughHeaders(headers); + sseHeaders.set("content-type", "text/event-stream"); + sseHeaders.set("cache-control", "no-store"); + return new Response(responsesJsonToSseBody(completed), { + status: upstreamResponse.status, + statusText: upstreamResponse.statusText, + headers: sseHeaders, + }); + } catch { + // Non-JSON despite content-type: fall through to the plain relay. + } + } + // WS turns reframe this JSON into events in the bridge, which is the + // other relay-free path — normalize ids so both bounded-JSON paths agree. + const outboundJson = options.inboundTransport === "websocket" + && providerModelResponsesUpstreamStreaming(route.providerName, route.provider, route.modelId) === false + && hasResponsesItemIdRepair(route.provider.responsesItemIdRepair) + ? (() => { + try { + return JSON.stringify(repairResponsesJsonItemIds( + JSON.parse(clientJson) as Record, + route.provider.responsesItemIdRepair!, + translatorBudget, + )); + } catch { + return clientJson; + } + })() + : clientJson; + return new Response(outboundJson, { status: upstreamResponse.status, statusText: upstreamResponse.statusText, headers, diff --git a/src/server/ws-bridge.ts b/src/server/ws-bridge.ts index eb989d21fb..23631fd525 100644 --- a/src/server/ws-bridge.ts +++ b/src/server/ws-bridge.ts @@ -1,4 +1,5 @@ import type { ServerWebSocket } from "bun"; +import { responsesJsonEventSequence } from "./responses-json-events"; import { FORWARD_HEADERS } from "../adapters/openai-responses"; import type { CodexAuthContext } from "../codex/auth-context"; import { headersForCodexAuthContext } from "../codex/auth-context"; @@ -307,25 +308,12 @@ export function sendResponsesJsonAsEvents( } sendTextFrame(ws, text); }; - const output = Array.isArray(response.output) ? response.output : []; - sendObservedFrame({ - type: "response.created", - response: { ...response, status: "in_progress", output: [] }, - }); - output.forEach((item, outputIndex) => { - sendObservedFrame({ - type: "response.output_item.done", - output_index: outputIndex, - item, - }); - }); const finalStatus = response.status === "failed" || response.status === "incomplete" ? response.status : "completed"; - sendObservedFrame({ - type: `response.${finalStatus}` as "response.completed" | "response.failed" | "response.incomplete", - response: { ...response, status: finalStatus }, - }); + for (const frame of responsesJsonEventSequence(response)) { + sendObservedFrame(frame); + } onTerminal?.(finalStatus); } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 7653876ee0..ed401de198 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -206,10 +206,13 @@ 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 (`modelResponsesUpstreamStreaming`) may ask the +upstream Responses endpoint for bounded JSON on ANY client transport — WebSocket or ordinary +HTTP/SSE. The bridge reframes that JSON into the same Responses event sequence +(`src/server/responses-json-events.ts`): WS turns send the frames as WebSocket messages, while +HTTP clients that requested streaming receive a synthesized terminal SSE body (created → +output_item.done → terminal → `[DONE]`). DeepSeek V4 Flash uses this path 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 4e7693c7b3..b560c5c2b3 100644 --- a/tests/deepseek-inbound-wire.test.ts +++ b/tests/deepseek-inbound-wire.test.ts @@ -132,8 +132,177 @@ 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("a Codex WebSocket turn keeps plain JSON downstream (no SSE synthesis)", async () => { + globalThis.fetch = (async () => Response.json({ + id: "resp_deepseek", + object: "response", + status: "completed", + output: [], + })) as typeof fetch; + const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: MODEL, input: "ping", stream: true }), + }), + config, + { model: "", provider: "" }, + { inboundTransport: "websocket" }, + ); + expect(response.headers.get("content-type")).not.toContain("text/event-stream"); + }); + + test("ordinary HTTP Responses requests also use bounded JSON upstream (#875)", async () => { + // The reliability policy is transport-neutral: DeepSeek's Responses stream can + // deliver output without a terminal, so HTTP turns get the same bounded JSON + // upstream as WS turns — and a synthesized terminal SSE back. + const request = await drive("responses"); + expect(request.body.stream).toBe(false); + }); + + test("an HTTP streaming client receives a synthesized terminal SSE instead of a stall (#875)", async () => { + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? "{}")) as { stream?: boolean }; + if (body.stream === true) { + // Old world: a terminal-less SSE that never closes — the stall the issue + // reported. The policy must never send stream:true, so fail loudly here. + return new Response(new ReadableStream({ start() {} }), { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); + } + return Response.json({ + id: "resp_deepseek", + object: "response", + status: "completed", + output: [{ + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "search", + arguments: "{\"q\":\"docs\"}", + status: "completed", + }], + }); + }) as typeof fetch; + + const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; + const deadline = AbortSignal.timeout(5_000); + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: MODEL, input: "ping", stream: true }), + }), + config, + { model: "", provider: "" }, + { abortSignal: deadline }, + ); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + const text = await response.text(); + const sequence = [...text.matchAll(/"type":"(response\.[^"]+)"/g)].map(match => match[1]); + expect(sequence).toEqual([ + "response.created", + "response.output_item.done", + "response.completed", + ]); + expect(text).toContain("data: [DONE]"); + // The function-call item survives with id/call_id byte-identical. + expect(text).toContain('"fc_1"'); + expect(text).toContain('"call_1"'); + }); + + /** + * Review finding on this layer: the bounded-JSON answer never touches the SSE + * relay, so it never picks up the relay's item-id rewrite. Without the + * normalization added here, enabling this reliability policy would silently + * DISABLE id repair for a provider that has it configured — the client would + * get canonical ids while streaming and placeholder ids the moment the policy + * switched the upstream to bounded JSON. + */ + function repairingProvider(): OcxProviderConfig { + return { + ...deepseekProvider(), + responsesItemIdRepair: { message: ["msg_placeholder"], reasoning: ["rs_placeholder"] }, + } as OcxProviderConfig; + } + + function completedWithPlaceholderIds(): Response { + return Response.json({ + id: "resp_deepseek", + object: "response", + status: "completed", + output: [ + { type: "reasoning", id: "rs_placeholder", summary: [] }, + { + type: "message", + id: "msg_placeholder", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "hello" }], + }, + ], + }); + } + + test("the synthesized terminal SSE carries repaired item ids, not the upstream placeholders", async () => { + globalThis.fetch = (async () => completedWithPlaceholderIds()) as typeof fetch; + const config = { providers: { deepseek: repairingProvider() } } as unknown as OcxConfig; + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: MODEL, input: "ping", stream: true }), + }), + config, + { model: "", provider: "" }, + { abortSignal: AbortSignal.timeout(5_000) }, + ); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + const text = await response.text(); + expect(text).not.toContain("msg_placeholder"); + expect(text).not.toContain("rs_placeholder"); + expect(text).toMatch(/"id":"msg_ocx_[0-9a-f]{8}/); + expect(text).toMatch(/"id":"rs_ocx_[0-9a-f]{8}/); + }); + + test("the WebSocket bounded-JSON reframe carries the same repaired ids", async () => { + globalThis.fetch = (async () => completedWithPlaceholderIds()) as typeof fetch; + const config = { providers: { deepseek: repairingProvider() } } as unknown as OcxConfig; + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: MODEL, input: "ping", stream: true }), + }), + config, + { model: "", provider: "" }, + { inboundWire: "responses", inboundTransport: "websocket" }, + ); + const text = await response.text(); + expect(text).not.toContain("msg_placeholder"); + expect(text).not.toContain("rs_placeholder"); + expect(text).toMatch(/"id":"msg_ocx_[0-9a-f]{8}/); + }); + + test("a provider without id repair keeps the bounded-JSON body byte-identical", async () => { + globalThis.fetch = (async () => completedWithPlaceholderIds()) as typeof fetch; + const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: MODEL, input: "ping" }), + }), + config, + { model: "", provider: "" }, + { inboundWire: "responses", inboundTransport: "websocket" }, + ); + const text = await response.text(); + expect(text).toContain("msg_placeholder"); + expect(text).toContain("rs_placeholder"); }); test("an oversized upstream JSON body fails closed instead of buffering without limit", async () => { diff --git a/tests/responses-json-events.test.ts b/tests/responses-json-events.test.ts new file mode 100644 index 0000000000..e477f02497 --- /dev/null +++ b/tests/responses-json-events.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, test } from "bun:test"; +import { + responsesJsonEventSequence, + responsesJsonToSseBody, +} from "../src/server/responses-json-events"; + +describe("responsesJsonEventSequence", () => { + test("completed: created → per-item done → completed terminal", () => { + const frames = responsesJsonEventSequence({ + id: "r1", + status: "completed", + output: [{ type: "message", id: "m1" }, { type: "function_call", id: "fc1" }], + }); + expect(frames.map(frame => frame.type)).toEqual([ + "response.created", + "response.output_item.done", + "response.output_item.done", + "response.completed", + ]); + const created = frames[0]!.response as Record; + expect(created.status).toBe("in_progress"); + expect(created.output).toEqual([]); + expect(frames[1]!.output_index).toBe(0); + expect(frames[2]!.output_index).toBe(1); + expect((frames[3]!.response as Record).status).toBe("completed"); + }); + + test("failed and incomplete statuses are preserved, not upgraded", () => { + for (const status of ["failed", "incomplete"]) { + const frames = responsesJsonEventSequence({ id: "r", status, output: [] }); + expect(frames.at(-1)!.type).toBe(`response.${status}`); + expect((frames.at(-1)!.response as Record).status).toBe(status); + } + }); + + test("empty and non-array outputs yield the minimal sequence", () => { + expect(responsesJsonEventSequence({ id: "r" }).map(frame => frame.type)) + .toEqual(["response.created", "response.completed"]); + expect(responsesJsonEventSequence({ id: "r", output: null }).map(frame => frame.type)) + .toEqual(["response.created", "response.completed"]); + }); + + test("the payload rewrite hook runs on every frame (060 seam)", () => { + const frames = responsesJsonEventSequence( + { id: "r", status: "completed", output: [{ type: "message", id: "uuid-1" }] }, + payload => ({ ...payload, stamped: true }), + ); + expect(frames.every(frame => frame.stamped === true)).toBe(true); + }); +}); + +describe("responsesJsonToSseBody", () => { + test("serializes the sequence with exactly one [DONE] trailer", () => { + const body = responsesJsonToSseBody({ id: "r", status: "completed", output: [{ type: "message", id: "m" }] }); + const frames = body.split("\n\n").filter(part => part.trim().length > 0); + expect(frames).toHaveLength(4); + expect(frames[0]).toContain('"type":"response.created"'); + expect(frames[1]).toContain('"type":"response.output_item.done"'); + expect(frames[2]).toContain('"type":"response.completed"'); + expect(frames[3]).toBe("data: [DONE]"); + expect(body.endsWith("data: [DONE]\n\n")).toBe(true); + }); +});