From 916bc787f709a79cc2d728d3c762fc4d278d95d5 Mon Sep 17 00:00:00 2001 From: PyEL666 <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 23:59:44 +0800 Subject: [PATCH] fix(responses): buffer ChatGPT SSE for non-stream clients and harden free adapters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ChatGPT Codex REST requires stream:true (and typically store:false). When clients send stream:false, force those fields on the wire and buffer the upstream SSE into a JSON response. Rebuild output from streamed output_item / output_text events when response.completed has output:[], which otherwise produces hollow 200s that Codex Desktop surfaces poorly. Also: retry pre-stream 5xx on the routed path (same as passthrough), clamp mimo-free reasoning_effort to low|medium|high, retry free-tier 5xx, and rotate the anonymous mimo client id once on 441 abuse blocks. Observed with Codex Desktop Design B (openai_base_url → local /v1) on Windows against gpt-* passthrough and free models; related to #261. --- src/adapters/mimo-free.ts | 63 +++++++++-- src/adapters/openai-responses.ts | 9 ++ src/server/responses.ts | 183 ++++++++++++++++++++++++++++++- 3 files changed, 239 insertions(+), 16 deletions(-) diff --git a/src/adapters/mimo-free.ts b/src/adapters/mimo-free.ts index e0d6d899aef..62a561fa1e8 100644 --- a/src/adapters/mimo-free.ts +++ b/src/adapters/mimo-free.ts @@ -175,7 +175,19 @@ export function createMimoFreeAdapter(provider: OcxProviderConfig): ProviderAdap // but override the URL and headers after. const baseReq = base.buildRequest(parsed) as AdapterRequest; const baseBody = JSON.parse(baseReq.body as string) as unknown; - const markedBody = injectMimoSystemMarker(baseBody); + // Xiaomi free chat only accepts reasoning_effort in {low, medium, high}. + // Codex Desktop often sends max/xhigh/ultra — clamp so free turns don't 400. + const clampReasoningEffort = (body: unknown): unknown => { + if (!body || typeof body !== "object" || Array.isArray(body)) return body; + const rec = body as Record; + const effort = rec.reasoning_effort; + if (typeof effort !== "string") return body; + const allowed = new Set(["low", "medium", "high"]); + if (allowed.has(effort)) return body; + const mapped = effort === "minimal" || effort === "none" ? "low" : "high"; + return { ...rec, reasoning_effort: mapped }; + }; + const markedBody = clampReasoningEffort(injectMimoSystemMarker(baseBody)); const headers: Record = { "Content-Type": "application/json", @@ -195,31 +207,58 @@ export function createMimoFreeAdapter(provider: OcxProviderConfig): ProviderAdap }, async fetchResponse(request: AdapterRequest, ctx): Promise { - const response = await fetch(request.url, { + const doFetch = (headers: Record) => fetch(request.url, { method: request.method, - headers: request.headers as Record, + headers, body: request.body, signal: ctx?.abortSignal, }); + let headers = request.headers as Record; + let response = await doFetch(headers); + // Retry predicate: 401 (expired/invalid JWT) retries ONCE with a fresh token. // 403 is NOT retried — Xiaomi uses it for anti-abuse "Illegal access" and there is // no documented token-expiry signature that would mark a 403 as retryable. if (response.status === 401) { - // Drain the first response body before issuing the retry. try { await response.body?.cancel(); } catch { /* already consumed */ } resetMimoJwtCache(); const freshJwt = await getMimoJwt(ctx?.abortSignal); - const retryHeaders = { - ...(request.headers as Record), + headers = { + ...headers, "Authorization": `Bearer ${freshJwt}`, }; - return fetch(request.url, { - method: request.method, - headers: retryHeaders, - body: request.body, - signal: ctx?.abortSignal, - }); + response = await doFetch(headers); + } + + // Pre-stream 5xx blips from the free tier — retry a couple times (body is replayable). + for (let attempt = 0; attempt < 2 && (response.status === 500 || response.status === 502 || response.status === 503); attempt++) { + try { await response.body?.cancel(); } catch { /* already consumed */ } + await new Promise(r => setTimeout(r, 400 * (attempt + 1))); + response = await doFetch(headers); + } + + // 441 / high-frequency abuse: rotate anonymous client id + JWT once and retry. + if (response.status === 400 || response.status === 403 || response.status === 441) { + let payloadText = ""; + try { payloadText = await response.clone().text(); } catch { /* ignore */ } + const blocked = response.status === 441 + || /高频|违规|441|Illegal access|rate/i.test(payloadText); + if (blocked) { + try { await response.body?.cancel(); } catch { /* already consumed */ } + resetMimoJwtCache(); + resetMimoClientIdCache(); + try { + const { unlinkSync, existsSync } = await import("node:fs"); + const { join } = await import("node:path"); + const { getConfigDir } = await import("../config"); + const idFile = join(getConfigDir(), "mimo-client-id"); + if (existsSync(idFile)) unlinkSync(idFile); + } catch { /* best-effort file rotate */ } + const freshJwt = await getMimoJwt(ctx?.abortSignal); + headers = { ...headers, "Authorization": `Bearer ${freshJwt}` }; + response = await doFetch(headers); + } } return response; diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index 3e5693b86f2..493f92f4df0 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -454,6 +454,15 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): ); if (forward) outBody = repairOrphanedInputItems(outBody, unexpandedMiss); else outBody = stripConflictingHostedTools(outBody); + // ChatGPT Codex REST rejects non-stream turns with {"detail":"Stream must be set to true"} + // and often requires store:false for unstored item ids. Force stream:true on the wire; + // the responses handler buffers SSE → JSON when the client asked for stream:false. + if (forward && isPlainObject(outBody)) { + const next: Record = { ...outBody }; + if (next.stream !== true) next.stream = true; + if (next.store !== false) next.store = false; + outBody = next; + } return { url, method: "POST", diff --git a/src/server/responses.ts b/src/server/responses.ts index 616452caabf..34e7f5dedb5 100644 --- a/src/server/responses.ts +++ b/src/server/responses.ts @@ -55,7 +55,7 @@ import { recordCodexUpstreamOutcome, type CodexUpstreamOutcome, } from "../codex/routing"; -import { fetchWithResetRetry, fetchWithTransientRetry, applyUpstreamRecoveryInit } from "../lib/upstream-retry"; +import { fetchWithTransientRetry, applyUpstreamRecoveryInit } from "../lib/upstream-retry"; import { ForwardAdmissionCredentialError, validateForwardAdmissionCredential } from "./auth-cors"; import { listOpenAiForwardSidecarCandidates, resolveFirstUsableOpenAiSidecar, type ResolvedOpenAiForwardSidecar } from "../providers/openai-sidecar"; import { applyOpenAiVirtualModel, resolveOpenAiCompactModel } from "../providers/openai-virtual-models"; @@ -92,6 +92,153 @@ import { sanitizePassthroughHeaders, } from "./relay"; +/** + * Drain a Responses SSE body and return a completed `response` JSON object. + * ChatGPT Codex often emits `response.completed` with `output: []` even when + * text/tool items streamed earlier — reconstruct output from item/text events + * so stream:false clients receive a usable non-stream body. + */ +export async function bufferResponsesSseToJson( + body: ReadableStream, + signal?: AbortSignal, +): Promise<{ response?: Record; error?: string }> { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let finalResponse: Record | undefined; + let createdResponse: Record | undefined; + let error: string | undefined; + const itemsByIndex = new Map>(); + const textByItemId = new Map(); + let looseDeltaText = ""; + let looseDoneText = ""; + + const handle = (payload: string): void => { + if (!payload || payload === "[DONE]") return; + let data: Record; + try { + data = JSON.parse(payload) as Record; + } catch { + return; + } + const type = typeof data.type === "string" ? data.type : ""; + if (type === "response.created" || type === "response.in_progress") { + const resp = data.response; + if (resp && typeof resp === "object" && !Array.isArray(resp)) { + createdResponse = resp as Record; + } + } else if (type === "response.output_item.added" || type === "response.output_item.done") { + const item = data.item; + const idx = typeof data.output_index === "number" ? data.output_index : itemsByIndex.size; + if (item && typeof item === "object" && !Array.isArray(item)) { + itemsByIndex.set(idx, item as Record); + } + } else if (type === "response.output_text.delta" && typeof data.delta === "string") { + const itemId = typeof data.item_id === "string" ? data.item_id : ""; + if (itemId) textByItemId.set(itemId, (textByItemId.get(itemId) ?? "") + data.delta); + else looseDeltaText += data.delta; + } else if (type === "response.output_text.done" && typeof data.text === "string") { + const itemId = typeof data.item_id === "string" ? data.item_id : ""; + if (itemId) textByItemId.set(itemId, data.text); + else looseDoneText = data.text; + } else if (type === "response.completed" || type === "response.done") { + const resp = data.response; + if (resp && typeof resp === "object" && !Array.isArray(resp)) { + finalResponse = resp as Record; + } + } else if (type === "response.failed" || type === "error") { + const resp = data.response as { error?: { message?: string } } | undefined; + error = resp?.error?.message + ?? (data.error as { message?: string } | undefined)?.message + ?? (typeof data.message === "string" ? data.message : "upstream stream failed"); + } + }; + + try { + while (true) { + if (signal?.aborted) throw signal.reason ?? new DOMException("The operation was aborted", "AbortError"); + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + for (const line of lines) { + if (line.startsWith("data: ")) handle(line.slice(6).trim()); + else if (line.startsWith("data:")) handle(line.slice(5).trim()); + } + } + if (buffer.trim()) { + for (const line of buffer.split("\n")) { + if (line.startsWith("data: ")) handle(line.slice(6).trim()); + else if (line.startsWith("data:")) handle(line.slice(5).trim()); + } + } + } finally { + try { reader.releaseLock(); } catch { /* already released */ } + } + + const base = finalResponse ?? createdResponse; + if (!base) return { response: undefined, error }; + + const completedOutput = Array.isArray(base.output) ? base.output : []; + const completedHasContent = completedOutput.some(item => { + if (!item || typeof item !== "object") return false; + const rec = item as Record; + if (rec.type === "message" && Array.isArray(rec.content) && rec.content.length > 0) return true; + if (rec.type === "function_call" || rec.type === "custom_tool_call") return true; + if (rec.type === "reasoning") return true; + return false; + }); + + if (!completedHasContent) { + const rebuilt: unknown[] = []; + if (itemsByIndex.size > 0) { + const indices = [...itemsByIndex.keys()].sort((a, b) => a - b); + for (const idx of indices) { + let item = { ...itemsByIndex.get(idx)! }; + const id = typeof item.id === "string" ? item.id : ""; + const streamedText = id ? textByItemId.get(id) : undefined; + if (item.type === "message" && streamedText !== undefined) { + const content = Array.isArray(item.content) ? [...item.content] : []; + if (content.length === 0) { + content.push({ type: "output_text", text: streamedText, annotations: [] }); + } else { + content[0] = typeof content[0] === "object" && content[0] + ? { ...(content[0] as Record), type: "output_text", text: streamedText } + : { type: "output_text", text: streamedText, annotations: [] }; + } + item = { ...item, status: item.status ?? "completed", content }; + } + rebuilt.push(item); + } + } + if (rebuilt.length === 0) { + const text = (looseDoneText || looseDeltaText || [...textByItemId.values()].join("")).trim(); + if (text) { + rebuilt.push({ + type: "message", + id: `msg_${Math.random().toString(36).slice(2, 14)}`, + role: "assistant", + status: "completed", + content: [{ type: "output_text", text, annotations: [] }], + }); + } + } + if (rebuilt.length > 0) { + return { + response: { + ...base, + status: typeof base.status === "string" ? base.status : "completed", + output: rebuilt, + }, + error, + }; + } + } + + return { response: base, error }; +} + export function buildToolBridgeMaps(parsed: OcxParsedRequest): { toolNsMap: Map; freeformToolNames: Set; @@ -1160,10 +1307,11 @@ export async function handleResponses( if (upstreamContentType) logCtx.usageDebugContentType = upstreamContentType; } // The chatgpt backend may omit Content-Type on SSE responses. Fall back to - // treating a successful body as SSE when the caller requested streaming. + // treating a successful body as SSE when CT is missing (we force stream:true + // on the ChatGPT wire even for non-stream clients). const passthroughCt = headers.get("content-type")?.toLowerCase(); const isEventStream = passthroughCt?.includes("text/event-stream") - || (upstreamResponse.ok && !!upstreamResponse.body && !passthroughCt && parsed.stream); + || (upstreamResponse.ok && !!upstreamResponse.body && !passthroughCt); const terminalRecorder = codexForwardTerminalOutcomeRecorder( config, authCtx, @@ -1214,6 +1362,32 @@ export async function handleResponses( // native relay, never enters JS Sink.write); branch[1] is consumed in the // background for terminal-outcome/quota inspection only. if (upstreamResponse.ok && isEventStream && upstreamResponse.body) { + // ChatGPT Codex REST requires stream:true; when the client asked for JSON we forced + // stream on the wire (openai-responses adapter) and buffer the completed event here. + // Rebuild output from streamed item/text events when completed.output is empty. + if (!parsed.stream) { + try { + const buffered = await bufferResponsesSseToJson(upstreamResponse.body, options.abortSignal); + if (buffered.error) { + return formatErrorResponse(502, "upstream_error", buffered.error); + } + if (buffered.response) { + if (rememberPassthroughResponse) { + try { rememberPassthroughResponse(buffered.response as { id?: unknown; output?: unknown; status?: unknown }); } catch { /* best-effort */ } + } + terminalRecorder?.("completed"); + options.onNativePassthroughTerminal?.("completed"); + return new Response(JSON.stringify(buffered.response), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + } + return formatErrorResponse(502, "upstream_error", "upstream stream ended without response.completed"); + } catch (err) { + if (options.abortSignal?.aborted) return clientCancelledResponse(); + return formatErrorResponse(502, "upstream_error", err instanceof Error ? err.message : String(err)); + } + } const [nativeBody, inspectBody] = upstreamResponse.body.tee(); const turnAc = new AbortController(); linkAbortSignal(upstream, turnAc.signal); @@ -1436,7 +1610,8 @@ export async function handleResponses( stream: parsed.stream, }); } else { - upstreamResponse = await fetchWithResetRetry( + // Retry pre-stream 5xx (OpenCode free tier / gateway blips) the same way passthrough does. + upstreamResponse = await fetchWithTransientRetry( recovery => { noteAttemptSend(logCtx.activeAttempt, inputTokenEstimate, recovery); return fetchWithHeaderTimeout(request.url, applyUpstreamRecoveryInit({