Something went wrong. Try again.
Something went wrong. Try again.
34 kB · 816 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817import fs from "node:fs"import os from "node:os"import path from "node:path"import crypto from "node:crypto"import { Readable } from "node:stream"import type { CodexWebSocketPool } from "./codex-websocket"import { tracedHttp, type TraceContext } from "./request-metrics"import { sseFrames } from "./sse"import Fastify, { type FastifyInstance } from "fastify"import { updateCredentialFileAsync } from "./credential-writer"import { extractSafeUpstreamError, fallbackUpstreamError } from "./upstream-error"import { CODEX_URL, codexAccountId, createCodexHeaders, createCodexRequestIdentity } from "./codex-wire"import { codexStartsNewTurn, transformCodexRequest } from "./codex-request"import { codexClientVersion, type CodexClientVersion } from "./codex-version"
const TOKEN_URL = "https://auth.openai.com/oauth/token"const CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"export { CODEX_CLIENT_VERSION } from "./codex-wire"const DEFAULT_MODEL = process.env.CODEX_BRIDGE_MODEL?.trim() || "gpt-5.6-sol"const DEFAULT_REASONING_EFFORT = process.env.CODEX_BRIDGE_REASONING_EFFORT?.trim() || "low"const DEFAULT_BODY_LIMIT = 64 * 1024 * 1024
type CodexAuth = { tokens?: { access_token?: string; refresh_token?: string; account_id?: string; id_token?: string } last_refresh?: string [key: string]: unknown}
type BridgeOptions = { authPath?: string fetchImpl?: typeof fetch /** Stable fallback used when the caller did not provide prompt_cache_key/user. */ promptCacheKey?: string signal?: AbortSignal websockets?: CodexWebSocketPool connectionScope?: string trace?: TraceContext /** Optional host attestation envelope for ChatGPT OAuth requests. */ attestation?: string clientVersion?: CodexClientVersion codexBetaFeatures?: string}
let bridgeServer: FastifyInstance | null = null
function decodeJwt(token: string): Record<string, any> | null { try { const payload = token.split(".")[1] return payload ? JSON.parse(Buffer.from(payload, "base64url").toString("utf8")) : null } catch { return null }}
function accountIdFromToken(token: string): string | undefined { return codexAccountId(token)}
/** * One refresh chain per credential file. Every request re-reads auth.json and * refreshes when the access token is nearly expired, so a burst of traffic * arriving in that window would otherwise present the same refresh token to the * token endpoint several times over. OpenAI rotates refresh tokens, and reuse * detection on a rotating credential can invalidate the chain outright — a * forced re-login on a process that had been up for days. Keyed by path so * unrelated credentials don't queue behind each other. */const codexRefreshChains = new Map<string, Promise<unknown>>()
export async function getCodexAccessToken( authPath: string, fetchImpl: typeof fetch = fetch,): Promise<{ token: string; accountId?: string }> { const previous = codexRefreshChains.get(authPath) ?? Promise.resolve() const next = previous.then(() => loadCodexAuth(authPath, fetchImpl)) // Keep the chain alive even when this attempt rejects, so a transient // upstream failure doesn't wedge every request queued behind it. codexRefreshChains.set( authPath, next.catch(() => {}), ) return next}
async function loadCodexAuth( authPath: string, fetchImpl: typeof fetch,): Promise<{ token: string; accountId?: string }> { const auth = JSON.parse(fs.readFileSync(authPath, "utf8")) as CodexAuth let token = auth.tokens?.access_token?.trim() const refresh = auth.tokens?.refresh_token?.trim() let accountId = auth.tokens?.account_id || (token ? accountIdFromToken(token) : undefined) if (!token) throw new Error(`Codex access token is missing from ${authPath}; run codex login`)
const expiresAt = Number(decodeJwt(token)?.exp ?? 0) * 1000 if (refresh && expiresAt <= Date.now() + 60_000) { const response = await fetchImpl(TOKEN_URL, { method: "POST", headers: { "content-type": "application/x-www-form-urlencoded" }, body: new URLSearchParams({ grant_type: "refresh_token", refresh_token: refresh, client_id: CLIENT_ID }), }) if (!response.ok) throw new Error(`Codex OAuth refresh failed (${response.status}): ${await response.text()}`) const next = (await response.json()) as { access_token?: string; refresh_token?: string; id_token?: string } if (!next.access_token) throw new Error("Codex OAuth refresh response did not contain an access token") const refreshedToken = next.access_token token = refreshedToken await updateCredentialFileAsync(authPath, (file) => { const latest = file as CodexAuth latest.tokens = { ...latest.tokens, access_token: refreshedToken, refresh_token: next.refresh_token || refresh || latest.tokens?.refresh_token, id_token: next.id_token || latest.tokens?.id_token, account_id: accountIdFromToken(refreshedToken) || latest.tokens?.account_id, } latest.last_refresh = new Date().toISOString() accountId = latest.tokens.account_id || accountIdFromToken(refreshedToken) }) } return { token, accountId }}
function mapModel(model: unknown, defaultModel?: string): string { const value = typeof model === "string" ? model.trim().toLowerCase() : "" if (!value || value === "codex" || value === "openai-codex" || value === "codex-latest") return defaultModel || DEFAULT_MODEL const requested = String(model).trim() return requested.toLowerCase().endsWith(":nitro") ? requested.slice(0, -":nitro".length) : requested}
function serviceTier(model: unknown): "priority" | undefined { if (typeof model === "string" && model.trim().toLowerCase().endsWith(":nitro")) return "priority" return undefined}
function mapMessageContent(role: "user" | "assistant", content: unknown): any[] { const textType = role === "assistant" ? "output_text" : "input_text" if (typeof content === "string") return [{ type: textType, text: content }] if (!Array.isArray(content)) return [{ type: textType, text: JSON.stringify(content ?? "") }] return content.map((part) => { if (!part || typeof part !== "object") return { type: textType, text: String(part ?? "") } if (part.type === "text") return { ...part, type: textType } if (part.type === "image_url") { const image = part.image_url return { type: "input_image", image_url: typeof image === "string" ? image : image?.url, ...(typeof image === "object" && image?.detail ? { detail: image.detail } : {}), } } return part })}
/** Return a client-facing reason when chat configuration updates cannot be safely routed. */export function codexConfigurationUpdateError( model: unknown, messages: unknown, codexBacked = true,): string | undefined { if (!Array.isArray(messages)) return undefined let hasUpdates = false let previousWasUpdate = false for (const [index, rawMessage] of messages.entries()) { const message = rawMessage && typeof rawMessage === "object" && !Array.isArray(rawMessage) ? (rawMessage as Record<string, unknown>) : undefined const update = message?.type === "configuration_update" if (update) { hasUpdates = true const keys = message ? Object.keys(message).sort() : [] if (keys.length !== 2 || keys[0] !== "reasoning" || keys[1] !== "type") { return `messages[${index}] configuration_update must contain only type and reasoning` } const reasoning = message.reasoning && typeof message.reasoning === "object" && !Array.isArray(message.reasoning) ? (message.reasoning as Record<string, unknown>) : undefined const reasoningKeys = reasoning ? Object.keys(reasoning) : [] if ( !reasoning || reasoningKeys.length !== 1 || reasoningKeys[0] !== "effort" || typeof reasoning.effort !== "string" || !reasoning.effort.trim() ) { return `messages[${index}] configuration_update requires a non-empty reasoning.effort string` } if (previousWasUpdate) return "Adjacent configuration_update messages are not supported; place each update between conversation messages" } previousWasUpdate = update } if (!hasUpdates) return undefined if (!codexBacked) return "configuration_update is supported only by GPT-6 Codex chat completions" const requestedModel = typeof model === "string" ? model.trim() : "" const routedModel = requestedModel.toLowerCase().endsWith(":nitro") ? requestedModel.slice(0, -":nitro".length) : requestedModel if (!/^gpt-6(?:[.-]|$)/i.test(routedModel)) return "configuration_update is supported only by GPT-6 Codex chat completions" return undefined}
function configurationUpdateResponse(message: string): Response { return new Response(JSON.stringify({ error: { message, type: "invalid_request_error" } }), { status: 400, headers: { "content-type": "application/json" }, })}
function mapMessages(messages: any[]): { instructions?: string; input: any[] } { const instructions: string[] = [] const input: any[] = [] for (const message of messages || []) { const role = message?.role if (message?.type === "compaction" || message?.type === "configuration_update") { input.push(message) } else if (role === "system" || role === "developer") { instructions.push(typeof message.content === "string" ? message.content : JSON.stringify(message.content)) } else if (role === "tool") { input.push({ type: "function_call_output", call_id: message.tool_call_id, output: typeof message.content === "string" ? message.content : JSON.stringify(message.content), }) } else if (role === "assistant" && Array.isArray(message.tool_calls)) { if (message.content) input.push({ role: "assistant", content: [{ type: "output_text", text: String(message.content) }] }) for (const call of message.tool_calls) input.push({ type: "function_call", call_id: call.id, name: call.function?.name, arguments: call.function?.arguments || "{}", }) } else if (role === "user" || role === "assistant") { input.push({ role, content: mapMessageContent(role, message.content) }) } } return { instructions: instructions.length ? instructions.join("\n\n") : undefined, input }}
function mapTools(tools: any[]): any[] | undefined { if (!Array.isArray(tools) || tools.length === 0) return undefined return tools .filter((tool) => tool?.type === "function") .map((tool) => ({ type: "function", name: tool.function.name, description: tool.function.description, parameters: tool.function.parameters || { type: "object", properties: {} }, strict: tool.function.strict ?? false, }))}
function parseSse(text: string): any[] { const events: any[] = [] for (const block of text.split(/\r?\n\r?\n/)) { const data = block .split(/\r?\n/) .filter((line) => line.startsWith("data:")) .map((line) => line.slice(5).trim()) .join("\n") if (!data || data === "[DONE]") continue try { events.push(JSON.parse(data)) } catch { /* ignore non-JSON keepalives */ } } return events}
function collectResult(events: any[]): { content: string toolCalls: any[] usage?: any responseId?: string terminal?: "completed" | "incomplete" | "failed" incompleteReason?: string error?: { message?: string; code?: string; type?: string; status?: number }} { let content = "" const toolCalls: any[] = [] let usage: any let responseId: string | undefined let terminal: "completed" | "incomplete" | "failed" | undefined let incompleteReason: string | undefined let error: { message?: string; code?: string; type?: string; status?: number } | undefined for (const event of events) { if (event.type === "response.output_text.delta") content += event.delta || "" if (event.type === "response.output_item.done" && event.item?.type === "function_call") { toolCalls.push({ id: event.item.call_id || event.item.id, type: "function", function: { name: event.item.name, arguments: event.item.arguments || "{}" }, }) } if (event.response) { responseId ||= event.response.id usage ||= event.response.usage } if (event.type === "response.completed") terminal = "completed" if (event.type === "response.incomplete") { terminal = "incomplete" incompleteReason = event.response?.incomplete_details?.reason } if (event.type === "response.failed" || event.type === "error" || event.type === "response.error") { terminal = "failed" const candidate = event.error ?? event.response?.error error = candidate && typeof candidate === "object" ? candidate : { message: typeof event.message === "string" ? event.message : "upstream response failed" } } } return { content, toolCalls, usage, responseId, terminal, incompleteReason, error }}
function completionUsage(usage: any): any | undefined { if (!usage || typeof usage !== "object") return undefined const promptTokens = Number(usage.input_tokens ?? usage.prompt_tokens ?? 0) const completionTokens = Number(usage.output_tokens ?? usage.completion_tokens ?? 0) const totalTokens = Number(usage.total_tokens ?? promptTokens + completionTokens) const inputDetails = usage.input_tokens_details ?? usage.prompt_tokens_details const outputDetails = usage.output_tokens_details ?? usage.completion_tokens_details
return { prompt_tokens: Number.isFinite(promptTokens) ? promptTokens : 0, completion_tokens: Number.isFinite(completionTokens) ? completionTokens : 0, total_tokens: Number.isFinite(totalTokens) ? totalTokens : 0, ...(inputDetails && typeof inputDetails === "object" ? { prompt_tokens_details: { ...(typeof inputDetails.cached_tokens === "number" ? { cached_tokens: inputDetails.cached_tokens } : {}), ...(typeof inputDetails.cache_write_tokens === "number" ? { cache_write_tokens: inputDetails.cache_write_tokens } : {}), ...(typeof inputDetails.audio_tokens === "number" ? { audio_tokens: inputDetails.audio_tokens } : {}), }, } : {}), ...(outputDetails && typeof outputDetails === "object" ? { completion_tokens_details: { ...outputDetails } } : {}), }}
export interface CodexHandlerResult { status: number headers: Record<string, string> body: string stream?: AsyncIterable<string> // Upstream x-codex-* rate-limit headers (usage snapshot), if any. Kept // separate from `headers` so they're not forwarded to the client. rateLimitHeaders?: Record<string, string> upstreamError?: { errorType: string; message: string }}
/** Pull the x-codex-* rate-limit header family off an upstream response. */export function extractCodexRateLimitHeaders(h: Headers): Record<string, string> { const out: Record<string, string> = {} if (h && typeof h.forEach === "function") { h.forEach((value: string, key: string) => { const lower = key.toLowerCase() if (lower.startsWith("x-codex-")) out[lower] = value }) } return out}
function normalizeCodexInput(input: unknown): any[] { if (typeof input === "string") return [{ role: "user", content: [{ type: "input_text", text: input }] }] return Array.isArray(input) ? input : []}
const RETRYABLE_CODEX_STATUSES = new Set([408, 425, 500, 502, 503, 504])
function retryDelay(response: Response | undefined, attempt: number): number { const retryAfter = response?.headers.get("retry-after")?.trim() if (retryAfter) { const seconds = Number(retryAfter) if (Number.isFinite(seconds) && seconds >= 0) return Math.min(seconds * 1_000, 30_000) const date = Date.parse(retryAfter) if (Number.isFinite(date)) return Math.min(Math.max(0, date - Date.now()), 30_000) } return Math.min(250 * 2 ** attempt, 4_000)}
async function waitForRetry(ms: number, signal?: AbortSignal): Promise<void> { if (signal?.aborted) throw signal.reason ?? new Error("Request aborted") await new Promise<void>((resolve, reject) => { const done = () => { signal?.removeEventListener("abort", aborted) resolve() } const aborted = () => { clearTimeout(timer) signal?.removeEventListener("abort", aborted) reject(signal?.reason ?? new Error("Request aborted")) } const timer = setTimeout(done, ms) // Deliberately referenced: this backoff is part of an in-flight request, so // an unref'd timer lets the event loop drain mid-retry and the awaiting // caller never settles (node:test reports it as cancelledByParent). signal?.addEventListener("abort", aborted, { once: true }) if (signal?.aborted) aborted() })}
/** Retry only replay-safe failures before any upstream body has been exposed. */async function fetchCodexSse( fetchImpl: typeof fetch, body: Record<string, unknown>, headers: Record<string, string>, signal?: AbortSignal,): Promise<Response> { const configured = Number.parseInt(process.env.CODEX_SSE_MAX_ATTEMPTS || "", 10) const maxAttempts = Number.isInteger(configured) && configured > 0 ? configured : 3 const bodyJson = JSON.stringify(body) let lastError: unknown for (let attempt = 0; attempt < maxAttempts; attempt++) { signal?.throwIfAborted() let response: Response | undefined try { response = await fetchImpl(CODEX_URL, { method: "POST", headers, body: bodyJson, signal }) if (!RETRYABLE_CODEX_STATUSES.has(response.status) || attempt + 1 >= maxAttempts) return response await response.body?.cancel().catch(() => {}) } catch (error) { lastError = error if (attempt + 1 >= maxAttempts) break } await waitForRetry(retryDelay(response, attempt), signal) } throw lastError instanceof Error ? lastError : new Error("Codex request failed")}
/** * Where this bridge reads (and the admin login writes) the codex credential. * Single source of truth: a login that writes somewhere else is a login that * silently doesn't take effect. */export function resolveCodexAuthPath(provider?: { auth_path?: string }, override?: string): string { return ( override || provider?.auth_path || process.env.CODEX_AUTH_PATH?.trim() || path.join(os.homedir(), ".codex", "auth.json") )}
/** * Build and send the whitelisted Codex Responses request. The raw upstream * Response is returned so callers can choose how to consume its body. */export async function codexResponsesRequest( requestBody: any, provider?: { model?: string; reasoning_effort?: string; auth_path?: string }, options: BridgeOptions = {},): Promise<Response> { const authPath = resolveCodexAuthPath(provider, options.authPath) const fetchImpl = options.fetchImpl || fetch const defaultModel = provider?.model || process.env.CODEX_BRIDGE_MODEL?.trim() || DEFAULT_MODEL const defaultReasoning = provider?.reasoning_effort || process.env.CODEX_BRIDGE_REASONING_EFFORT?.trim() || DEFAULT_REASONING_EFFORT const legacyChat = requestBody.input === undefined && requestBody.messages !== undefined const model = mapModel(requestBody.model, defaultModel) const updateError = legacyChat ? codexConfigurationUpdateError(model, requestBody.messages) : undefined if (updateError) return configurationUpdateResponse(updateError) const { token, accountId } = await getCodexAccessToken(authPath, fetchImpl) const mapped = legacyChat ? mapMessages(requestBody.messages || []) : { instructions: requestBody.instructions, input: normalizeCodexInput(requestBody.input) } const legacyReasoning = { effort: requestBody.reasoning_effort || defaultReasoning } const legacyText = { verbosity: requestBody.verbosity || "low" } const selectedServiceTier = requestBody.service_tier ?? serviceTier(requestBody.model) const requestedCacheKey = typeof requestBody.prompt_cache_key === "string" && requestBody.prompt_cache_key.trim() ? requestBody.prompt_cache_key.trim() : typeof options.promptCacheKey === "string" && options.promptCacheKey.trim() ? options.promptCacheKey.trim() : undefined const requestedUser = typeof requestBody.user === "string" && requestBody.user.trim() ? requestBody.user.trim() : undefined const identityScope = options.connectionScope || (typeof requestBody.conversation === "string" ? requestBody.conversation : requestBody.conversation?.id) || requestedUser || requestedCacheKey || crypto.randomUUID() const identity = createCodexRequestIdentity({ scope: identityScope, startsNewTurn: codexStartsNewTurn(mapped.input), clientMetadata: requestBody.client_metadata, }) const responsesLite = requestBody.responses_lite === true || requestBody.use_responses_lite === true const baseBody: any = { model, input: mapped.input, instructions: mapped.instructions, tools: legacyChat ? mapTools(requestBody.tools) : requestBody.tools, tool_choice: requestBody.tool_choice, stream: true, store: false, reasoning: legacyChat ? legacyReasoning : (requestBody.reasoning ?? legacyReasoning), text: legacyChat ? legacyText : (requestBody.text ?? legacyText), parallel_tool_calls: requestBody.parallel_tool_calls, prompt_cache_options: requestBody.prompt_cache_options, include: requestBody.include, service_tier: selectedServiceTier, ...(options.websockets && requestBody.previous_response_id ? { previous_response_id: requestBody.previous_response_id } : {}), ...(options.websockets && requestBody.generate === false ? { generate: false } : {}), prompt_cache_key: requestedCacheKey ?? identity.sessionId, client_metadata: identity.clientMetadata, } for (const key of Object.keys(baseBody)) if (baseBody[key] === undefined) delete baseBody[key] const upstreamBody = transformCodexRequest(baseBody, { responsesLite }) return (options.clientVersion ?? codexClientVersion).request((clientVersion) => { const headers = createCodexHeaders({ token, accountId, identity, model, serviceTier: selectedServiceTier, responsesLite, attestation: options.attestation, clientVersion, codexBetaFeatures: options.codexBetaFeatures, }) const fallback = () => tracedHttp(options.trace, upstreamBody, headers, () => fetchCodexSse(fetchImpl, upstreamBody, headers, options.signal), ) return options.websockets && options.connectionScope ? options.websockets.request({ scope: options.connectionScope, url: CODEX_URL, headers, body: upstreamBody as any, chat: legacyChat, signal: options.signal, fallback, trace: options.trace, }) : fallback() })}
/** Translate actual Responses deltas, not a timed replay of a completed response. * https://developers.openai.com/api/docs/guides/function-calling#streaming */async function* codexChatStream( upstream: Response, model: string, includeUsage: boolean, signal?: AbortSignal,): AsyncGenerator<string> { const id = `chatcmpl-${crypto.randomUUID()}` const created = Math.floor(Date.now() / 1000) const chunk = (delta: object, finish: string | null = null) => `data: ${JSON.stringify({ id, object: "chat.completion.chunk", created, model, choices: [{ index: 0, delta, finish_reason: finish }] })}\n\n` type Call = { index: number; arguments: string } const calls = new Map<number, Call>() const startCall = (outputIndex: number, item: any): string[] => { if (calls.has(outputIndex)) return [] const call = { index: calls.size, arguments: item.arguments || "" } calls.set(outputIndex, call) return [ chunk({ tool_calls: [ { index: call.index, id: item.call_id || item.id, type: "function", function: { name: item.name, arguments: call.arguments }, }, ], }), ] } const finishArguments = (outputIndex: number, arguments_: string): string[] => { const call = calls.get(outputIndex) if (!call) throw new Error("tool arguments arrived without a function-call item") if (!arguments_.startsWith(call.arguments)) throw new Error("upstream revised already-streamed tool arguments") const delta = arguments_.slice(call.arguments.length) call.arguments = arguments_ return delta ? [chunk({ tool_calls: [{ index: call.index, function: { arguments: delta } }] })] : [] } let completed = false try { if (!upstream.body) throw new Error("upstream response has no stream") yield chunk({ role: "assistant" }) for await (const raw of sseFrames(upstream.body)) { const event = raw as any if (event.type === "response.output_text.delta") { yield chunk({ content: event.delta || "" }) } else if ( event.type === "response.reasoning_summary_text.delta" || event.type === "response.reasoning_text.delta" ) { yield chunk({ reasoning_content: event.delta || "" }) } else if (event.type === "response.output_item.added" && event.item?.type === "function_call") { yield* startCall(event.output_index, event.item) } else if (event.type === "response.function_call_arguments.delta") { const call = calls.get(event.output_index) if (!call) throw new Error("tool arguments arrived without a function-call item") const delta = event.delta || "" call.arguments += delta yield chunk({ tool_calls: [{ index: call.index, function: { arguments: delta } }] }) } else if (event.type === "response.function_call_arguments.done") { yield* finishArguments(event.output_index, event.arguments || "") } else if (event.type === "response.output_item.done" && event.item?.type === "function_call") { yield* startCall(event.output_index, event.item) yield* finishArguments(event.output_index, event.item.arguments || "") } else if (event.type === "response.completed" || event.type === "response.incomplete") { const response = event.response if (response?.status === "failed" || response?.error) throw new Error(response.error?.message || "upstream response failed") for (const [index, item] of (response?.output ?? []).entries()) { if (item.type !== "function_call") continue yield* startCall(index, item) yield* finishArguments(index, item.arguments || "") } const finish = event.type === "response.incomplete" ? response?.incomplete_details?.reason === "content_filter" ? "content_filter" : "length" : calls.size ? "tool_calls" : "stop" yield chunk({}, finish) const usage = completionUsage(response?.usage) if (includeUsage && usage) yield `data: ${JSON.stringify({ id, object: "chat.completion.chunk", created, model, choices: [], usage })}\n\n` completed = true break } else if (event.type === "error" || event.type === "response.failed") { throw new Error( event.response?.error?.message || event.error?.message || event.message || "upstream response failed", ) } } if (!completed) throw new Error("upstream stream ended before response completion") } catch (error) { if (signal?.aborted) return yield `data: ${JSON.stringify({ error: { type: "codex_upstream_error", message: error instanceof Error ? error.message : "upstream stream failed" } })}\n\n` } yield "data: [DONE]\n\n"}
/** * Handle a chat completion request using the Codex backend. * Standalone function — can be called from any server context. */export async function handleCodexCompletion( requestBody: any, provider?: { model?: string; reasoning_effort?: string; auth_path?: string }, options: BridgeOptions = {},): Promise<CodexHandlerResult> { const upstream = await codexResponsesRequest(requestBody, provider, options) return codexCompletionResponse(requestBody, provider, upstream, options)}
export async function codexCompletionResponse( requestBody: any, provider: Parameters<typeof codexResponsesRequest>[1], upstream: Response, options: BridgeOptions = {},): Promise<CodexHandlerResult> { const defaultModel = provider?.model || process.env.CODEX_BRIDGE_MODEL?.trim() || DEFAULT_MODEL const rateLimitHeaders = extractCodexRateLimitHeaders(upstream.headers) if (!upstream.ok) { const raw = await upstream.text() const upstreamError = extractSafeUpstreamError(raw) ?? fallbackUpstreamError(upstream.status, upstream.statusText) return { status: upstream.status, headers: { "content-type": "application/json" }, body: JSON.stringify({ error: { message: raw || upstream.statusText, type: "codex_upstream_error" } }), upstreamError, rateLimitHeaders, } } if (requestBody.stream) { return { status: 200, headers: { "content-type": "text/event-stream; charset=utf-8", "cache-control": "no-cache" }, body: "", stream: codexChatStream( upstream, mapModel(requestBody.model, defaultModel), requestBody.stream_options?.include_usage === true, options.signal, ), rateLimitHeaders, } } const result = collectResult(parseSse(await upstream.text())) if (result.terminal === "failed") { const status = Number(result.error?.status) return { status: Number.isInteger(status) && status >= 400 && status <= 599 ? status : 502, headers: { "content-type": "application/json" }, rateLimitHeaders, upstreamError: { errorType: result.error?.code || result.error?.type || "codex_upstream_error", message: result.error?.message || "upstream response failed", }, body: JSON.stringify({ error: { ...result.error, message: result.error?.message || "upstream response failed", type: result.error?.type || "codex_upstream_error", }, }), } } if (!result.terminal) { return { status: 502, headers: { "content-type": "application/json" }, rateLimitHeaders, upstreamError: { errorType: "codex_upstream_error", message: "upstream stream ended before response completion" }, body: JSON.stringify({ error: { message: "upstream stream ended before response completion", type: "codex_upstream_error" }, }), } } const usage = completionUsage(result.usage) const created = Math.floor(Date.now() / 1000) const id = result.responseId || `chatcmpl-${crypto.randomUUID()}` const finishReason = result.terminal === "incomplete" ? result.incompleteReason === "content_filter" ? "content_filter" : "length" : result.toolCalls.length ? "tool_calls" : "stop" return { status: 200, headers: { "content-type": "application/json" }, rateLimitHeaders, body: JSON.stringify({ id, object: "chat.completion", created, model: mapModel(requestBody.model, defaultModel), choices: [ { index: 0, message: { role: "assistant", content: result.content || null, ...(result.toolCalls.length ? { tool_calls: result.toolCalls } : {}), }, finish_reason: finishReason, }, ], usage, }), }}
export function createCodexBridgeServer(options: BridgeOptions = {}): FastifyInstance { const configuredBodyLimit = Number.parseInt(process.env.CODEX_BRIDGE_BODY_LIMIT_BYTES || "", 10) const bodyLimit = Number.isSafeInteger(configuredBodyLimit) && configuredBodyLimit > 0 ? configuredBodyLimit : DEFAULT_BODY_LIMIT const app = Fastify({ logger: false, bodyLimit }) const authPath = resolveCodexAuthPath(undefined, options.authPath) const fetchImpl = options.fetchImpl || fetch
app.get("/health", async () => ({ ok: true, authPath })) app.get("/v1/models", async () => ({ object: "list", data: [ { id: "Codex", object: "model" }, { id: DEFAULT_MODEL, object: "model" }, ], })) app.post("/v1/chat/completions", async (request, reply) => { const abort = new AbortController() reply.raw.once("close", () => abort.abort()) const result = await handleCodexCompletion(request.body, undefined, { ...options, authPath, fetchImpl, signal: abort.signal, }) return reply .code(result.status) .headers(result.headers) .send(result.stream ? Readable.from(result.stream) : result.body) }) return app}
export async function startCodexBridge(): Promise<void> { if (process.env.CODEX_BRIDGE_ENABLED?.trim().toLowerCase() !== "true" || bridgeServer) return const port = Number.parseInt(process.env.CODEX_BRIDGE_PORT || "8001", 10) if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error(`invalid Codex bridge port: ${process.env.CODEX_BRIDGE_PORT}`) bridgeServer = createCodexBridgeServer() await bridgeServer.listen({ port, host: "127.0.0.1" }) console.log(`[codex bridge] listening on http://127.0.0.1:${port}`)}
export async function stopCodexBridge(): Promise<void> { if (!bridgeServer) return await bridgeServer.close() bridgeServer = null}
export const __codexBridgeTest = { parseSse }