fix(deepseek): apply bounded JSON policy on HTTP SSE for Flash

src/providers/registry.ts

@@ -154,11 +154,12 @@ export interface ProviderRegistryEntry {
    */
   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.
+   * 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<string, boolean>;
   /**
@@ -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<OcxProviderConfig, "baseUrl" | "adapter"> & Partial<Pick<OcxProviderConfig, "authMode">>,

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, unknown>): string {
+  const output = Array.isArray(response.output) ? response.output : [];
+  const frames: Array<Record<string, unknown>> = [
+    {
+      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<void> {
-  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 "<provider>/" 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<string, unknown>).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<string, unknown>;
+        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<string, unknown>;
+        } 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,

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. */

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

tests/deepseek-inbound-wire.test.ts

@@ -112,18 +112,56 @@ 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<Response> {
+    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");
   });
 
   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 () => {