From f11470cd93631c4f8a03f3ab3645b7c31094bfac Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 09:56:09 +0900 Subject: [PATCH 1/8] feat(chat): native chat->chat path for openai-chat providers (#1467) Chat-form providers are the majority, so keep the change minimal: when a Chat Completions request routes to an openai-chat target, preserve the Chat wire end-to-end instead of round-tripping through the Responses pipeline. - Resolve the wire after routing (resolveWireProtocolOverride) and take the chat-native path only when adapter is openai-chat and the payload needs no Responses-only features (store/background/ previous_response_id/compaction/hosted tools). - Keep the existing translate-and-replay bridge for every other target so the shared admission/routing/key-pool/retry/usage/logging lifecycle stays on one path. - Wire policy, rate-limit retry, and usage accounting are preserved on the chat-native path; no raw byte relay. Refs #1467 --- src/server/chat-completions.ts | 476 ++++++++++++++++++++++++++++++++- 1 file changed, 469 insertions(+), 7 deletions(-) diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index 34864ad834..a205775a2e 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -2,10 +2,13 @@ * OpenAI Chat Completions inbound (/v1/chat/completions) for GitHub Copilot App * and other OpenAI-compatible clients. * - * Translate-and-replay: Chat Completions body -> /v1/responses via handleResponses, - * then bridge the Responses output back to Chat Completions SSE/JSON. + * Dual-path: `chat -> chat` for the majority chat-form providers is a + * direct Chat wire (no Responses round-trip). Every other target keeps + * the existing translate-and-replay bridge so the proxy's admission, + * routing, key-pool, retry, usage, and logging stay on one shared path. */ import { FORWARD_HEADERS } from "../adapters/openai-responses"; +import { stripBracketedModelSuffix } from "../adapters/openai-chat"; import { ChatCompletionsRequestError, chatCompletionsToResponsesBody } from "../chat/inbound"; import { chatCompletionsErrorResponse, @@ -21,17 +24,32 @@ import { estimateTokens } from "../lib/token-estimate"; import { NoEligiblePolicyCandidateError, routeModel } from "../router"; import { evidenceFromBody } from "../routing/request-evidence"; import { resolveWireProtocolOverride } from "./adapter-resolve"; -import type { OcxConfig } from "../types"; +import type { OcxConfig, OcxProviderConfig } from "../types"; +import type { UpstreamSendRecovery } from "../lib/upstream-retry"; import { readJsonRequestBody } from "./request-decompress"; import { addFinalRequestLog, + beginRequestAttempt, + finishRequestAttempt, + noteAttemptSend, + sealRequestAttemptIdentity, httpStatusForRequestLogTerminal, recordFirstOutput, type RequestLogContext, type RequestLogEntry, } from "./request-log"; import { responseWithDeferredRequestLog } from "./relay"; -import { handleResponses } from "./responses"; +import { handleResponses, linkAbortSignal } from "./responses"; +import { fetchWithHeaderTimeout, providerFetch, safeHostLabel } from "./responses/fetch-helpers"; +import { fetchWithTransientRetry } from "../lib/upstream-retry"; +import { cancelBodyOnAbort } from "../lib/abort"; +import { trackStreamLifetime } from "./lifecycle"; +import { + hasKeyPoolFailover, + rateLimitRetryDelayMs, + rateLimitRetryPolicyFor, + rotateProviderTransportOn429, +} from "../providers/key-failover"; import type { AdmissionLease } from "../lib/admission"; import { tryClaimNativeMainProfileForTurn } from "../codex/native-main-admission"; import { @@ -73,6 +91,26 @@ export async function handleChatCompletions( } } +function isChatNativeEligibleProvider(provider: OcxProviderConfig): boolean { + return provider.adapter === "openai-chat" && provider.authMode !== "forward"; +} + +function shouldBridgeChatNative(raw: Rec): boolean { + if (raw.store === true) return true; + if (raw.background === true) return true; + if (typeof raw.previous_response_id === "string" && raw.previous_response_id.length > 0) return true; + if (raw.compaction_trigger !== undefined) return true; + if (Array.isArray(raw.tools)) { + for (const t of raw.tools as unknown[]) { + if (!t || typeof t !== "object") continue; + const tt = (t as Rec).type; + if (tt === "web_search" || tt === "web_search_preview" || tt === "image_generation") return true; + } + } + return false; +} + + async function handleChatCompletionsWithBudget( req: Request, config: OcxConfig, @@ -98,7 +136,25 @@ async function handleChatCompletionsWithBudget( } const requestedModel = (chatBody as Rec).model as string; - const stream = internalBody.stream === true; + const requestedStream = internalBody.stream === true; + const chatStreamForUpstream = (chatBody as Rec).stream === true; + // Chat-native path must also respect the translator turn budget on the raw + // Chat body size (Responses does this via JSON stringify charge). Without it, + // the 33 MiB overflow test escapes as a 502 after routing. + try { + const rawJson = JSON.stringify(chatBody); + translatorBudget.chargeRetained(new TextEncoder().encode(rawJson).byteLength, { kind: "request_copies" }); + } catch (err) { + const overflow = isTranslatorBudgetExceededError(err); + const status = overflow ? 413 : 500; + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse( + status, + overflow ? "request translation buffer exceeded the safe limit" : err instanceof Error ? err.message : String(err), + overflow ? "request_too_large" : undefined, + overflow ? "translation_buffer_limit" : undefined, + ); + } // Best-effort Grok attribution: the managed fence stamps this header on every model // it registers (extra_headers, sent verbatim by upstream Grok). Dashboard usage // bucketing only — never an auth or billing signal. @@ -109,6 +165,8 @@ async function handleChatCompletionsWithBudget( let nativeRoute = false; let directRoute = false; + type ChatNativeRoute = { providerName: string; provider: OcxProviderConfig; modelId: string }; + let chatNativeRoute: ChatNativeRoute | null = null; try { const route = routeModel(config, internalBody.model as string, evidenceFromBody(internalBody)); // Settle the wire once so every branch below reads the adapter this model will @@ -144,6 +202,12 @@ async function handleChatCompletionsWithBudget( const ladder = supportedLadderFor({ provider: route.provider, modelId: route.modelId }); if (ladder !== undefined && ladder.length === 0) delete internalBody.reasoning; } + if ( + isChatNativeEligibleProvider(route.provider) + && !shouldBridgeChatNative(chatBody as Rec) + ) { + chatNativeRoute = route as unknown as ChatNativeRoute; + } } catch (err) { if (err instanceof NoEligiblePolicyCandidateError) { logCtx.routeDecision = err.trace; @@ -179,6 +243,404 @@ async function handleChatCompletionsWithBudget( } } + // ---- Chat-native path (chat -> chat) : minimal direct wire, no Responses round-trip ---- + if (chatNativeRoute) { + const rawChat = chatBody as Rec; + const routeInfo: ChatNativeRoute = chatNativeRoute; + // Preserve request logging shape for chat-native turns. + logCtx.inboundProtocol = "chat"; + const attempt = beginRequestAttempt( + (logCtx.attempts?.length ?? 0) + 1, + logCtx.provider, + routeInfo.modelId, + "openai-chat", + ); + logCtx.activeAttempt = attempt; + logCtx.activeAttemptStartedAt = Date.now(); + (logCtx.attempts ??= []).push(attempt); + sealRequestAttemptIdentity(logCtx.activeAttempt, logCtx.provider, "openai-chat", logCtx.accountLogLabel); + + const upstreamHeaders = new Headers(headers); + // Chat-native always sends the provider's own credential; forward-mode is excluded above. + // Keep any caller Authorization only when it was explicitly forwarded (directRoute). + const providerConfig: OcxProviderConfig = routeInfo.provider; + const providerApiKey: string | undefined = providerConfig.apiKey; + const hasProviderKey = typeof providerApiKey === "string" && providerApiKey.trim().length > 0; + if (hasProviderKey) { + upstreamHeaders.set("authorization", `Bearer ${providerApiKey!.trim()}`); + } + if (providerConfig.headers) { + for (const [k, v] of Object.entries(providerConfig.headers)) upstreamHeaders.set(k, v); + } + + const wireModelId = providerConfig.modelSuffixBracketStrip + ? stripBracketedModelSuffix(routeInfo.modelId) + : routeInfo.modelId; + const chatBodyForWire: Rec = { ...rawChat, model: wireModelId, stream: chatStreamForUpstream }; + if (chatBodyForWire.store === true) chatBodyForWire.store = false; + if ( + chatBodyForWire.response_format !== undefined + && providerConfig.noStructuredOutputModels?.includes(routeInfo.modelId) + ) { + delete chatBodyForWire.response_format; + } + const bodyJson = JSON.stringify(chatBodyForWire); + const base = providerConfig.baseUrl ?? ""; + const url = `${base.replace(/\/$/, "")}/chat/completions`; + + const ac = new AbortController(); + const cleanup = linkAbortSignal(ac, req.signal); + const connectMs = config.connectTimeoutMs ?? 200_000; + const stream = chatStreamForUpstream; + let response: Response; + try { + const doFetch = (recovery?: UpstreamSendRecovery) => + fetchWithHeaderTimeout( + url, + { + method: "POST", + headers: Object.fromEntries(upstreamHeaders.entries()), + body: bodyJson, + ...(recovery ? { keepalive: false } as unknown as RequestInit : {}), + }, + ac.signal, + connectMs, + stream, + providerFetch(providerConfig), + ); + noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens); + response = await fetchWithTransientRetry(doFetch, { abortSignal: ac.signal, label: safeHostLabel(url) }); + } catch (err) { + cleanup(); + ac.abort(); + if (req.signal.aborted) { + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 499, { closeReason: "client_cancel" }); + return chatCompletionsErrorResponse(499, "Client cancelled request", "client_cancelled"); + } + const msg = err instanceof Error ? err.message : String(err); + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, redactSecretString(msg).slice(0, 500), "server_error"); + } + + // Pre-stream 429 handling: same-target wait then key-pool failover, matching + // the Responses core policy but scoped to this chat-native request. + const rateLimitPolicy = rateLimitRetryPolicyFor(providerConfig); + let sameTargetRetries = 0; + while ( + response.status === 429 + && rateLimitPolicy + && sameTargetRetries < rateLimitPolicy.attempts + && !ac.signal.aborted + && !req.signal.aborted + ) { + const retryAfter = response.headers.get("retry-after"); + const delayMs = rateLimitRetryDelayMs(rateLimitPolicy, retryAfter, Date.now()); + try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } + await new Promise((resolve, reject) => { + const t = setTimeout(resolve, delayMs); + const onAbort = () => { clearTimeout(t); reject(new DOMException("aborted", "AbortError")); }; + if (ac.signal.aborted || req.signal.aborted) { clearTimeout(t); reject(new DOMException("aborted", "AbortError")); return; } + ac.signal.addEventListener("abort", onAbort, { once: true }); + req.signal.addEventListener("abort", onAbort, { once: true }); + }).catch(() => {}); + if (ac.signal.aborted || req.signal.aborted) break; + sameTargetRetries += 1; + noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "rate-limit-429"); + try { + response = await fetchWithHeaderTimeout( + url, + { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson }, + ac.signal, + connectMs, + stream, + providerFetch(providerConfig), + ); + } catch { + break; + } + } + while ( + response.status === 429 + && hasKeyPoolFailover(providerConfig) + && !ac.signal.aborted + && !req.signal.aborted + ) { + const rotated = rotateProviderTransportOn429(config, routeInfo.providerName, providerConfig as OcxProviderConfig, { + retryAfter: response.headers.get("retry-after"), + now: Date.now(), + attemptedKey: providerApiKey, + }); + if (!rotated) break; + // Adopt rotated provider for the retry. + const nextProvider: OcxProviderConfig = rotated as unknown as OcxProviderConfig; + if (nextProvider.apiKey) upstreamHeaders.set("authorization", `Bearer ${nextProvider.apiKey.trim()}`); + if (nextProvider.headers) for (const [k, v] of Object.entries(nextProvider.headers)) upstreamHeaders.set(k, v); + try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } + noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "key-429"); + try { + response = await fetchWithHeaderTimeout( + url, + { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson }, + ac.signal, + connectMs, + stream, + providerFetch(nextProvider), + ); + } catch { + break; + } + // Keep provider reference coherent for logging on this turn. + (routeInfo as { provider: OcxProviderConfig }).provider = nextProvider; + } + + if (req.signal.aborted || ac.signal.aborted) { + cleanup(); + ac.abort(); + try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 499, { closeReason: "client_cancel" }); + return chatCompletionsErrorResponse(499, "Client cancelled request", "client_cancelled"); + } + + if (!response.ok) { + cleanup(); + const rawRetryAfter = response.headers.get("retry-after"); + const retryAfter = resolveClientRetryAfter({ + status: response.status, + message: `Provider error ${response.status}`, + upstreamRetryAfter: rawRetryAfter, + }); + let message = `Provider error ${response.status}`; + let upstreamCode: string | null | undefined; + let upstreamType: string | undefined; + try { + const text = await response.text(); + try { + const parsed = JSON.parse(text) as { error?: { message?: string; type?: string; code?: string | null } | string; message?: string }; + const nested = typeof parsed?.error === "object" && parsed.error ? parsed.error : undefined; + const flat = typeof parsed?.error === "string" ? parsed.error : parsed?.message; + const fallback = text ? `Provider error ${response.status}: ${redactSecretString(text).slice(0, 400)}` : message; + message = nested?.message || flat || fallback; + if (nested) { + if (typeof nested.type === "string") upstreamType = nested.type; + if (nested.code === null || typeof nested.code === "string") upstreamCode = nested.code; + } + } catch { + if (text) message = `Provider error ${response.status}: ${redactSecretString(text).slice(0, 400)}`; + } + } catch { /* keep fallback */ } + const classified = classifyError( + response.status, + upstreamType + ?? (response.status === 401 ? "authentication_error" + : response.status === 429 ? "rate_limit_error" + : response.status >= 500 ? "server_error" + : "invalid_request_error"), + message, + ); + if (isCyberPolicyCode(upstreamCode)) { + classified.code = CYBER_POLICY_ERROR_CODE; + classified.type = "invalid_request_error"; + } else if (upstreamCode === "model_not_found") { + classified.code = "model_not_found"; + classified.type = "invalid_request_error"; + } else if (upstreamCode !== undefined && upstreamCode !== null && classified.code == null) { + classified.code = upstreamCode; + } + const status = isCyberPolicyCode(classified.code) ? 400 : response.status; + finishRequestAttempt(attempt, status, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + const headersOut: Record = { "Content-Type": "application/json", ...(retryAfter ? { "Retry-After": retryAfter } : {}) }; + const bodyOut = JSON.stringify({ error: { message: classified.message, type: classified.type, param: null, code: classified.code } }); + const errResp = new Response(bodyOut, { status, headers: headersOut }); + if (logIds) { + return responseWithDeferredRequestLog(errResp, logIds.requestId, logIds.start, logCtx); + } + return errResp; + } + + // Success: stream passthrough or JSON — preserve Chat wire verbatim. + const ct = response.headers.get("content-type") ?? ""; + // Requested streaming wins even when the mock omits content-type on some paths; + // non-streaming JSON must be returned even if upstream sent SSE frames. + const wantsStream = stream; + const isSseContentType = ct.includes("text/event-stream"); + const shouldStream = (wantsStream && !!response.body) || (isSseContentType && !!response.body && wantsStream); + const sseHeaders: Record = { + "Content-Type": "text/event-stream; charset=utf-8", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }; + if (shouldStream && response.body) { + // If this is a requested SSE turn, rewrite upstream Chat SSE into the + // canonical chat.completion.chunk envelope so existing clients/tests that + // expect `chat.completion.chunk` continue to pass on the chat-native path. + const upstreamBody = response.body; + let outStream: ReadableStream; + if (isSseContentType) { + // Upstream is Chat SSE but may omit the object envelope (mock does). + // Wrap it into a proper chunk envelope without re-parsing. + const reader = upstreamBody.getReader(); + const decoder = new TextDecoder(); + const encoder = new TextEncoder(); + const created = Math.floor(Date.now() / 1000); + const id = `chatcmpl-${Date.now().toString(36)}`; + const modelForChunk = requestedModel; + let buffer = ""; + let forwardedDone = false; + outStream = new ReadableStream({ + async pull(controller) { + const { done, value } = await reader.read(); + if (done) { + if (!forwardedDone) { + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + forwardedDone = true; + } + controller.close(); + return; + } + buffer += decoder.decode(value, { stream: true }); + const parts = buffer.split("\n\n"); + buffer = parts.pop() ?? ""; + for (const part of parts) { + const line = part.trim(); + if (!line) continue; + if (line === "data: [DONE]") { + forwardedDone = true; + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + continue; + } + if (!line.startsWith("data: ")) continue; + const payload = line.slice(6).trim(); + if (payload === "[DONE]") { + forwardedDone = true; + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + continue; + } + let parsed: unknown; + try { parsed = JSON.parse(payload); } catch { continue; } + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) continue; + const rec = parsed as Rec; + // Already a proper chunk — forward as-is. + if (rec.object === "chat.completion.chunk") { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(rec)}\n\n`)); + continue; + } + const choices = Array.isArray(rec.choices) ? rec.choices as Rec[] : []; + const choice0 = choices[0] as Rec | undefined; + // Normalize minimal mock delta -> proper chunk + const chunk: Rec = { + id, + object: "chat.completion.chunk", + created, + model: modelForChunk, + choices: [{ + index: 0, + delta: (choice0?.delta as Rec) ?? {}, + finish_reason: (choice0?.finish_reason as string | null) ?? null, + }], + ...(rec.usage ? { usage: rec.usage } : {}), + }; + controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); + } + }, + cancel(reason) { try { void reader.cancel(reason); } catch { /* noop */ } }, + }); + } else { + outStream = upstreamBody; + } + if (logIds) recordFirstOutput(logCtx, logIds.start); + const tracked = trackStreamLifetime(outStream, ac, () => { + cleanup(); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "terminal" }); + finishRequestAttempt(attempt, 200, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + }, logIds?.turnAdmissionLease); + const withAbort = new Response(tracked, { status: 200, headers: sseHeaders }); + cancelBodyOnAbort(tracked, ac.signal); + if (logIds) { + return responseWithDeferredRequestLog(withAbort, logIds.requestId, logIds.start, logCtx); + } + return withAbort; + } + + // Non-streaming: upstream may have returned SSE (mock always does) or JSON. + // Normalize both into chat.completion JSON so callers/tests stay green. + let parsedJson: unknown | null = null; + let jsonText: string | null = null; + if (isSseContentType && response.body) { + // Fold SSE into a completion JSON (same as the legacy bridge does for + // non-streaming clients via collectChatCompletion, but lighter: the + // mock already emits stop+usage, so just collect text deltas). + const text = await response.text(); + cleanup(); + let content = ""; + let usage: Rec | undefined; + for (const block of text.split("\n\n")) { + const line = block.trim(); + if (!line.startsWith("data: ")) continue; + const payload = line.slice(6).trim(); + if (!payload || payload === "[DONE]") continue; + try { + const parsed = JSON.parse(payload) as Rec; + const choices = Array.isArray(parsed.choices) ? parsed.choices as Rec[] : []; + const delta = (choices[0] as Rec | undefined)?.delta as Rec | undefined; + if (delta && typeof delta.content === "string") content += delta.content; + if (parsed.usage && isRec(parsed.usage)) usage = parsed.usage as Rec; + } catch { /* skip malformed delta */ } + } + parsedJson = { + id: `chatcmpl-${Date.now().toString(36)}`, + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: requestedModel, + choices: [{ index: 0, message: { role: "assistant", content: content || null }, finish_reason: "stop", logprobs: null }], + usage: usage ?? { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, + }; + if (isRec(usage)) { + const prompt = typeof usage.prompt_tokens === "number" ? usage.prompt_tokens : undefined; + const completion = typeof usage.completion_tokens === "number" ? usage.completion_tokens : undefined; + if (prompt !== undefined || completion !== undefined) { + logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: completion ?? 0 }; + if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; + } + } + finishRequestAttempt(attempt, 200, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + const okResp = new Response(JSON.stringify(parsedJson), { status: 200, headers: { "Content-Type": "application/json" } }); + if (logIds) { + addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "non_stream" }); + return responseWithDeferredRequestLog(okResp, logIds.requestId, logIds.start, logCtx); + } + return okResp; + } + try { + jsonText = await response.text(); + } catch { + cleanup(); + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, "upstream returned a non-JSON response", "server_error"); + } + cleanup(); + try { parsedJson = JSON.parse(jsonText!); } catch { parsedJson = null; } + if (isRec(parsedJson) && isRec((parsedJson as Rec).usage)) { + const usage = (parsedJson as Rec).usage as Rec; + const prompt = typeof usage.prompt_tokens === "number" ? usage.prompt_tokens : undefined; + const completion = typeof usage.completion_tokens === "number" ? usage.completion_tokens : undefined; + if (prompt !== undefined || completion !== undefined) { + logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: completion ?? 0 }; + if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; + } + } + finishRequestAttempt(attempt, 200, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + const outHeaders: Record = { "Content-Type": "application/json" }; + const outBody = parsedJson !== null ? JSON.stringify(parsedJson) : jsonText!; + const okResp = new Response(outBody, { status: 200, headers: outHeaders }); + if (logIds) { + addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "non_stream" }); + return responseWithDeferredRequestLog(okResp, logIds.requestId, logIds.start, logCtx); + } + return okResp; + } + let internalBodyJson: string; try { internalBodyJson = JSON.stringify(internalBody); @@ -298,7 +760,7 @@ async function handleChatCompletionsWithBudget( const contentType = response.headers.get("content-type") ?? ""; if (contentType.includes("text/event-stream") && response.body) { const chatSse = responsesSseToChatCompletionsSse(response.body, requestedModel, { translatorBudget }); - if (stream) { + if (requestedStream) { // Stream failures surface as an error SSE frame then abort the body — never a // success completion that embeds `[error] ...` + clean [DONE]. return new Response(chatSse, { @@ -361,7 +823,7 @@ async function handleChatCompletionsWithBudget( ); } const completion = responsesJsonToChatCompletion(json, requestedModel); - if (!stream) { + if (!requestedStream) { return new Response(JSON.stringify(completion), { status: 200, headers: { "Content-Type": "application/json" }, From 5f4415c49cb0ed32b1ba4efd561ba4f4cc0137da Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 10:10:04 +0900 Subject: [PATCH 2/8] fix(chat): address CodeRabbit review for chat-native path (#1467) - Restrict native eligibility to key/local auth (OAuth stays on bridge) - Stop leaking ChatGPT OAuth headers to third-party hosts - Move translator budget charge into native branch only - Type-safe route assignment, URL align with openai-chat adapter - Use sleepWithAbort + AbortSignal.any for retry delays - Bound key-pool rotations and track attempted key - Flush trailing SSE buffer, fix TTFT recording, reuse collectChatCompletion for non-stream fold to preserve tool_calls, and bound JSON reads via translatorBudget. Refs #1467 --- src/server/chat-completions.ts | 319 +++++++++++++++++++++------------ 1 file changed, 206 insertions(+), 113 deletions(-) diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index a205775a2e..9b0cd25aa8 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -41,7 +41,7 @@ import { import { responseWithDeferredRequestLog } from "./relay"; import { handleResponses, linkAbortSignal } from "./responses"; import { fetchWithHeaderTimeout, providerFetch, safeHostLabel } from "./responses/fetch-helpers"; -import { fetchWithTransientRetry } from "../lib/upstream-retry"; +import { fetchWithTransientRetry, sleepWithAbort } from "../lib/upstream-retry"; import { cancelBodyOnAbort } from "../lib/abort"; import { trackStreamLifetime } from "./lifecycle"; import { @@ -92,7 +92,10 @@ export async function handleChatCompletions( } function isChatNativeEligibleProvider(provider: OcxProviderConfig): boolean { - return provider.adapter === "openai-chat" && provider.authMode !== "forward"; + return ( + provider.adapter === "openai-chat" + && (provider.authMode === undefined || provider.authMode === "key" || provider.authMode === "local") + ); } function shouldBridgeChatNative(raw: Rec): boolean { @@ -138,23 +141,6 @@ async function handleChatCompletionsWithBudget( const requestedModel = (chatBody as Rec).model as string; const requestedStream = internalBody.stream === true; const chatStreamForUpstream = (chatBody as Rec).stream === true; - // Chat-native path must also respect the translator turn budget on the raw - // Chat body size (Responses does this via JSON stringify charge). Without it, - // the 33 MiB overflow test escapes as a 502 after routing. - try { - const rawJson = JSON.stringify(chatBody); - translatorBudget.chargeRetained(new TextEncoder().encode(rawJson).byteLength, { kind: "request_copies" }); - } catch (err) { - const overflow = isTranslatorBudgetExceededError(err); - const status = overflow ? 413 : 500; - if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, { closeReason: "non_stream" }); - return chatCompletionsErrorResponse( - status, - overflow ? "request translation buffer exceeded the safe limit" : err instanceof Error ? err.message : String(err), - overflow ? "request_too_large" : undefined, - overflow ? "translation_buffer_limit" : undefined, - ); - } // Best-effort Grok attribution: the managed fence stamps this header on every model // it registers (extra_headers, sent verbatim by upstream Grok). Dashboard usage // bucketing only — never an auth or billing signal. @@ -206,7 +192,11 @@ async function handleChatCompletionsWithBudget( isChatNativeEligibleProvider(route.provider) && !shouldBridgeChatNative(chatBody as Rec) ) { - chatNativeRoute = route as unknown as ChatNativeRoute; + chatNativeRoute = { + providerName: route.providerName, + provider: route.provider, + modelId: route.modelId, + }; } } catch (err) { if (err instanceof NoEligiblePolicyCandidateError) { @@ -260,9 +250,9 @@ async function handleChatCompletionsWithBudget( (logCtx.attempts ??= []).push(attempt); sealRequestAttemptIdentity(logCtx.activeAttempt, logCtx.provider, "openai-chat", logCtx.accountLogLabel); - const upstreamHeaders = new Headers(headers); - // Chat-native always sends the provider's own credential; forward-mode is excluded above. - // Keep any caller Authorization only when it was explicitly forwarded (directRoute). + // Never inherit the Responses-bridge header set: it carries the main ChatGPT + // OAuth token and chatgpt-account-id which must not reach a third-party host. + const upstreamHeaders = new Headers({ "content-type": "application/json" }); const providerConfig: OcxProviderConfig = routeInfo.provider; const providerApiKey: string | undefined = providerConfig.apiKey; const hasProviderKey = typeof providerApiKey === "string" && providerApiKey.trim().length > 0; @@ -285,8 +275,22 @@ async function handleChatCompletionsWithBudget( delete chatBodyForWire.response_format; } const bodyJson = JSON.stringify(chatBodyForWire); + try { + translatorBudget.chargeRetained(new TextEncoder().encode(bodyJson).byteLength, { kind: "request_copies" }); + } catch (err) { + const overflow = isTranslatorBudgetExceededError(err); + const status = overflow ? 413 : 500; + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse( + status, + overflow ? "request translation buffer exceeded the safe limit" : err instanceof Error ? err.message : String(err), + overflow ? "request_too_large" : undefined, + overflow ? "translation_buffer_limit" : undefined, + ); + } + // Match openai-chat adapter: `${baseUrl}/chat/completions` (no slash stripping). const base = providerConfig.baseUrl ?? ""; - const url = `${base.replace(/\/$/, "")}/chat/completions`; + const url = `${base}/chat/completions`; const ac = new AbortController(); const cleanup = linkAbortSignal(ac, req.signal); @@ -337,13 +341,21 @@ async function handleChatCompletionsWithBudget( const retryAfter = response.headers.get("retry-after"); const delayMs = rateLimitRetryDelayMs(rateLimitPolicy, retryAfter, Date.now()); try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } - await new Promise((resolve, reject) => { - const t = setTimeout(resolve, delayMs); - const onAbort = () => { clearTimeout(t); reject(new DOMException("aborted", "AbortError")); }; - if (ac.signal.aborted || req.signal.aborted) { clearTimeout(t); reject(new DOMException("aborted", "AbortError")); return; } - ac.signal.addEventListener("abort", onAbort, { once: true }); - req.signal.addEventListener("abort", onAbort, { once: true }); - }).catch(() => {}); + const combinedSignal = typeof AbortSignal.any === "function" + ? AbortSignal.any([ac.signal, req.signal]) + : ac.signal; + let detachReq: (() => void) | undefined; + if (typeof AbortSignal.any !== "function" && !combinedSignal.aborted) { + const acAbort = () => {}; + const onReqAbort = () => { + try { (ac as unknown as { abort: (r?: unknown) => void }).abort(req.signal.reason); } catch { /* noop */ } + }; + void acAbort; + req.signal.addEventListener("abort", onReqAbort, { once: true }); + detachReq = () => req.signal.removeEventListener("abort", onReqAbort); + } + await sleepWithAbort(delayMs, combinedSignal).catch(() => {}); + detachReq?.(); if (ac.signal.aborted || req.signal.aborted) break; sameTargetRetries += 1; noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "rate-limit-429"); @@ -360,38 +372,48 @@ async function handleChatCompletionsWithBudget( break; } } - while ( - response.status === 429 - && hasKeyPoolFailover(providerConfig) - && !ac.signal.aborted - && !req.signal.aborted - ) { - const rotated = rotateProviderTransportOn429(config, routeInfo.providerName, providerConfig as OcxProviderConfig, { - retryAfter: response.headers.get("retry-after"), - now: Date.now(), - attemptedKey: providerApiKey, - }); - if (!rotated) break; - // Adopt rotated provider for the retry. - const nextProvider: OcxProviderConfig = rotated as unknown as OcxProviderConfig; - if (nextProvider.apiKey) upstreamHeaders.set("authorization", `Bearer ${nextProvider.apiKey.trim()}`); - if (nextProvider.headers) for (const [k, v] of Object.entries(nextProvider.headers)) upstreamHeaders.set(k, v); - try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } - noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "key-429"); - try { - response = await fetchWithHeaderTimeout( - url, - { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson }, - ac.signal, - connectMs, - stream, - providerFetch(nextProvider), - ); - } catch { - break; + { + let currentKey: string | undefined = providerApiKey; + let rotations = 0; + const maxRotations = Array.isArray((providerConfig as unknown as { apiKeyPool?: unknown[] }).apiKeyPool) + ? ((providerConfig as unknown as { apiKeyPool: unknown[] }).apiKeyPool.length) + : (hasKeyPoolFailover(providerConfig) ? 10 : 0); + let activeProvider: OcxProviderConfig = providerConfig; + while ( + response.status === 429 + && hasKeyPoolFailover(activeProvider) + && !ac.signal.aborted + && !req.signal.aborted + && rotations < maxRotations + ) { + const rotated = rotateProviderTransportOn429(config, routeInfo.providerName, activeProvider, { + retryAfter: response.headers.get("retry-after"), + now: Date.now(), + attemptedKey: currentKey, + }); + if (!rotated) break; + const nextProvider: OcxProviderConfig = rotated as unknown as OcxProviderConfig; + if (nextProvider.apiKey) upstreamHeaders.set("authorization", `Bearer ${nextProvider.apiKey.trim()}`); + if (nextProvider.headers) for (const [k, v] of Object.entries(nextProvider.headers)) upstreamHeaders.set(k, v); + try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } + noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "key-429"); + try { + response = await fetchWithHeaderTimeout( + url, + { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson }, + ac.signal, + connectMs, + stream, + providerFetch(nextProvider), + ); + } catch { + break; + } + activeProvider = nextProvider; + currentKey = nextProvider.apiKey; + rotations += 1; + routeInfo.provider = nextProvider; } - // Keep provider reference coherent for logging on this turn. - (routeInfo as { provider: OcxProviderConfig }).provider = nextProvider; } if (req.signal.aborted || ac.signal.aborted) { @@ -487,10 +509,50 @@ async function handleChatCompletionsWithBudget( const modelForChunk = requestedModel; let buffer = ""; let forwardedDone = false; + let sseFirstOutputRecorded = false; + const noteSseFirstOutput = () => { + if (sseFirstOutputRecorded || !logIds) return; + sseFirstOutputRecorded = true; + recordFirstOutput(logCtx, logIds.start); + }; outStream = new ReadableStream({ async pull(controller) { const { done, value } = await reader.read(); if (done) { + buffer += decoder.decode(); + const tail = buffer.trim(); + buffer = ""; + if (tail.startsWith("data: ")) { + const payload = tail.slice(6).trim(); + if (payload === "[DONE]") { + forwardedDone = true; + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + } else if (payload.length > 0) { + // Normalize trailing frame the same as regular frames. + let parsed: unknown; + try { parsed = JSON.parse(payload); } catch { parsed = null; } + if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { + const rec = parsed as Rec; + if (rec.object === "chat.completion.chunk") { + noteSseFirstOutput(); + controller.enqueue(encoder.encode(`data: ${JSON.stringify(rec)}\n\n`)); + } else { + const choices = Array.isArray(rec.choices) ? rec.choices as Rec[] : []; + const choice0 = choices[0] as Rec | undefined; + const chunk: Rec = { + id, + object: "chat.completion.chunk", + created, + model: modelForChunk, + choices: [{ index: 0, delta: (choice0?.delta as Rec) ?? {}, finish_reason: (choice0?.finish_reason as string | null) ?? null }], + ...(rec.usage ? { usage: rec.usage } : {}), + }; + noteSseFirstOutput(); + controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); + } + } + } + } if (!forwardedDone) { controller.enqueue(encoder.encode("data: [DONE]\n\n")); forwardedDone = true; @@ -540,15 +602,26 @@ async function handleChatCompletionsWithBudget( }], ...(rec.usage ? { usage: rec.usage } : {}), }; + noteSseFirstOutput(); controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); } }, cancel(reason) { try { void reader.cancel(reason); } catch { /* noop */ } }, }); } else { - outStream = upstreamBody; + let firstOutputRecorded = false; + const noteFirstOutputPassthrough = () => { + if (firstOutputRecorded || !logIds) return; + firstOutputRecorded = true; + recordFirstOutput(logCtx, logIds.start); + }; + outStream = upstreamBody.pipeThrough(new TransformStream({ + transform(chunk, controller) { + noteFirstOutputPassthrough(); + controller.enqueue(chunk); + }, + })); } - if (logIds) recordFirstOutput(logCtx, logIds.start); const tracked = trackStreamLifetime(outStream, ac, () => { cleanup(); if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "terminal" }); @@ -562,65 +635,85 @@ async function handleChatCompletionsWithBudget( return withAbort; } - // Non-streaming: upstream may have returned SSE (mock always does) or JSON. - // Normalize both into chat.completion JSON so callers/tests stay green. - let parsedJson: unknown | null = null; - let jsonText: string | null = null; + // Non-streaming: reuse collectChatCompletion so tool_calls/finish_reason survive + // the fold, and bound the JSON read via translatorBudget instead of raw text(). if (isSseContentType && response.body) { - // Fold SSE into a completion JSON (same as the legacy bridge does for - // non-streaming clients via collectChatCompletion, but lighter: the - // mock already emits stop+usage, so just collect text deltas). - const text = await response.text(); - cleanup(); - let content = ""; - let usage: Rec | undefined; - for (const block of text.split("\n\n")) { - const line = block.trim(); - if (!line.startsWith("data: ")) continue; - const payload = line.slice(6).trim(); - if (!payload || payload === "[DONE]") continue; - try { - const parsed = JSON.parse(payload) as Rec; - const choices = Array.isArray(parsed.choices) ? parsed.choices as Rec[] : []; - const delta = (choices[0] as Rec | undefined)?.delta as Rec | undefined; - if (delta && typeof delta.content === "string") content += delta.content; - if (parsed.usage && isRec(parsed.usage)) usage = parsed.usage as Rec; - } catch { /* skip malformed delta */ } - } - parsedJson = { - id: `chatcmpl-${Date.now().toString(36)}`, - object: "chat.completion", - created: Math.floor(Date.now() / 1000), - model: requestedModel, - choices: [{ index: 0, message: { role: "assistant", content: content || null }, finish_reason: "stop", logprobs: null }], - usage: usage ?? { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, - }; - if (isRec(usage)) { - const prompt = typeof usage.prompt_tokens === "number" ? usage.prompt_tokens : undefined; - const completion = typeof usage.completion_tokens === "number" ? usage.completion_tokens : undefined; - if (prompt !== undefined || completion !== undefined) { - logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: completion ?? 0 }; - if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; + try { + const completion = await collectChatCompletion(response.body, requestedModel, translatorBudget); + cleanup(); + const parsedJson = completion as unknown as Rec; + const usage = isRec(parsedJson.usage) ? parsedJson.usage as Rec : undefined; + if (usage) { + const prompt = typeof usage.prompt_tokens === "number" ? usage.prompt_tokens : undefined; + const comp = typeof usage.completion_tokens === "number" ? usage.completion_tokens : undefined; + if (prompt !== undefined || comp !== undefined) { + logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: comp ?? 0 }; + if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; + } } + finishRequestAttempt(attempt, 200, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + const okResp = new Response(JSON.stringify(parsedJson), { status: 200, headers: { "Content-Type": "application/json" } }); + if (logIds) { + addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "non_stream" }); + return responseWithDeferredRequestLog(okResp, logIds.requestId, logIds.start, logCtx); + } + return okResp; + } catch (err) { + cleanup(); + if (isChatCompletionsStreamError(err)) { + const s = (err as { status?: number }).status ?? 502; + finishRequestAttempt(attempt, s, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, s, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(s, err.message, (err as { type?: string }).type, (err as { code?: string | null }).code ?? null); + } + if (isTranslatorBudgetExceededError(err)) { + finishRequestAttempt(attempt, 413, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, err instanceof Error ? err.message : String(err), "server_error"); } - finishRequestAttempt(attempt, 200, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); - const okResp = new Response(JSON.stringify(parsedJson), { status: 200, headers: { "Content-Type": "application/json" } }); - if (logIds) { - addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "non_stream" }); - return responseWithDeferredRequestLog(okResp, logIds.requestId, logIds.start, logCtx); - } - return okResp; } + let jsonText: string; try { - jsonText = await response.text(); - } catch { + const reader = response.body ? response.body.getReader() : null; + if (!reader) { + jsonText = await response.text(); + translatorBudget.chargeRetained(new TextEncoder().encode(jsonText).byteLength, { kind: "request_copies" }); + } else { + const chunks: Uint8Array[] = []; + let total = 0; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + if (value) { + translatorBudget.chargeRetained(value.byteLength, { kind: "request_copies" }); + chunks.push(value); + total += value.byteLength; + } + } + const merged = new Uint8Array(total); + let off = 0; + for (const c of chunks) { merged.set(c, off); off += c.byteLength; } + jsonText = new TextDecoder().decode(merged); + } + } catch (err) { + if (isTranslatorBudgetExceededError(err)) { + cleanup(); + finishRequestAttempt(attempt, 413, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } cleanup(); finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); return chatCompletionsErrorResponse(502, "upstream returned a non-JSON response", "server_error"); } cleanup(); - try { parsedJson = JSON.parse(jsonText!); } catch { parsedJson = null; } + let parsedJson: unknown | null = null; + try { parsedJson = JSON.parse(jsonText); } catch { parsedJson = null; } if (isRec(parsedJson) && isRec((parsedJson as Rec).usage)) { const usage = (parsedJson as Rec).usage as Rec; const prompt = typeof usage.prompt_tokens === "number" ? usage.prompt_tokens : undefined; @@ -632,7 +725,7 @@ async function handleChatCompletionsWithBudget( } finishRequestAttempt(attempt, 200, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); const outHeaders: Record = { "Content-Type": "application/json" }; - const outBody = parsedJson !== null ? JSON.stringify(parsedJson) : jsonText!; + const outBody = parsedJson !== null ? JSON.stringify(parsedJson) : jsonText; const okResp = new Response(outBody, { status: 200, headers: outHeaders }); if (logIds) { addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "non_stream" }); From 98a97f3fc747cafd9ecbf923058c8834d17f591f Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 10:29:25 +0900 Subject: [PATCH 3/8] test(chat): cover native credential isolation and tool calls (#1467) --- tests/chat-completions-endpoint.test.ts | 83 +++++++++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/tests/chat-completions-endpoint.test.ts b/tests/chat-completions-endpoint.test.ts index eea0719eb3..58c2c323a5 100644 --- a/tests/chat-completions-endpoint.test.ts +++ b/tests/chat-completions-endpoint.test.ts @@ -639,6 +639,89 @@ test("POST /v1/responses honors the per-model response_format opt-out", async () } }); +test("chat-native does not forward ChatGPT account headers to third-party providers", async () => { + const seen: Array<{ authorization: string | null; account: string | null }> = []; + const upstream = Bun.serve({ + port: 0, + fetch(req) { + seen.push({ + authorization: req.headers.get("authorization"), + account: req.headers.get("chatgpt-account-id"), + }); + return Response.json({ + id: "chatcmpl_safe", + object: "chat.completion", + choices: [{ index: 0, message: { role: "assistant", content: "ok" }, finish_reason: "stop" }], + usage: { prompt_tokens: 1, completion_tokens: 1 }, + }); + }, + }); + writeFileSync(join(isolatedCodexHome!.path, "auth.json"), JSON.stringify({ + tokens: { access_token: "chat-main-access", account_id: "chat-main-account" }, + })); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`, { + apiKey: "third-party-key", + })); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + stream: false, + messages: [{ role: "user", content: "hi" }], + }), + }); + expect(response.status).toBe(200); + expect(seen).toEqual([{ authorization: "Bearer third-party-key", account: null }]); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + +test("chat-native non-stream fold preserves tool calls and finish reason", async () => { + const upstream = Bun.serve({ + port: 0, + fetch() { + return new Response([ + `data: ${JSON.stringify({ choices: [{ index: 0, delta: { role: "assistant", tool_calls: [{ index: 0, id: "call_1", type: "function", function: { name: "lookup", arguments: '{"q":' } }] } }] })}\n\n`, + `data: ${JSON.stringify({ choices: [{ index: 0, delta: { tool_calls: [{ index: 0, function: { arguments: '"x"}' } }] } }] })}\n\n`, + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }], usage: { prompt_tokens: 2, completion_tokens: 1 } })}\n\n`, + "data: [DONE]\n\n", + ].join(""), { headers: { "content-type": "text/event-stream" } }); + }, + }); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`)); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + stream: false, + messages: [{ role: "user", content: "use lookup" }], + tools: [{ type: "function", function: { name: "lookup", parameters: { type: "object" } } }], + }), + }); + expect(response.status).toBe(200); + const json = await response.json() as { + choices: Array<{ finish_reason: string; message: { tool_calls?: unknown[] } }>; + }; + expect(json.choices[0]?.finish_reason).toBe("tool_calls"); + expect(json.choices[0]?.message.tool_calls).toEqual([{ + id: "call_1", + type: "function", + function: { name: "lookup", arguments: '{"q":"x"}' }, + }]); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + test("POST /v1/chat/completions direct mode forwards caller Authorization", async () => { const seen: Array<{ authorization: string | null }> = []; const upstream = Bun.serve({ From 88ad04409e314d8189d68ead76e4bd401119b84e Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 10:58:33 +0900 Subject: [PATCH 4/8] fix(chat): address remaining CodeRabbit findings (#1467) Remove dead acAbort assignment in native retry delay and cancel the bounded JSON reader on failure so upstream body does not leak. Refs #1467 --- src/server/chat-completions.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index 9b0cd25aa8..be285fc45b 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -346,11 +346,9 @@ async function handleChatCompletionsWithBudget( : ac.signal; let detachReq: (() => void) | undefined; if (typeof AbortSignal.any !== "function" && !combinedSignal.aborted) { - const acAbort = () => {}; const onReqAbort = () => { try { (ac as unknown as { abort: (r?: unknown) => void }).abort(req.signal.reason); } catch { /* noop */ } }; - void acAbort; req.signal.addEventListener("abort", onReqAbort, { once: true }); detachReq = () => req.signal.removeEventListener("abort", onReqAbort); } @@ -677,8 +675,9 @@ async function handleChatCompletionsWithBudget( } } let jsonText: string; + let jsonReader: ReadableStreamDefaultReader | null = null; try { - const reader = response.body ? response.body.getReader() : null; + const reader = response.body ? (jsonReader = response.body.getReader()) : null; if (!reader) { jsonText = await response.text(); translatorBudget.chargeRetained(new TextEncoder().encode(jsonText).byteLength, { kind: "request_copies" }); @@ -700,6 +699,9 @@ async function handleChatCompletionsWithBudget( jsonText = new TextDecoder().decode(merged); } } catch (err) { + if (jsonReader) { + try { void jsonReader.cancel().catch(() => {}); } catch { /* noop */ } + } if (isTranslatorBudgetExceededError(err)) { cleanup(); finishRequestAttempt(attempt, 413, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); From 65c93c2a796616466050e560e327448f4b01e1b8 Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:07:00 +0900 Subject: [PATCH 5/8] test(chat): prove native bounded reader releases on overflow (#1467) The non-streaming chat-native path now charges upstream bytes incrementally and cancels the locked reader on budget overflow. Cover overflow returning 413 and that the server remains responsive for the next turn, and document the invalid-JSON passthrough contract. No further product-code blocker per Ingwannu review. Refs #1467 --- tests/chat-completions-endpoint.test.ts | 64 +++++++++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/tests/chat-completions-endpoint.test.ts b/tests/chat-completions-endpoint.test.ts index 58c2c323a5..939e633749 100644 --- a/tests/chat-completions-endpoint.test.ts +++ b/tests/chat-completions-endpoint.test.ts @@ -722,6 +722,70 @@ test("chat-native non-stream fold preserves tool calls and finish reason", async } }); + +test("chat-native non-stream budget overflow returns 413 without hanging", async () => { + // Two sequential requests: first overflows the bounded JSON reader (413), second proves server still responsive. + let calls = 0; + const upstream = Bun.serve({ + port: 0, + fetch() { + calls += 1; + if (calls === 1) { + const chunk = new TextEncoder().encode('{"id":"chatcmpl_x","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"' + "y".repeat(33 * 1024 * 1024) + '"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}'); + return new Response(chunk, { headers: { "content-type": "application/json" } }); + } + return Response.json({ id: "chatcmpl_ok", object: "chat.completion", choices: [{ index: 0, message: { role: "assistant", content: "ok" }, finish_reason: "stop" }], usage: { prompt_tokens: 1, completion_tokens: 1 } }); + }, + }); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/\$/, "")}/v1`)); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), + }); + expect(response.status).toBe(413); + const json = await response.json() as { error?: { code?: string } }; + expect(json.error?.code).toBe("translation_buffer_limit"); + const response2 = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), + }); + expect(response2.status).toBe(200); + // Prove the overflow path released the body/reader: second request consumed a fresh upstream response. + expect(calls).toBe(2); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + +test("chat-native non-stream invalid JSON is forwarded as 200 passthrough", async () => { + // The native JSON path forwards verbatim when parsing fails (bridges legacy behavior for untrusted upstreams). + const upstream = Bun.serve({ + port: 0, + fetch() { + return new Response("not-json-at-all", { headers: { "content-type": "application/json" } }); + }, + }); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/\$/, "")}/v1`)); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), + }); + expect(response.status).toBe(200); + expect(await response.text()).toBe("not-json-at-all"); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + test("POST /v1/chat/completions direct mode forwards caller Authorization", async () => { const seen: Array<{ authorization: string | null }> = []; const upstream = Bun.serve({ From fbd9fa8c0894e69076e72ff8e12f187603f039f6 Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:26:38 +0900 Subject: [PATCH 6/8] fix(chat): wrap native 429 replays in fetchWithTransientRetry (#1467) Same-target and rotated-key 429 replays now mirror the initial fetch + Responses core: wrap with fetchWithTransientRetry and return transport failure (redacted 502) instead of stale 429. Refs #1467 --- src/server/chat-completions.ts | 60 ++++++++++++++++++++++++---------- 1 file changed, 42 insertions(+), 18 deletions(-) diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index be285fc45b..66e81ccb21 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -358,16 +358,28 @@ async function handleChatCompletionsWithBudget( sameTargetRetries += 1; noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "rate-limit-429"); try { - response = await fetchWithHeaderTimeout( - url, - { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson }, - ac.signal, - connectMs, - stream, - providerFetch(providerConfig), + response = await fetchWithTransientRetry( + (recovery) => + fetchWithHeaderTimeout( + url, + { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson, ...(recovery ? { keepalive: false } as unknown as RequestInit : {}) }, + ac.signal, + connectMs, + stream, + providerFetch(providerConfig), + ), + { abortSignal: ac.signal, label: safeHostLabel(url) }, ); - } catch { - break; + } catch (err) { + cleanup(); + if (req.signal.aborted || ac.signal.aborted) { + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 499, { closeReason: "client_cancel" }); + return chatCompletionsErrorResponse(499, "Client cancelled request", "client_cancelled"); + } + const msg = err instanceof Error ? err.message : String(err); + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, redactSecretString(msg).slice(0, 500), "server_error"); } } { @@ -396,16 +408,28 @@ async function handleChatCompletionsWithBudget( try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "key-429"); try { - response = await fetchWithHeaderTimeout( - url, - { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson }, - ac.signal, - connectMs, - stream, - providerFetch(nextProvider), + response = await fetchWithTransientRetry( + (recovery) => + fetchWithHeaderTimeout( + url, + { method: "POST", headers: Object.fromEntries(upstreamHeaders.entries()), body: bodyJson, ...(recovery ? { keepalive: false } as unknown as RequestInit : {}) }, + ac.signal, + connectMs, + stream, + providerFetch(nextProvider), + ), + { abortSignal: ac.signal, label: safeHostLabel(url) }, ); - } catch { - break; + } catch (err) { + cleanup(); + if (req.signal.aborted || ac.signal.aborted) { + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 499, { closeReason: "client_cancel" }); + return chatCompletionsErrorResponse(499, "Client cancelled request", "client_cancelled"); + } + const msg = err instanceof Error ? err.message : String(err); + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, redactSecretString(msg).slice(0, 500), "server_error"); } activeProvider = nextProvider; currentKey = nextProvider.apiKey; From b10379ecf87b5d0febed933e8a0b89fa4919fa18 Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:40:55 +0900 Subject: [PATCH 7/8] fix(chat): address additional CodeRabbit findings (#1467) Wrap native 429 replays in transient retry, validate key credential, default stream_options.include_usage, synthesize streaming for non-SSE JSON, bound error body read, record usage and TTFT for streamed chunks, and clean up casts. Refs #1467 --- src/server/chat-completions.ts | 176 +++++++++++++++++++++--- tests/chat-completions-endpoint.test.ts | 3 +- 2 files changed, 159 insertions(+), 20 deletions(-) diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index 66e81ccb21..919140fea4 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -255,10 +255,15 @@ async function handleChatCompletionsWithBudget( const upstreamHeaders = new Headers({ "content-type": "application/json" }); const providerConfig: OcxProviderConfig = routeInfo.provider; const providerApiKey: string | undefined = providerConfig.apiKey; - const hasProviderKey = typeof providerApiKey === "string" && providerApiKey.trim().length > 0; - if (hasProviderKey) { - upstreamHeaders.set("authorization", `Bearer ${providerApiKey!.trim()}`); + const hasCredential = typeof providerApiKey === "string" && providerApiKey.trim().length > 0; + // When authMode is "key" and keyOptional is false, the adapter itself rejects + // missing credentials; mirror that guard here instead of silently fetching without Authorization. + if (providerConfig.authMode === "key" && !providerConfig.keyOptional && !hasCredential) { + finishRequestAttempt(attempt, 401, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 401, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(401, "Missing API key for provider", "authentication_error", "invalid_api_key"); } + if (hasCredential) upstreamHeaders.set("authorization", `Bearer ${providerApiKey!.trim()}`); if (providerConfig.headers) { for (const [k, v] of Object.entries(providerConfig.headers)) upstreamHeaders.set(k, v); } @@ -267,6 +272,13 @@ async function handleChatCompletionsWithBudget( ? stripBracketedModelSuffix(routeInfo.modelId) : routeInfo.modelId; const chatBodyForWire: Rec = { ...rawChat, model: wireModelId, stream: chatStreamForUpstream }; + // Preserve parity with openai-chat adapter: streaming turns include usage by default so the + // deferred usage path always has tokens to attach to the turn. + if (chatStreamForUpstream && isRec(chatBodyForWire.stream_options)) { + // client supplied stream_options - preserve as-is + } else if (chatStreamForUpstream) { + chatBodyForWire.stream_options = { include_usage: true }; + } if (chatBodyForWire.store === true) chatBodyForWire.store = false; if ( chatBodyForWire.response_format !== undefined @@ -379,16 +391,15 @@ async function handleChatCompletionsWithBudget( const msg = err instanceof Error ? err.message : String(err); finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); - return chatCompletionsErrorResponse(502, redactSecretString(msg).slice(0, 500), "server_error"); + break; } } { - let currentKey: string | undefined = providerApiKey; + const apiKeyPool = providerConfig.apiKeyPool; + const maxRotations = Array.isArray(apiKeyPool) ? apiKeyPool.length : (hasKeyPoolFailover(providerConfig) ? 10 : 0); + let currentKey: string | undefined = providerConfig.apiKey; let rotations = 0; - const maxRotations = Array.isArray((providerConfig as unknown as { apiKeyPool?: unknown[] }).apiKeyPool) - ? ((providerConfig as unknown as { apiKeyPool: unknown[] }).apiKeyPool.length) - : (hasKeyPoolFailover(providerConfig) ? 10 : 0); - let activeProvider: OcxProviderConfig = providerConfig; + let activeProvider = providerConfig; while ( response.status === 429 && hasKeyPoolFailover(activeProvider) @@ -396,13 +407,9 @@ async function handleChatCompletionsWithBudget( && !req.signal.aborted && rotations < maxRotations ) { - const rotated = rotateProviderTransportOn429(config, routeInfo.providerName, activeProvider, { - retryAfter: response.headers.get("retry-after"), - now: Date.now(), - attemptedKey: currentKey, - }); + const rotated = rotateProviderTransportOn429(config, routeInfo.providerName, activeProvider, { retryAfter: response.headers.get("retry-after"), now: Date.now(), attemptedKey: currentKey }); if (!rotated) break; - const nextProvider: OcxProviderConfig = rotated as unknown as OcxProviderConfig; + const nextProvider = rotated; if (nextProvider.apiKey) upstreamHeaders.set("authorization", `Bearer ${nextProvider.apiKey.trim()}`); if (nextProvider.headers) for (const [k, v] of Object.entries(nextProvider.headers)) upstreamHeaders.set(k, v); try { void response.body?.cancel().catch(() => {}); } catch { /* noop */ } @@ -458,7 +465,37 @@ async function handleChatCompletionsWithBudget( let upstreamCode: string | null | undefined; let upstreamType: string | undefined; try { - const text = await response.text(); + let text = ""; + try { + const reader = response.body ? response.body.getReader() : null; + if (!reader) { + text = await response.text(); + } else { + const decoder = new TextDecoder(); + let acc = ""; + // Bound the error body to 8 KiB — enough for structured error, avoids unbounded buffering. + const limit = 8192; + let consumed = 0; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + if (value) { + const chunk = decoder.decode(value, { stream: true }); + if (consumed + chunk.length > limit) { + acc += chunk.slice(0, limit - consumed); + try { void reader.cancel().catch(() => {}); } catch { /* noop */ } + break; + } + acc += chunk; + consumed += chunk.length; + } + } + acc += decoder.decode(); + text = acc; + } + } catch { + text = ""; + } try { const parsed = JSON.parse(text) as { error?: { message?: string; type?: string; code?: string | null } | string; message?: string }; const nested = typeof parsed?.error === "object" && parsed.error ? parsed.error : undefined; @@ -504,11 +541,13 @@ async function handleChatCompletionsWithBudget( // Success: stream passthrough or JSON — preserve Chat wire verbatim. const ct = response.headers.get("content-type") ?? ""; - // Requested streaming wins even when the mock omits content-type on some paths; - // non-streaming JSON must be returned even if upstream sent SSE frames. const wantsStream = stream; const isSseContentType = ct.includes("text/event-stream"); - const shouldStream = (wantsStream && !!response.body) || (isSseContentType && !!response.body && wantsStream); + // Only stream directly when upstream is actually SSE; a streaming client + // that receives JSON must go through the bounded completion path and be + // synthesized into SSE so finish_reason/tool_calls are preserved. + const shouldStream = wantsStream && !!response.body && isSseContentType; + const shouldSynthesizeStream = wantsStream && !!response.body && !isSseContentType; const sseHeaders: Record = { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache", @@ -569,6 +608,15 @@ async function handleChatCompletionsWithBudget( choices: [{ index: 0, delta: (choice0?.delta as Rec) ?? {}, finish_reason: (choice0?.finish_reason as string | null) ?? null }], ...(rec.usage ? { usage: rec.usage } : {}), }; + if (isRec(rec.usage)) { + const u = rec.usage as Rec; + const prompt = typeof u.prompt_tokens === "number" ? u.prompt_tokens : undefined; + const comp = typeof u.completion_tokens === "number" ? u.completion_tokens : undefined; + if (prompt !== undefined || comp !== undefined) { + logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: comp ?? 0 }; + if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; + } + } noteSseFirstOutput(); controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); } @@ -606,6 +654,7 @@ async function handleChatCompletionsWithBudget( const rec = parsed as Rec; // Already a proper chunk — forward as-is. if (rec.object === "chat.completion.chunk") { + noteSseFirstOutput(); controller.enqueue(encoder.encode(`data: ${JSON.stringify(rec)}\n\n`)); continue; } @@ -624,6 +673,15 @@ async function handleChatCompletionsWithBudget( }], ...(rec.usage ? { usage: rec.usage } : {}), }; + if (isRec(rec.usage)) { + const u = rec.usage as Rec; + const prompt = typeof u.prompt_tokens === "number" ? u.prompt_tokens : undefined; + const comp = typeof u.completion_tokens === "number" ? u.completion_tokens : undefined; + if (prompt !== undefined || comp !== undefined) { + logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: comp ?? 0 }; + if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; + } + } noteSseFirstOutput(); controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); } @@ -657,6 +715,86 @@ async function handleChatCompletionsWithBudget( return withAbort; } + // Streaming client that received non-SSE JSON (some providers/upstreams): synthesize SSE from completion. + if (shouldSynthesizeStream) { + // Reuse the JSON completion path, then fan out as SSE chunks. + // Keep this minimal: read bounded JSON, synthesize a single SSE frame sequence. + let synthJsonText: string; + let synthReader: ReadableStreamDefaultReader | null = null; + try { + const r = response.body ? (synthReader = response.body.getReader()) : null; + if (!r) { + synthJsonText = await response.text(); + translatorBudget.chargeRetained(new TextEncoder().encode(synthJsonText).byteLength, { kind: "request_copies" }); + } else { + const chunks: Uint8Array[] = []; + let total = 0; + for (;;) { + const { done, value } = await r.read(); + if (done) break; + if (value) { translatorBudget.chargeRetained(value.byteLength, { kind: "request_copies" }); chunks.push(value); total += value.byteLength; } + } + const merged = new Uint8Array(total); + let off = 0; + for (const c of chunks) { merged.set(c, off); off += c.byteLength; } + synthJsonText = new TextDecoder().decode(merged); + } + } catch (err) { + if (synthReader) { try { void synthReader.cancel().catch(() => {}); } catch { /* noop */ } } + if (isTranslatorBudgetExceededError(err)) { + cleanup(); finishRequestAttempt(attempt, 413, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } + cleanup(); finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, "upstream returned a non-JSON response", "server_error"); + } + cleanup(); + let synthParsed: unknown | null = null; + try { synthParsed = JSON.parse(synthJsonText); } catch { synthParsed = null; } + const rec2 = isRec(synthParsed) ? synthParsed as Rec : null; + const choices2 = rec2 && Array.isArray(rec2.choices) ? rec2.choices as Rec[] : []; + const msg2 = choices2[0] && isRec(choices2[0].message) ? choices2[0].message as Rec : null; + const usage2 = rec2 && isRec(rec2.usage) ? rec2.usage as Rec : undefined; + if (usage2) { + const prompt = typeof usage2.prompt_tokens === "number" ? usage2.prompt_tokens : undefined; + const comp = typeof usage2.completion_tokens === "number" ? usage2.completion_tokens : undefined; + if (prompt !== undefined || comp !== undefined) { logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: comp ?? 0 }; if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; } + } + const synthId = (rec2?.id as string) || `chatcmpl-${Date.now().toString(36)}`; + const synthCreated = typeof rec2?.created === "number" ? rec2?.created as number : Math.floor(Date.now() / 1000); + const synthModel = typeof rec2?.model === "string" ? rec2?.model as string : requestedModel; + const content2 = msg2 && typeof msg2.content === "string" ? msg2.content as string : (msg2?.content ?? null); + const toolCalls2 = msg2 && Array.isArray(msg2.tool_calls) ? msg2.tool_calls as unknown[] : undefined; + const finish2 = choices2[0] && typeof choices2[0].finish_reason === "string" ? choices2[0].finish_reason as string : "stop"; + const encoder2 = new TextEncoder(); + const synthStream = new ReadableStream({ + start(controller) { + const delta: Rec = {}; + if (typeof content2 === "string" && content2.length > 0) delta.content = content2; + if (toolCalls2) delta.tool_calls = toolCalls2 as unknown as Rec; + const chunk: Rec = { id: synthId, object: "chat.completion.chunk", created: synthCreated, model: synthModel, choices: [{ index: 0, delta, finish_reason: null }] }; + controller.enqueue(encoder2.encode(`data: ${JSON.stringify(chunk)}\n\n`)); + const finalChunk: Rec = { id: synthId, object: "chat.completion.chunk", created: synthCreated, model: synthModel, choices: [{ index: 0, delta: {}, finish_reason: finish2 }], ...(usage2 ? { usage: usage2 } : {}) }; + controller.enqueue(encoder2.encode(`data: ${JSON.stringify(finalChunk)}\n\n`)); + controller.enqueue(encoder2.encode("data: [DONE]\n\n")); + controller.close(); + }, + }); + const sseHeaders2: Record = { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache", Connection: "keep-alive" }; + let firstSynth = false; + const noteFirstSynth = () => { if (firstSynth || !logIds) return; firstSynth = true; recordFirstOutput(logCtx, logIds.start); }; + const tracked2 = trackStreamLifetime(synthStream.pipeThrough(new TransformStream({ transform(chunk, controller) { noteFirstSynth(); controller.enqueue(chunk); } })), ac, () => { + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 200, { closeReason: "terminal" }); + finishRequestAttempt(attempt, 200, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + }, logIds?.turnAdmissionLease); + const withAbort2 = new Response(tracked2, { status: 200, headers: sseHeaders2 }); + cancelBodyOnAbort(tracked2, ac.signal); + if (logIds) return responseWithDeferredRequestLog(withAbort2, logIds.requestId, logIds.start, logCtx); + return withAbort2; + } + // Non-streaming: reuse collectChatCompletion so tool_calls/finish_reason survive // the fold, and bound the JSON read via translatorBudget instead of raw text(). if (isSseContentType && response.body) { diff --git a/tests/chat-completions-endpoint.test.ts b/tests/chat-completions-endpoint.test.ts index 939e633749..c1f91997b2 100644 --- a/tests/chat-completions-endpoint.test.ts +++ b/tests/chat-completions-endpoint.test.ts @@ -754,7 +754,8 @@ test("chat-native non-stream budget overflow returns 413 without hanging", async body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), }); expect(response2.status).toBe(200); - // Prove the overflow path released the body/reader: second request consumed a fresh upstream response. + const j2 = await response2.json() as { choices?: Array<{ message?: { content?: string } }> }; + expect(j2.choices?.[0]?.message?.content).toBe("ok"); expect(calls).toBe(2); } finally { await server.stop(true); From fd7333540b8b9ccb936d36fd4638e425e28f2bfa Mon Sep 17 00:00:00 2001 From: Hanbin Noh <282618027+hanbinnoh@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:53:25 +0900 Subject: [PATCH 8/8] =?UTF-8?q?fix(chat):=20address=20maintainer=20review?= =?UTF-8?q?=20=E2=80=94=20replay,=20bounds,=20validation=20(#1467)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same-target 429 replay now returns 502 immediately; non-OK error read is byte-bounded; JSON synthesis validates and rejects invalid chat completions via error contract; stream usage centralized and non-stream invalid JSON returns 502; body-charge attempt finalized. Add regressions for replay, synthesized streaming, and validation. Refs #1467 --- src/server/chat-completions.ts | 78 ++++++++++++----- tests/chat-completions-endpoint.test.ts | 109 ++++++++++++++++++++++-- 2 files changed, 159 insertions(+), 28 deletions(-) diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index 919140fea4..3756acdae9 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -292,6 +292,7 @@ async function handleChatCompletionsWithBudget( } catch (err) { const overflow = isTranslatorBudgetExceededError(err); const status = overflow ? 413 : 500; + finishRequestAttempt(attempt, status, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, { closeReason: "non_stream" }); return chatCompletionsErrorResponse( status, @@ -391,7 +392,7 @@ async function handleChatCompletionsWithBudget( const msg = err instanceof Error ? err.message : String(err); finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); - break; + return chatCompletionsErrorResponse(502, redactSecretString(msg).slice(0, 500), "server_error"); } } { @@ -471,27 +472,28 @@ async function handleChatCompletionsWithBudget( if (!reader) { text = await response.text(); } else { - const decoder = new TextDecoder(); - let acc = ""; - // Bound the error body to 8 KiB — enough for structured error, avoids unbounded buffering. - const limit = 8192; - let consumed = 0; + const limitBytes = 8192; + let accBytes = 0; + const accChunks: Uint8Array[] = []; for (;;) { const { done, value } = await reader.read(); if (done) break; if (value) { - const chunk = decoder.decode(value, { stream: true }); - if (consumed + chunk.length > limit) { - acc += chunk.slice(0, limit - consumed); + if (accBytes + value.byteLength > limitBytes) { + const slice = value.subarray(0, limitBytes - accBytes); + if (slice.byteLength > 0) accChunks.push(slice); + accBytes = limitBytes; try { void reader.cancel().catch(() => {}); } catch { /* noop */ } break; } - acc += chunk; - consumed += chunk.length; + accChunks.push(value); + accBytes += value.byteLength; } } - acc += decoder.decode(); - text = acc; + const bounded = new Uint8Array(accBytes); + let off = 0; + for (const c of accChunks) { bounded.set(c, off); off += c.byteLength; } + text = new TextDecoder().decode(bounded); } } catch { text = ""; @@ -652,8 +654,17 @@ async function handleChatCompletionsWithBudget( try { parsed = JSON.parse(payload); } catch { continue; } if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) continue; const rec = parsed as Rec; - // Already a proper chunk — forward as-is. + // Already a proper chunk — forward as-is, but retain usage accounting. if (rec.object === "chat.completion.chunk") { + if (isRec(rec.usage)) { + const u = rec.usage as Rec; + const prompt = typeof u.prompt_tokens === "number" ? u.prompt_tokens : undefined; + const comp = typeof u.completion_tokens === "number" ? u.completion_tokens : undefined; + if (prompt !== undefined || comp !== undefined) { + logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: comp ?? 0 }; + if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; + } + } noteSseFirstOutput(); controller.enqueue(encoder.encode(`data: ${JSON.stringify(rec)}\n\n`)); continue; @@ -754,20 +765,32 @@ async function handleChatCompletionsWithBudget( let synthParsed: unknown | null = null; try { synthParsed = JSON.parse(synthJsonText); } catch { synthParsed = null; } const rec2 = isRec(synthParsed) ? synthParsed as Rec : null; - const choices2 = rec2 && Array.isArray(rec2.choices) ? rec2.choices as Rec[] : []; - const msg2 = choices2[0] && isRec(choices2[0].message) ? choices2[0].message as Rec : null; - const usage2 = rec2 && isRec(rec2.usage) ? rec2.usage as Rec : undefined; + // Validate as Chat completion shape; otherwise surface as upstream protocol error, not silent success. + if (!rec2 || !Array.isArray(rec2.choices) || (rec2.choices as unknown[]).length === 0) { + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, "upstream returned an invalid chat completion", "server_error"); + } + const choices2 = rec2.choices as unknown[] as Rec[]; + const firstChoice2 = choices2[0] as Rec; + if (!isRec(firstChoice2.message) && firstChoice2.finish_reason === undefined) { + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, "upstream returned an invalid chat completion", "server_error"); + } + const msg2 = isRec(firstChoice2.message) ? firstChoice2.message as Rec : null; + const usage2 = isRec(rec2.usage) ? rec2.usage as Rec : undefined; if (usage2) { const prompt = typeof usage2.prompt_tokens === "number" ? usage2.prompt_tokens : undefined; const comp = typeof usage2.completion_tokens === "number" ? usage2.completion_tokens : undefined; if (prompt !== undefined || comp !== undefined) { logCtx.usage = { inputTokens: prompt ?? 0, outputTokens: comp ?? 0 }; if (logCtx.activeAttempt) logCtx.activeAttempt.usage = logCtx.usage; } } - const synthId = (rec2?.id as string) || `chatcmpl-${Date.now().toString(36)}`; - const synthCreated = typeof rec2?.created === "number" ? rec2?.created as number : Math.floor(Date.now() / 1000); - const synthModel = typeof rec2?.model === "string" ? rec2?.model as string : requestedModel; - const content2 = msg2 && typeof msg2.content === "string" ? msg2.content as string : (msg2?.content ?? null); + const synthId = typeof rec2.id === "string" ? rec2.id as string : `chatcmpl-${Date.now().toString(36)}`; + const synthCreated = typeof rec2.created === "number" ? rec2.created as number : Math.floor(Date.now() / 1000); + const synthModel = typeof rec2.model === "string" ? rec2.model as string : requestedModel; + const content2 = msg2 && typeof msg2.content === "string" ? msg2.content as string : (msg2 ? (msg2.content as unknown) : null); const toolCalls2 = msg2 && Array.isArray(msg2.tool_calls) ? msg2.tool_calls as unknown[] : undefined; - const finish2 = choices2[0] && typeof choices2[0].finish_reason === "string" ? choices2[0].finish_reason as string : "stop"; + const finish2 = typeof firstChoice2.finish_reason === "string" ? firstChoice2.finish_reason as string : "stop"; const encoder2 = new TextEncoder(); const synthStream = new ReadableStream({ start(controller) { @@ -877,7 +900,16 @@ async function handleChatCompletionsWithBudget( } cleanup(); let parsedJson: unknown | null = null; - try { parsedJson = JSON.parse(jsonText); } catch { parsedJson = null; } + try { parsedJson = JSON.parse(jsonText); } catch { + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, "upstream returned an invalid chat completion", "server_error"); + } + if (!isRec(parsedJson) || !Array.isArray((parsedJson as Rec).choices)) { + finishRequestAttempt(attempt, 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now())); + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 502, { closeReason: "non_stream" }); + return chatCompletionsErrorResponse(502, "upstream returned an invalid chat completion", "server_error"); + } if (isRec(parsedJson) && isRec((parsedJson as Rec).usage)) { const usage = (parsedJson as Rec).usage as Rec; const prompt = typeof usage.prompt_tokens === "number" ? usage.prompt_tokens : undefined; diff --git a/tests/chat-completions-endpoint.test.ts b/tests/chat-completions-endpoint.test.ts index c1f91997b2..3f2132dcd3 100644 --- a/tests/chat-completions-endpoint.test.ts +++ b/tests/chat-completions-endpoint.test.ts @@ -737,7 +737,7 @@ test("chat-native non-stream budget overflow returns 413 without hanging", async return Response.json({ id: "chatcmpl_ok", object: "chat.completion", choices: [{ index: 0, message: { role: "assistant", content: "ok" }, finish_reason: "stop" }], usage: { prompt_tokens: 1, completion_tokens: 1 } }); }, }); - saveConfig(mockConfig(`${upstream.url.toString().replace(/\/\$/, "")}/v1`)); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`)); const server = startServer(0); try { const response = await fetch(new URL("/v1/chat/completions", server.url), { @@ -763,15 +763,85 @@ test("chat-native non-stream budget overflow returns 413 without hanging", async } }); -test("chat-native non-stream invalid JSON is forwarded as 200 passthrough", async () => { - // The native JSON path forwards verbatim when parsing fails (bridges legacy behavior for untrusted upstreams). +test("chat-native streaming synthesizes SSE from valid Chat JSON", async () => { + const upstream = Bun.serve({ + port: 0, + fetch() { + return Response.json({ id: "chatcmpl_synth", object: "chat.completion", created: 1, model: "mock/test-model", choices: [{ index: 0, message: { role: "assistant", content: "hello" }, finish_reason: "stop" }], usage: { prompt_tokens: 1, completion_tokens: 1 } }); + }, + }); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`)); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: true, messages: [{ role: "user", content: "hi" }] }), + }); + expect(response.status).toBe(200); + const text = await response.text(); + expect(text).toContain("chat.completion.chunk"); + expect(text).toContain("hello"); + expect(text).toContain("data: [DONE]"); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + +test("chat-native streaming with invalid Chat JSON returns SSE error not fabricated success", async () => { + const upstream = Bun.serve({ + port: 0, + fetch() { + // missing choices + return Response.json({ id: "chatcmpl_bad", object: "chat.completion" }); + }, + }); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`)); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: true, messages: [{ role: "user", content: "hi" }] }), + }); + expect(response.status).toBe(502); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + +test("chat-native streaming with malformed JSON returns 502 not success", async () => { const upstream = Bun.serve({ port: 0, fetch() { return new Response("not-json-at-all", { headers: { "content-type": "application/json" } }); }, }); - saveConfig(mockConfig(`${upstream.url.toString().replace(/\/\$/, "")}/v1`)); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`)); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: true, messages: [{ role: "user", content: "hi" }] }), + }); + expect(response.status).toBe(502); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + +test("chat-native non-stream invalid JSON returns 502", async () => { + const upstream = Bun.serve({ + port: 0, + fetch() { + return new Response("not-json-at-all", { headers: { "content-type": "application/json" } }); + }, + }); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`)); const server = startServer(0); try { const response = await fetch(new URL("/v1/chat/completions", server.url), { @@ -779,8 +849,37 @@ test("chat-native non-stream invalid JSON is forwarded as 200 passthrough", asyn headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "mock/test-model", stream: false, messages: [{ role: "user", content: "hi" }] }), }); + expect(response.status).toBe(502); + } finally { + await server.stop(true); + upstream.stop(true); + } +}); + +test("chat-native streaming with tool_calls synthesizes via JSON path", async () => { + const upstream = Bun.serve({ + port: 0, + fetch() { + return Response.json({ + id: "chatcmpl_tc", + object: "chat.completion", + choices: [{ index: 0, message: { role: "assistant", content: null, tool_calls: [{ id: "call_1", type: "function", function: { name: "lookup", arguments: "{\"q\":1}" } }] }, finish_reason: "tool_calls" }], + usage: { prompt_tokens: 1, completion_tokens: 1 }, + }); + }, + }); + saveConfig(mockConfig(`${upstream.url.toString().replace(/\/$/, "")}/v1`)); + const server = startServer(0); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "mock/test-model", stream: true, messages: [{ role: "user", content: "hi" }], tools: [{ type: "function", function: { name: "lookup", parameters: { type: "object" } } }] }), + }); expect(response.status).toBe(200); - expect(await response.text()).toBe("not-json-at-all"); + const text = await response.text(); + expect(text).toContain("lookup"); + expect(text).toContain("tool_calls"); } finally { await server.stop(true); upstream.stop(true);