Skip to content

[WRONG BRANCH] fix(responses): ChatGPT non-stream buffer + free-tier adapter hardening #293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
GoldenLoaf24h wants to merge 1 commit into from
Closed

[WRONG BRANCH] fix(responses): ChatGPT non-stream buffer + free-tier adapter hardening #293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 51 additions & 12 deletions src/adapters/mimo-free.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, unknown>;
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 };
Comment on lines +183 to +188

@coderabbitai coderabbitai Bot Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win

Case-sensitive comparison can silently escalate effort.

allowed.has(effort) and the "minimal"/"none" check are case-sensitive. A value like "Low" or "Medium" (differently-cased than the exact lowercase literals) fails the allowed check, isn't "minimal"/"none", and falls through to mapped = "high" — silently escalating a lightweight request to the most expensive tier instead of clamping it down. Given the surrounding comment already anticipates non-standard client values (max/xhigh/ultra), casing variance is plausible too.

🔧 Proposed fix 
-        const effort = rec.reasoning_effort;
-        if (typeof effort !== "string") return body;
+        const effort = typeof rec.reasoning_effort === "string" ? rec.reasoning_effort.toLowerCase() : undefined;
+        if (effort === undefined) return body;
         const allowed = new Set(["low", "medium", "high"]);
-        if (allowed.has(effort)) return body;
+        if (allowed.has(effort)) return { ...rec, reasoning_effort: effort };
         const mapped = effort === "minimal" || effort === "none" ? "low" : "high";
         return { ...rec, reasoning_effort: mapped };
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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 effort = typeof rec.reasoning_effort === "string" ? rec.reasoning_effort.toLowerCase() : undefined;
if (effort === undefined) return body;
const allowed = new Set(["low", "medium", "high"]);
if (allowed.has(effort)) return { ...rec, reasoning_effort: effort };
const mapped = effort === "minimal" || effort === "none" ? "low" : "high";
return { ...rec, reasoning_effort: mapped };
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/adapters/mimo-free.ts` around lines 183 - 188, Update the
reasoning_effort normalization logic in the shown adapter flow to compare values
case-insensitively before checking allowed values or mapping “minimal”/“none”.
Preserve canonical lowercase outputs, and ensure differently-cased lightweight
values such as “Low”, “Medium”, or “Minimal” are not escalated to “high”.

};
const markedBody = clampReasoningEffort(injectMimoSystemMarker(baseBody));

const headers: Record<string, string> = {
"Content-Type": "application/json",
Expand All @@ -195,31 +207,58 @@ export function createMimoFreeAdapter(provider: OcxProviderConfig): ProviderAdap
},

async fetchResponse(request: AdapterRequest, ctx): Promise<Response> {
const response = await fetch(request.url, {
const doFetch = (headers: Record<string, string>) => fetch(request.url, {
method: request.method,
headers: request.headers as Record<string, string>,
headers,
body: request.body,
signal: ctx?.abortSignal,
});

let headers = request.headers as Record<string, string>;
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<string, string>),
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);
}
}
Comment on lines +241 to 262

@coderabbitai coderabbitai Bot Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf 'Repo root: '; pwd
echo '--- relevant files ---'
git ls-files 'src/adapters/mimo-free.ts' 'tests/mimo-free-provider.test.ts'

echo '--- outline src/adapters/mimo-free.ts ---'
ast-grep outline src/adapters/mimo-free.ts --view expanded || true

echo '--- outline tests/mimo-free-provider.test.ts ---'
ast-grep outline tests/mimo-free-provider.test.ts --view expanded || true

echo '--- targeted snippets ---'
sed -n '210,280p' src/adapters/mimo-free.ts
echo '--- tests around 403 case ---'
sed -n '240,320p' tests/mimo-free-provider.test.ts

Repository: lidge-jun/opencodex

Length of output: 7686

Tighten the abuse-detection branch. src/adapters/mimo-free.ts:241-262 treats any 400/403/441 body containing Illegal access or the bare substring rate as blocked. That will false-positive on ordinary messages like generate or moderate, and it also changes the existing 403 behavior: tests/mimo-free-provider.test.ts still expects 403 ("Illegal access") to be returned without retry. Narrow the match to explicit rate-limit markers and keep 403 out of this retry path unless the test and adjacent comment are updated together.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/adapters/mimo-free.ts` around lines 241 - 262, The abuse-detection logic
in the response handling branch is too broad and incorrectly retries ordinary
403 responses. Update the blocked check near doFetch to match only explicit
rate-limit markers, remove the broad bare “rate” and “Illegal access” matching,
and exclude 403 from this retry path so existing 403 behavior remains unchanged.


return response;
Expand Down
9 changes: 9 additions & 0 deletions src/adapters/openai-responses.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, unknown> = { ...outBody };
if (next.stream !== true) next.stream = true;
if (next.store !== false) next.store = false;
outBody = next;
}
return {
url,
method: "POST",
Expand Down
183 changes: 179 additions & 4 deletions src/server/responses.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<Uint8Array>,
signal?: AbortSignal,
): Promise<{ response?: Record<string, unknown>; error?: string }> {
const reader = body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let finalResponse: Record<string, unknown> | undefined;
let createdResponse: Record<string, unknown> | undefined;
let error: string | undefined;
const itemsByIndex = new Map<number, Record<string, unknown>>();
const textByItemId = new Map<string, string>();
let looseDeltaText = "";
let looseDoneText = "";

const handle = (payload: string): void => {
if (!payload || payload === "[DONE]") return;
let data: Record<string, unknown>;
try {
data = JSON.parse(payload) as Record<string, unknown>;
} 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<string, unknown>;
}
} 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<string, unknown>);
}
} 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<string, unknown>;
}
} 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<string, unknown>;
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;
});
Comment on lines +183 to +191

@coderabbitai coderabbitai Bot Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Inspect the target file and nearby helpers.
file="src/server/responses.ts"
wc -l "$file"
sed -n '1,280p' "$file"

# Find related response buffering / SSE handling code and any tests.
rg -n "completedHasContent|response\.completed|output_text|output_item|reasoning|bufferResponsesSseToJson|stream:false|stream false" src test .

Repository: lidge-jun/opencodex

Length of output: 50376

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Read the exact buffering logic.
sed -n '150,250p' src/server/responses.ts

# Read the outbound SSE handling around completion and item assembly.
sed -n '220,340p' src/claude/outbound.ts
sed -n '380,460p' src/claude/outbound.ts

# Show any focused references to the buffering fallback and reasoning-only handling.
rg -n "completedHasContent|response\.completed|reasoning|output_item\.done|output_text\.delta|stream:false|buffers SSE" src/server/responses.ts src/claude/outbound.ts src/adapters/openai-responses.ts src/claude/inbound.ts

Repository: lidge-jun/opencodex

Length of output: 18922

🏁 Script executed:

#!/bin/bash
set -euo pipefail

sed -n '1,170p' src/server/responses.ts
sed -n '170,220p' src/server/responses.ts

# Focus on any tests or fixtures for the stream:false fallback.
rg -n "stream:false|stream false|output: \[\]|reasoning-only|response.completed.*reasoning|completedHasContent" test src --glob '!src/codex/data/**'

Repository: lidge-jun/opencodex

Length of output: 10745

Treat reasoning-only completed outputs as incomplete here.

At src/server/responses.ts:184-191, reasoning currently counts as content, so a response.completed that contains only a reasoning item skips the reconstruction block even when the assistant text arrived in response.output_text.* / response.output_item.*. That returns a JSON body with reasoning but no answer for stream:false callers. Require a real message or tool item to short-circuit, and let reasoning-only completions fall through to the rebuild path.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/server/responses.ts` around lines 183 - 191, The completed-output check
in the completed response handling must not treat a reasoning-only item as
content. Update completedHasContent so only non-empty message items and
function_call/custom_tool_call items short-circuit reconstruction; allow
reasoning-only outputs to fall through the existing rebuild path using
response.output_text/output_item data.


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<string, unknown>), 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<string, { namespace: string; name: string }>;
freeformToolNames: Set<string>;
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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));
}
Comment on lines +1386 to +1389

@coderabbitai coderabbitai Bot Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Buffered non-stream path leaks the upstream connection on read error.

Unlike the fetch catch at Lines 1288‑1301, this catch returns without calling upstream.abort(). bufferResponsesSseToJson only releases the reader lock in its finally (Line 177) — it never cancels the stream. So when reader.read() throws mid-body (mid-stream reset, decode failure, abort), the upstream socket is left dangling until runtime cleanup. Abort the controller before returning:

🔧 Proposed fix 
         } catch (err) {
+          upstream.abort();
           if (options.abortSignal?.aborted) return clientCancelledResponse();
           return formatErrorResponse(502, "upstream_error", err instanceof Error ? err.message : String(err));
         }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
} catch (err) {
if (options.abortSignal?.aborted) return clientCancelledResponse();
return formatErrorResponse(502, "upstream_error", err instanceof Error ? err.message : String(err));
}
} catch (err) {
upstream.abort();
if (options.abortSignal?.aborted) return clientCancelledResponse();
return formatErrorResponse(502, "upstream_error", err instanceof Error ? err.message : String(err));
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/server/responses.ts` around lines 1386 - 1389, Update the catch block
surrounding bufferResponsesSseToJson in the non-stream response path to call
upstream.abort() before returning the formatted 502 error, while preserving the
existing client-cancellation response behavior. Ensure the abort also occurs
when reader.read() fails mid-body.

}
const [nativeBody, inspectBody] = upstreamResponse.body.tee();
const turnAc = new AbortController();
linkAbortSignal(upstream, turnAc.signal);
Expand Down Expand Up @@ -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({
Expand Down
Loading