Files
codex-retry-gateway/gateway.mjs
T

3787 lines
123 KiB
JavaScript

#!/usr/bin/env node
import http from "node:http";
import { spawn } from "node:child_process";
import { createHash } from "node:crypto";
import { chmod, copyFile, mkdir, readFile, readdir, rm, writeFile } from "node:fs/promises";
import fs from "node:fs";
import path from "node:path";
import { DatabaseSync } from "node:sqlite";
import zlib from "node:zlib";
import { TextDecoder } from "node:util";
import { fileURLToPath } from "node:url";
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
const ADMIN_BASE_PATH = "/__codex_retry_gateway";
const UI_PATH = `${ADMIN_BASE_PATH}/ui`;
const UI_STATIC_ROOT = path.join(__dirname, "public", "ui");
const STATUS_API_PATH = `${ADMIN_BASE_PATH}/api/status`;
const CONFIG_API_PATH = `${ADMIN_BASE_PATH}/api/config`;
const LOGS_API_PATH = `${ADMIN_BASE_PATH}/api/logs`;
const REQUESTS_API_PATH = `${ADMIN_BASE_PATH}/api/requests`;
const PROFILES_API_PATH = `${ADMIN_BASE_PATH}/api/profiles`;
const PROFILE_PROBE_API_PATH = `${ADMIN_BASE_PATH}/api/profiles/probe`;
const PROFILE_SWITCH_API_PATH = `${ADMIN_BASE_PATH}/api/profiles/switch`;
const PROFILE_ITEM_API_PREFIX = `${ADMIN_BASE_PATH}/api/profiles/`;
const RESTORE_API_PATH = `${ADMIN_BASE_PATH}/api/restore`;
const STATUS_REASONING_COUNT_LIMIT = 24;
const MANAGEMENT_ACCESS_COOKIE = "codex_retry_gateway_access";
const DEFAULT_CONFIG = {
profile_name: "default",
listen_host: "127.0.0.1",
listen_port: 4610,
upstream_base_url: "",
upstream_auth_mode: "passthrough",
upstream_auth_env: "CODEX_RETRY_GATEWAY_UPSTREAM_API_KEY",
upstream_auth_file: "",
upstream_auth_json_path: "",
upstream_auth_json_key: "OPENAI_API_KEY",
request_body_limit_bytes: 1024 * 1024 * 1024,
request_history_limit: 0,
model_remap: "",
endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"],
reasoning_equals: [516],
retryable_status_codes: [429, 503],
retryable_error_messages: [
"Selected model is at capacity. Please try a different model.",
"stream disconnected before completion: Concurrency limit exceeded for account, please retry later",
],
management_access_key: "",
upstream_fetch_retry_attempts: 5,
upstream_fetch_retry_backoff_ms: 350,
non_stream_status_code: 502,
stream_action: "strict_502",
log_match: true,
health_path: "/__codex_retry_gateway/health",
};
const REASONING_POINTERS = [
"/usage/output_tokens_details/reasoning_tokens",
"/usage/completion_tokens_details/reasoning_tokens",
"/response/usage/output_tokens_details/reasoning_tokens",
"/response/usage/completion_tokens_details/reasoning_tokens",
];
const REQUEST_THREAD_ID_POINTERS = [
"/thread_id",
"/client_metadata/thread_id",
"/client_metadata/x-codex-thread-id",
"/thread",
"/thread/id",
"/threadId",
"/client_metadata/threadId",
"/conversation_id",
"/conversationId",
"/conversation",
"/conversation/id",
"/metadata/thread_id",
"/metadata/threadId",
"/metadata/conversation_id",
"/metadata/conversationId",
"/x-codex-turn-metadata/thread_id",
"/x-codex-turn-metadata/threadId",
"/x-codex-turn-metadata/conversation_id",
"/x-codex-turn-metadata/conversationId",
"/previous_response_id",
];
const RESPONSE_THREAD_ID_POINTERS = [
"/thread_id",
"/client_metadata/thread_id",
"/client_metadata/x-codex-thread-id",
"/thread",
"/thread/id",
"/threadId",
"/client_metadata/threadId",
"/conversation_id",
"/conversationId",
"/conversation",
"/conversation/id",
"/response/thread_id",
"/response/threadId",
"/response/thread",
"/response/thread/id",
"/response/conversation_id",
"/response/conversationId",
"/response/conversation",
"/response/conversation/id",
];
const NON_STREAM_RESPONSE_ID_POINTERS = [
"/id",
"/response_id",
"/response/id",
];
const STREAM_RESPONSE_ID_POINTERS = [
"/response_id",
"/response/id",
];
function parseArgs(argv) {
const args = { config: null, log: null };
for (let i = 2; i < argv.length; i += 1) {
const current = argv[i];
if (current === "--config") {
args.config = argv[i + 1];
i += 1;
} else if (current === "--log") {
args.log = argv[i + 1];
i += 1;
} else if (current === "--help" || current === "-h") {
printHelp();
process.exit(0);
}
}
return args;
}
function printHelp() {
process.stdout.write(
[
"用法:",
" node gateway.mjs --config <config.json> [--log <gateway.log>]",
"",
"说明:",
" 独立 Codex 本地重试网关。",
" 非流式命中 reasoning_tokens=516 时返回 502。",
" 流式命中时默认缓存并返回 502,避免半截流返回。",
"",
].join("\n"),
);
}
function normalizePath(inputPath) {
const [withoutQuery] = `${inputPath || "/"}`.split("?");
const trimmed = withoutQuery.length > 1 ? withoutQuery.replace(/\/+$/, "") : withoutQuery;
return trimmed || "/";
}
function flattenValues(value) {
if (Array.isArray(value)) {
return value.flatMap((item) => flattenValues(item));
}
return [value];
}
function isJsonContentType(contentType) {
return `${contentType || ""}`.toLowerCase().includes("application/json");
}
function isSseContentType(contentType) {
return `${contentType || ""}`.toLowerCase().includes("text/event-stream");
}
function jsonPointerGet(value, pointer) {
if (!pointer.startsWith("/")) {
return undefined;
}
return pointer
.slice(1)
.split("/")
.map((segment) => segment.replace(/~1/g, "/").replace(/~0/g, "~"))
.reduce((current, segment) => {
if (current === null || current === undefined) {
return undefined;
}
return current[segment];
}, value);
}
function extractReasoningTokens(payload) {
for (const pointer of REASONING_POINTERS) {
const raw = jsonPointerGet(payload, pointer);
if (Number.isInteger(raw)) {
return raw;
}
}
return null;
}
function firstInteger(...values) {
for (const value of values) {
if (Number.isInteger(value)) {
return value;
}
}
return null;
}
function firstNonEmptyString(...values) {
for (const value of values) {
if (typeof value !== "string") {
continue;
}
const trimmed = value.trim();
if (trimmed) {
return trimmed;
}
}
return null;
}
function timingSafeEquals(left, right) {
const leftText = Buffer.from(`${left || ""}`);
const rightText = Buffer.from(`${right || ""}`);
if (leftText.length !== rightText.length) {
return false;
}
let mismatch = 0;
for (let i = 0; i < leftText.length; i += 1) {
mismatch |= leftText[i] ^ rightText[i];
}
return mismatch === 0;
}
function parseCookieHeader(cookieHeader) {
const cookies = {};
for (const part of `${cookieHeader || ""}`.split(";")) {
const trimmed = part.trim();
if (!trimmed) {
continue;
}
const separatorIndex = trimmed.indexOf("=");
if (separatorIndex <= 0) {
continue;
}
const key = trimmed.slice(0, separatorIndex).trim();
const value = trimmed.slice(separatorIndex + 1).trim();
cookies[key] = value;
}
return cookies;
}
function normalizeManagementAccessKey(value) {
return `${value || ""}`.trim();
}
function currentManagementAccessKey(config) {
return normalizeManagementAccessKey(config?.management_access_key);
}
function managementAccessEnabled(config) {
return Boolean(currentManagementAccessKey(config));
}
function requestManagementAccessKey(req, requestUrl) {
const headerKey = firstNonEmptyString(
req.headers["x-codex-retry-gateway-key"],
req.headers["x-codex-retry-gateway-access-key"],
req.headers.authorization?.startsWith?.("Bearer ")
? req.headers.authorization.slice("Bearer ".length)
: null,
);
if (headerKey) {
return headerKey;
}
const queryKey = firstNonEmptyString(
requestUrl?.searchParams?.get("key"),
requestUrl?.searchParams?.get("access_key"),
);
if (queryKey) {
return queryKey;
}
const cookies = parseCookieHeader(req.headers.cookie || "");
return firstNonEmptyString(cookies[MANAGEMENT_ACCESS_COOKIE]);
}
function hasManagementAccess(req, requestUrl, config) {
const expected = currentManagementAccessKey(config);
if (!expected) {
return true;
}
const provided = requestManagementAccessKey(req, requestUrl);
return Boolean(provided) && timingSafeEquals(provided, expected);
}
function clearManagementAccessCookieHeaders() {
return [
`${MANAGEMENT_ACCESS_COOKIE}=; Path=${ADMIN_BASE_PATH}; HttpOnly; SameSite=Lax; Max-Age=0`,
];
}
function buildManagementAccessCookieHeaders(config, accessKey) {
const expected = currentManagementAccessKey(config);
if (!expected || !accessKey || !timingSafeEquals(accessKey, expected)) {
return clearManagementAccessCookieHeaders();
}
return [
`${MANAGEMENT_ACCESS_COOKIE}=${accessKey}; Path=${ADMIN_BASE_PATH}; HttpOnly; SameSite=Lax`,
];
}
function managementUnauthorizedPayload() {
return {
error: {
message: "management access key required",
code: "management_access_key_required",
},
};
}
function extractStringByPointers(payload, pointers) {
for (const pointer of pointers) {
const raw = jsonPointerGet(payload, pointer);
if (typeof raw !== "string") {
continue;
}
const trimmed = raw.trim();
if (trimmed) {
return trimmed;
}
}
return null;
}
function extractRequestThreadId(payload) {
return extractStringByPointers(payload, REQUEST_THREAD_ID_POINTERS);
}
function extractResponseThreadId(payload) {
return extractStringByPointers(payload, RESPONSE_THREAD_ID_POINTERS);
}
function extractHeaderThreadId(headers) {
if (!headers || typeof headers !== "object") {
return null;
}
const candidates = [
"thread-id",
"x-client-request-id",
"conversation_id",
"conversation-id",
"session_id",
"session-id",
"x-codex-parent-thread-id",
];
for (const key of candidates) {
const raw = headers[key];
if (Array.isArray(raw)) {
for (const item of raw) {
const text = firstNonEmptyString(item);
if (text) {
return text;
}
}
continue;
}
const text = firstNonEmptyString(raw);
if (text) {
return text;
}
}
return null;
}
function extractNonStreamingResponseId(payload) {
return extractStringByPointers(payload, NON_STREAM_RESPONSE_ID_POINTERS);
}
function extractStreamingResponseId(payload) {
return extractStringByPointers(payload, STREAM_RESPONSE_ID_POINTERS);
}
function normalizeUsageSnapshot(payload) {
const usage = payload?.usage || payload?.response?.usage || null;
if (!usage || typeof usage !== "object") {
return null;
}
const inputTokens = firstInteger(usage.input_tokens, usage.prompt_tokens);
const outputTokens = firstInteger(usage.output_tokens, usage.completion_tokens);
const totalTokens = firstInteger(
usage.total_tokens,
inputTokens !== null && outputTokens !== null ? inputTokens + outputTokens : null,
);
const reasoningTokens = firstInteger(
usage.output_tokens_details?.reasoning_tokens,
usage.completion_tokens_details?.reasoning_tokens,
payload?.response?.usage?.output_tokens_details?.reasoning_tokens,
payload?.response?.usage?.completion_tokens_details?.reasoning_tokens,
);
const cachedTokens = firstInteger(
usage.input_tokens_details?.cached_tokens,
usage.prompt_tokens_details?.cached_tokens,
payload?.response?.usage?.input_tokens_details?.cached_tokens,
payload?.response?.usage?.prompt_tokens_details?.cached_tokens,
);
if (
inputTokens === null &&
outputTokens === null &&
totalTokens === null &&
reasoningTokens === null &&
cachedTokens === null
) {
return null;
}
return {
input_tokens: inputTokens,
output_tokens: outputTokens,
total_tokens: totalTokens,
reasoning_tokens: reasoningTokens,
cached_tokens: cachedTokens,
};
}
function mergeUsageSnapshots(current, next) {
if (!next) {
return current || null;
}
return {
input_tokens: next.input_tokens ?? current?.input_tokens ?? null,
output_tokens: next.output_tokens ?? current?.output_tokens ?? null,
total_tokens: next.total_tokens ?? current?.total_tokens ?? null,
reasoning_tokens: next.reasoning_tokens ?? current?.reasoning_tokens ?? null,
cached_tokens: next.cached_tokens ?? current?.cached_tokens ?? null,
};
}
function normalizeIntegerList(values, fallback = []) {
const source = values === undefined || values === null ? fallback : values;
const normalized = flattenValues(source)
.flatMap((value) => {
if (typeof value === "string") {
return value.split(/[\s,]+/).filter(Boolean);
}
return [value];
})
.map((value) => Number.parseInt(`${value}`, 10))
.filter((value) => Number.isInteger(value));
return [...new Set(normalized)];
}
function normalizeStringList(values, fallback = []) {
const source = values === undefined || values === null ? fallback : values;
const normalized = flattenValues(source)
.flatMap((value) => `${value ?? ""}`.split(/[\s,]+/))
.map((value) => value.trim())
.filter(Boolean);
return [...new Set(normalized)];
}
function normalizePhraseList(values, fallback = []) {
const source = values === undefined || values === null ? fallback : values;
const normalized = flattenValues(source)
.flatMap((value) => {
if (typeof value === "string") {
return value.split(/\r?\n/);
}
return [value];
})
.map((value) => `${value ?? ""}`.trim())
.filter(Boolean);
return [...new Set(normalized)];
}
function normalizePositiveInteger(value, fallback) {
const parsed = Number.parseInt(`${value ?? ""}`, 10);
if (Number.isInteger(parsed) && parsed > 0) {
return parsed;
}
return fallback;
}
function normalizeNonNegativeInteger(value, fallback) {
const parsed = Number.parseInt(`${value ?? ""}`, 10);
if (Number.isInteger(parsed) && parsed >= 0) {
return parsed;
}
return fallback;
}
function sleep(ms) {
if (!Number.isFinite(ms) || ms <= 0) {
return Promise.resolve();
}
return new Promise((resolve) => setTimeout(resolve, ms));
}
function parseModelRemapMap(value) {
const map = {};
for (const rawEntry of `${value || ""}`.split(/\r?\n|[;,]/)) {
const entry = rawEntry.trim();
if (!entry) {
continue;
}
const separatorIndex = entry.indexOf("=");
if (separatorIndex <= 0) {
continue;
}
const from = entry.slice(0, separatorIndex).trim();
const to = entry.slice(separatorIndex + 1).trim();
if (!from || !to) {
continue;
}
map[from] = to;
}
return map;
}
function canHotSwapProfile(currentConfig, nextConfig) {
return (
`${currentConfig?.listen_host || ""}` === `${nextConfig?.listen_host || ""}` &&
Number.parseInt(`${currentConfig?.listen_port || ""}`, 10) === Number.parseInt(`${nextConfig?.listen_port || ""}`, 10)
);
}
function parseEnvText(content) {
const values = {};
for (const rawLine of `${content || ""}`.split(/\r?\n/)) {
const line = rawLine.trim();
if (!line || line.startsWith("#")) {
continue;
}
const separatorIndex = line.indexOf("=");
if (separatorIndex <= 0) {
continue;
}
const key = line.slice(0, separatorIndex).trim();
let value = line.slice(separatorIndex + 1).trim();
if (
(value.startsWith('"') && value.endsWith('"')) ||
(value.startsWith("'") && value.endsWith("'"))
) {
value = value.slice(1, -1);
}
values[key] = value;
}
return values;
}
function isSecretLikeKey(key) {
return /TOKEN|SECRET|PASSWORD|API_KEY|AUTH|BEARER|KEY/i.test(`${key || ""}`);
}
function redactProfileValue(key, value) {
if (!value) {
return "";
}
if (isSecretLikeKey(key) || /^sk-[A-Za-z0-9_-]+/.test(value)) {
return "[configured]";
}
return value;
}
function getProfileNameFromFile(fileName) {
if (!fileName.endsWith(".env")) {
return null;
}
const profileName = fileName.slice(0, -4);
return /^[A-Za-z0-9_.-]+$/.test(profileName) ? profileName : null;
}
function validateProfileName(profileName) {
if (!/^[A-Za-z0-9_.-]+$/.test(profileName)) {
throw new Error("profile 名称只能包含字母、数字、下划线、点和短横线");
}
}
function buildBlockedBody(pathname, reasoning, statusCode) {
return JSON.stringify({
error: {
message: `codex retry gateway blocked suspicious reasoning response on ${pathname}`,
type: "codex_retry_gateway",
code: "reasoning_guard_triggered",
reasoning_tokens: reasoning,
status_code: statusCode,
},
});
}
function buildRetryableUpstreamErrorBody(pathname, upstreamStatusCode, upstreamMessage, statusCode) {
return JSON.stringify({
error: {
message: `codex retry gateway converted retryable upstream error on ${pathname}`,
type: "codex_retry_gateway",
code: "upstream_error_retry_triggered",
upstream_status_code: upstreamStatusCode,
upstream_error_message: upstreamMessage,
status_code: statusCode,
},
});
}
function buildGatewayErrorBody(message) {
return JSON.stringify({
error: {
message,
type: "codex_retry_gateway_error",
code: "gateway_error",
},
});
}
function createMonitor() {
return {
started_at: new Date().toISOString(),
persistent_since: null,
next_log_seq: 1,
next_request_seq: 1,
log_entries: [],
request_entries: [],
total_proxy_request_count: 0,
inspected_response_count: 0,
matched_response_count: 0,
token_totals: {
input_tokens: 0,
output_tokens: 0,
total_tokens: 0,
reasoning_tokens: 0,
cached_tokens: 0,
},
observed_reasoning_counts: {},
};
}
function buildLogEntry(seq, at, message) {
return {
seq,
at,
message,
};
}
function parseLogLine(line, seq) {
const match = `${line || ""}`.match(/^(\S+)\s([\s\S]*)$/);
if (!match) {
return buildLogEntry(seq, null, line);
}
return buildLogEntry(seq, match[1], match[2]);
}
function parseJsonLine(line) {
if (!line.trim()) {
return null;
}
try {
return JSON.parse(line);
} catch {
return null;
}
}
async function readJsonlFile(filePath) {
const text = await readOptionalText(filePath);
if (!text) {
return [];
}
return text
.split(/\r?\n/)
.map(parseJsonLine)
.filter(Boolean);
}
async function appendJsonl(filePath, value) {
await mkdir(path.dirname(filePath), { recursive: true });
await fs.promises.appendFile(filePath, `${JSON.stringify(value)}\n`, "utf8");
}
function openRequestsDatabase(dbPath) {
fs.mkdirSync(path.dirname(dbPath), { recursive: true });
const db = new DatabaseSync(dbPath);
db.exec(`
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS requests (
seq INTEGER PRIMARY KEY,
request_id TEXT,
response_id TEXT,
thread_id TEXT,
started_at TEXT,
finished_at TEXT,
duration_ms INTEGER,
profile_name TEXT,
method TEXT,
path TEXT,
model TEXT,
requested_model TEXT,
forwarded_model TEXT,
request_stream INTEGER,
response_stream INTEGER,
inspected INTEGER,
matched INTEGER,
status_code INTEGER,
upstream_status_code INTEGER,
reasoning_tokens INTEGER,
input_tokens INTEGER,
output_tokens INTEGER,
total_tokens INTEGER,
cached_tokens INTEGER,
error TEXT,
upstream_origin TEXT,
upstream_path TEXT,
upstream_auth_mode TEXT,
upstream_auth_source TEXT,
payload_json TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_requests_started_at ON requests(started_at DESC);
CREATE INDEX IF NOT EXISTS idx_requests_profile_name ON requests(profile_name);
CREATE INDEX IF NOT EXISTS idx_requests_status_code ON requests(status_code);
CREATE INDEX IF NOT EXISTS idx_requests_matched ON requests(matched);
CREATE INDEX IF NOT EXISTS idx_requests_response_stream ON requests(response_stream);
`);
const requestColumns = db.prepare("PRAGMA table_info(requests)").all();
const requestColumnNames = new Set(requestColumns.map((column) => column.name));
if (!requestColumnNames.has("request_id")) {
db.exec("ALTER TABLE requests ADD COLUMN request_id TEXT");
}
if (!requestColumnNames.has("response_id")) {
db.exec("ALTER TABLE requests ADD COLUMN response_id TEXT");
}
if (!requestColumnNames.has("thread_id")) {
db.exec("ALTER TABLE requests ADD COLUMN thread_id TEXT");
}
db.exec("CREATE INDEX IF NOT EXISTS idx_requests_request_id ON requests(request_id)");
db.exec("CREATE INDEX IF NOT EXISTS idx_requests_response_id ON requests(response_id)");
db.exec("CREATE INDEX IF NOT EXISTS idx_requests_thread_id ON requests(thread_id)");
return db;
}
function boolToInt(value) {
return value ? 1 : 0;
}
function usageField(entry, key) {
const value = entry?.usage?.[key];
return Number.isInteger(value) ? value : null;
}
function buildPersistedRequestPayload(entry) {
return Object.fromEntries(
Object.entries(entry || {}).filter(([key]) => !key.startsWith("_")),
);
}
function computeRequestId(pathname, rawBody) {
const hash = createHash("sha256");
hash.update(normalizePath(pathname));
hash.update("\n");
hash.update(Buffer.isBuffer(rawBody) ? rawBody : Buffer.from(rawBody || ""));
return `req_${hash.digest("hex").slice(0, 16)}`;
}
function requestRowFromEntry(entry) {
const payload = buildPersistedRequestPayload(entry);
return {
seq: entry.seq,
request_id: entry.request_id || null,
response_id: entry.response_id || null,
thread_id: entry.thread_id || null,
started_at: entry.started_at || null,
finished_at: entry.finished_at || null,
duration_ms: Number.isInteger(entry.duration_ms) ? entry.duration_ms : null,
profile_name: entry.profile_name || null,
method: entry.method || null,
path: entry.path || null,
model: entry.model || null,
requested_model: entry.requested_model || null,
forwarded_model: entry.forwarded_model || null,
request_stream: boolToInt(Boolean(entry.request_stream)),
response_stream: boolToInt(Boolean(entry.response_stream)),
inspected: boolToInt(Boolean(entry.inspected)),
matched: boolToInt(Boolean(entry.matched)),
status_code: Number.isInteger(entry.status_code) ? entry.status_code : null,
upstream_status_code: Number.isInteger(entry.upstream_status_code) ? entry.upstream_status_code : null,
reasoning_tokens: Number.isInteger(entry.reasoning_tokens) ? entry.reasoning_tokens : usageField(entry, "reasoning_tokens"),
input_tokens: usageField(entry, "input_tokens"),
output_tokens: usageField(entry, "output_tokens"),
total_tokens: usageField(entry, "total_tokens"),
cached_tokens: usageField(entry, "cached_tokens"),
error: entry.error || null,
upstream_origin: entry.upstream?.origin || null,
upstream_path: entry.upstream?.path || null,
upstream_auth_mode: entry.upstream?.auth_mode || null,
upstream_auth_source: entry.upstream?.auth_source || null,
payload_json: JSON.stringify(payload),
};
}
function insertRequestRow(db, row) {
db.prepare(`
INSERT OR REPLACE INTO requests (
seq, request_id, response_id, thread_id, started_at, finished_at, duration_ms, profile_name, method, path, model,
requested_model, forwarded_model, request_stream, response_stream, inspected, matched,
status_code, upstream_status_code, reasoning_tokens, input_tokens, output_tokens,
total_tokens, cached_tokens, error, upstream_origin, upstream_path,
upstream_auth_mode, upstream_auth_source, payload_json
) VALUES (
@seq, @request_id, @response_id, @thread_id, @started_at, @finished_at, @duration_ms, @profile_name, @method, @path, @model,
@requested_model, @forwarded_model, @request_stream, @response_stream, @inspected, @matched,
@status_code, @upstream_status_code, @reasoning_tokens, @input_tokens, @output_tokens,
@total_tokens, @cached_tokens, @error, @upstream_origin, @upstream_path,
@upstream_auth_mode, @upstream_auth_source, @payload_json
)
`).run(row);
}
async function importRequestsJsonlToDb(db, filePath) {
const rows = await readJsonlFile(filePath);
if (rows.length === 0) {
return 0;
}
const countRow = db.prepare("SELECT COUNT(*) AS count FROM requests").get();
if ((countRow?.count || 0) >= rows.length) {
return 0;
}
db.exec("BEGIN");
try {
for (const entry of rows) {
if (!Number.isInteger(entry?.seq)) {
continue;
}
insertRequestRow(db, requestRowFromEntry(entry));
}
db.exec("COMMIT");
} catch (error) {
try {
db.exec("ROLLBACK");
} catch {
// Ignore rollback failures so the original insert error is preserved.
}
throw error;
}
return rows.length;
}
function parseRequestRowPayload(row) {
if (!row?.payload_json) {
return null;
}
try {
return JSON.parse(row.payload_json);
} catch {
return null;
}
}
function buildRequestQueryFilters({ query, filter }) {
const clauses = [];
const params = {};
const trimmedQuery = `${query || ""}`.trim().toLowerCase();
if (trimmedQuery) {
clauses.push(`
(
lower(coalesce(profile_name, '')) LIKE @query OR
lower(coalesce(method, '')) LIKE @query OR
lower(coalesce(path, '')) LIKE @query OR
lower(coalesce(request_id, '')) LIKE @query OR
lower(coalesce(response_id, '')) LIKE @query OR
lower(coalesce(thread_id, '')) LIKE @query OR
lower(coalesce(model, '')) LIKE @query OR
lower(coalesce(requested_model, '')) LIKE @query OR
lower(coalesce(forwarded_model, '')) LIKE @query OR
lower(coalesce(error, '')) LIKE @query OR
lower(coalesce(upstream_origin, '')) LIKE @query OR
lower(coalesce(upstream_path, '')) LIKE @query OR
CAST(coalesce(status_code, '') AS TEXT) LIKE @query OR
CAST(coalesce(upstream_status_code, '') AS TEXT) LIKE @query OR
CAST(coalesce(input_tokens, '') AS TEXT) LIKE @query OR
CAST(coalesce(output_tokens, '') AS TEXT) LIKE @query OR
CAST(coalesce(total_tokens, '') AS TEXT) LIKE @query OR
CAST(coalesce(cached_tokens, '') AS TEXT) LIKE @query OR
CAST(coalesce(reasoning_tokens, '') AS TEXT) LIKE @query
)
`);
params.query = `%${trimmedQuery}%`;
}
if (filter === "matched") {
clauses.push("matched = 1");
} else if (filter === "stream") {
clauses.push("response_stream = 1");
} else if (filter === "error") {
clauses.push("(error IS NOT NULL AND error != '')");
}
return {
whereSql: clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : "",
params,
};
}
async function hydrateMonitorFromDisk(monitor, paths, requestHistoryLimit, requestsDb = null) {
monitor.request_entries = [];
if (requestsDb) {
const totalsRow = requestsDb.prepare(`
SELECT
COUNT(*) AS total_proxy_request_count,
COALESCE(SUM(CASE WHEN inspected = 1 THEN 1 ELSE 0 END), 0) AS inspected_response_count,
COALESCE(SUM(CASE WHEN matched = 1 THEN 1 ELSE 0 END), 0) AS matched_response_count,
COALESCE(SUM(COALESCE(input_tokens, 0)), 0) AS input_tokens,
COALESCE(SUM(COALESCE(output_tokens, 0)), 0) AS output_tokens,
COALESCE(SUM(COALESCE(total_tokens, 0)), 0) AS total_tokens,
COALESCE(SUM(COALESCE(reasoning_tokens, 0)), 0) AS reasoning_tokens,
COALESCE(SUM(COALESCE(cached_tokens, 0)), 0) AS cached_tokens,
MIN(NULLIF(started_at, '')) AS persistent_since,
MAX(seq) AS max_seq
FROM requests
`).get();
monitor.total_proxy_request_count = totalsRow?.total_proxy_request_count || 0;
monitor.inspected_response_count = totalsRow?.inspected_response_count || 0;
monitor.matched_response_count = totalsRow?.matched_response_count || 0;
monitor.token_totals = {
input_tokens: totalsRow?.input_tokens || 0,
output_tokens: totalsRow?.output_tokens || 0,
total_tokens: totalsRow?.total_tokens || 0,
reasoning_tokens: totalsRow?.reasoning_tokens || 0,
cached_tokens: totalsRow?.cached_tokens || 0,
};
monitor.persistent_since = totalsRow?.persistent_since || null;
monitor.next_request_seq = Number.isInteger(totalsRow?.max_seq) ? totalsRow.max_seq + 1 : 1;
monitor.observed_reasoning_counts = {};
const reasoningRows = requestsDb.prepare(`
SELECT reasoning_tokens, COUNT(*) AS count
FROM requests
WHERE reasoning_tokens IS NOT NULL
GROUP BY reasoning_tokens
`).all();
for (const row of reasoningRows) {
if (!Number.isInteger(row?.reasoning_tokens)) {
continue;
}
monitor.observed_reasoning_counts[`${row.reasoning_tokens}`] = row.count || 0;
}
return;
}
const requestEntries = await readJsonlFile(paths.requestsPath);
monitor.next_request_seq = requestEntries.reduce((maxSeq, entry) => {
return Math.max(maxSeq, Number.isInteger(entry.seq) ? entry.seq + 1 : maxSeq);
}, 1);
for (const entry of requestEntries) {
monitor.total_proxy_request_count += 1;
if (entry.inspected) {
recordInspectedResponse(monitor, entry.reasoning_tokens ?? entry.usage?.reasoning_tokens ?? null, Boolean(entry.matched));
}
addTokenTotals(monitor, entry.usage);
}
const firstStartedAt = requestEntries
.map((entry) => `${entry?.started_at || ""}`.trim())
.filter(Boolean)
.sort()[0];
monitor.persistent_since = firstStartedAt || null;
}
function createMonitorRecorder(monitor) {
return (message) => {
const entry = {
seq: monitor.next_log_seq,
at: new Date().toISOString(),
message,
};
monitor.next_log_seq += 1;
monitor.log_entries.push(entry);
return entry;
};
}
function createLogger(logPath, recordEntry) {
if (!logPath) {
return (message) => {
const entry = recordEntry ? recordEntry(message) : { at: new Date().toISOString(), message };
process.stdout.write(`${entry.at} ${entry.message}\n`);
};
}
const stream = fs.createWriteStream(logPath, { flags: "a" });
return (message) => {
const entry = recordEntry ? recordEntry(message) : { at: new Date().toISOString(), message };
const line = `${entry.at} ${entry.message}\n`;
stream.write(line);
process.stdout.write(line);
};
}
function incrementReasoningCount(counter, reasoning) {
if (!Number.isInteger(reasoning)) {
return;
}
const key = `${reasoning}`;
counter[key] = (counter[key] || 0) + 1;
}
function recordInspectedResponse(monitor, reasoning, matched) {
monitor.inspected_response_count += 1;
incrementReasoningCount(monitor.observed_reasoning_counts, reasoning);
if (matched) {
monitor.matched_response_count += 1;
}
}
function addTokenTotals(monitor, usage) {
if (!usage) {
return;
}
for (const key of ["input_tokens", "output_tokens", "total_tokens", "reasoning_tokens", "cached_tokens"]) {
const value = usage[key];
if (Number.isInteger(value)) {
monitor.token_totals[key] += value;
}
}
}
function upsertRequestEntry(runtime, entry, { persistJsonl = false, includeUsage = false } = {}) {
if (includeUsage) {
addTokenTotals(runtime.monitor, entry.usage);
}
if (runtime.requestsDb) {
insertRequestRow(runtime.requestsDb, requestRowFromEntry(entry));
}
if (persistJsonl) {
appendJsonl(runtime.paths.requestsPath, buildPersistedRequestPayload(entry)).catch((error) => {
runtime.logger?.(`[requests] failed to persist request seq=${entry.seq}: ${error?.message || error}`);
});
}
}
function recordRequestEntry(runtime, entry, limit = DEFAULT_CONFIG.request_history_limit) {
upsertRequestEntry(runtime, entry, {
persistJsonl: true,
includeUsage: true,
});
}
function markAndPersistFirstResponse(runtime, entry, at = new Date()) {
if (!markRequestFirstResponse(entry, at)) {
return false;
}
upsertRequestEntry(runtime, entry);
return true;
}
function summarizeProfileAuthSource(env) {
const mode = env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE || "passthrough";
if (mode === "auth_json") {
return env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_PATH ? "auth.json path configured" : "~/.codex/auth.json";
}
if (mode === "manual_bearer") {
const secretPath = env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_FILE || "";
if (!secretPath) {
return "system secret file missing";
}
return fs.existsSync(secretPath) ? "system secret file configured" : "system secret file missing";
}
if (mode === "fixed_bearer") {
if (env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_FILE) {
return "token file configured";
}
if (env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_ENV) {
return `env:${env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_ENV}`;
}
}
return "passthrough";
}
function buildProfileFormModel(env) {
const manualSecretFile = env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_FILE || "";
return {
listen_host: env.CODEX_RETRY_GATEWAY_LISTEN_HOST || "",
listen_port: env.CODEX_RETRY_GATEWAY_LISTEN_PORT || "",
upstream_base_url: env.CODEX_RETRY_GATEWAY_UPSTREAM_BASE_URL || "",
auth_mode: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE || "passthrough",
auth_env: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_ENV || "",
auth_file: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE === "manual_bearer"
? ""
: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_FILE || "",
manual_secret_file: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE === "manual_bearer"
? manualSecretFile
: "",
manual_secret_configured: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE === "manual_bearer"
&& Boolean(manualSecretFile)
&& fs.existsSync(manualSecretFile),
auth_json_path: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_PATH || "",
auth_json_key: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_KEY || "",
request_history_limit: env.CODEX_RETRY_GATEWAY_REQUEST_HISTORY_LIMIT || `${DEFAULT_CONFIG.request_history_limit}`,
model_remap: env.CODEX_RETRY_GATEWAY_MODEL_REMAP || "",
reasoning_equals: env.CODEX_RETRY_GATEWAY_REASONING_EQUALS || "",
retryable_status_codes: env.CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES || "",
retryable_error_messages: normalizePhraseList(
env.CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES || DEFAULT_CONFIG.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
),
upstream_fetch_retry_attempts:
env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS || `${DEFAULT_CONFIG.upstream_fetch_retry_attempts}`,
upstream_fetch_retry_backoff_ms:
env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS || `${DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms}`,
endpoints: normalizeStringList(env.CODEX_RETRY_GATEWAY_ENDPOINTS || DEFAULT_CONFIG.endpoints, DEFAULT_CONFIG.endpoints),
};
}
function buildConfigFromProfileEnv(profileName, env) {
const config = {
...DEFAULT_CONFIG,
profile_name: profileName,
listen_host: env.CODEX_RETRY_GATEWAY_LISTEN_HOST || DEFAULT_CONFIG.listen_host,
listen_port: env.CODEX_RETRY_GATEWAY_LISTEN_PORT
? Number.parseInt(`${env.CODEX_RETRY_GATEWAY_LISTEN_PORT}`, 10)
: DEFAULT_CONFIG.listen_port,
upstream_base_url: env.CODEX_RETRY_GATEWAY_UPSTREAM_BASE_URL || "",
upstream_auth_mode: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE || DEFAULT_CONFIG.upstream_auth_mode,
upstream_auth_env: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_ENV || DEFAULT_CONFIG.upstream_auth_env,
upstream_auth_file: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_FILE || "",
upstream_auth_json_path: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_PATH || "",
upstream_auth_json_key: env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_KEY || DEFAULT_CONFIG.upstream_auth_json_key,
request_body_limit_bytes: env.CODEX_RETRY_GATEWAY_REQUEST_BODY_LIMIT_BYTES
? Number.parseInt(`${env.CODEX_RETRY_GATEWAY_REQUEST_BODY_LIMIT_BYTES}`, 10)
: DEFAULT_CONFIG.request_body_limit_bytes,
request_history_limit: env.CODEX_RETRY_GATEWAY_REQUEST_HISTORY_LIMIT
? Number.parseInt(`${env.CODEX_RETRY_GATEWAY_REQUEST_HISTORY_LIMIT}`, 10)
: DEFAULT_CONFIG.request_history_limit,
model_remap: env.CODEX_RETRY_GATEWAY_MODEL_REMAP || "",
endpoints: normalizeStringList(env.CODEX_RETRY_GATEWAY_ENDPOINTS || DEFAULT_CONFIG.endpoints, DEFAULT_CONFIG.endpoints).map(normalizePath),
reasoning_equals: normalizeIntegerList(env.CODEX_RETRY_GATEWAY_REASONING_EQUALS || DEFAULT_CONFIG.reasoning_equals, DEFAULT_CONFIG.reasoning_equals),
retryable_status_codes: normalizeIntegerList(
env.CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES || DEFAULT_CONFIG.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
),
retryable_error_messages: normalizePhraseList(
env.CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES || DEFAULT_CONFIG.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
),
management_access_key: normalizeManagementAccessKey(
env.CODEX_RETRY_GATEWAY_MANAGEMENT_ACCESS_KEY || DEFAULT_CONFIG.management_access_key,
),
upstream_fetch_retry_attempts: env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS
? normalizePositiveInteger(
env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS,
DEFAULT_CONFIG.upstream_fetch_retry_attempts,
)
: DEFAULT_CONFIG.upstream_fetch_retry_attempts,
upstream_fetch_retry_backoff_ms: env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS
? normalizeNonNegativeInteger(
env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS,
DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms,
)
: DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms,
non_stream_status_code: env.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE
? Number.parseInt(`${env.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE}`, 10)
: DEFAULT_CONFIG.non_stream_status_code,
stream_action: env.CODEX_RETRY_GATEWAY_STREAM_ACTION || DEFAULT_CONFIG.stream_action,
log_match: env.CODEX_RETRY_GATEWAY_LOG_MATCH === undefined
? DEFAULT_CONFIG.log_match
: ["1", "true", "yes", "on"].includes(`${env.CODEX_RETRY_GATEWAY_LOG_MATCH}`.trim().toLowerCase()),
health_path: env.CODEX_RETRY_GATEWAY_HEALTH_PATH || DEFAULT_CONFIG.health_path,
};
config.model_remap_map = parseModelRemapMap(config.model_remap);
return config;
}
function assertNoInlineProfileSecret(payload) {
const suspectFields = [
"auth_env",
"auth_file",
"auth_json_path",
"auth_json_key",
"upstream_base_url",
];
for (const field of suspectFields) {
const value = `${payload?.[field] || ""}`.trim();
if (/^sk-[A-Za-z0-9_-]+/.test(value) || /Bearer\s+sk-[A-Za-z0-9_-]+/i.test(value)) {
throw new Error("profile 不保存明文 sk 密钥;请改用 env/file/auth.json 引用");
}
}
}
function defaultManualSecretPath(profileName) {
const homeDir = process.env.HOME || "";
return path.join(homeDir, ".codex-retry-gateway", "secrets", `${profileName}.token`);
}
async function writeManualSecret(profileName, secretValue) {
const text = `${secretValue || ""}`.trim();
if (!text) {
return null;
}
const secretPath = defaultManualSecretPath(profileName);
await mkdir(path.dirname(secretPath), { recursive: true, mode: 0o700 });
await writeFile(secretPath, `${text}\n`, { encoding: "utf8", mode: 0o600 });
await chmod(path.dirname(secretPath), 0o700).catch(() => {});
await chmod(secretPath, 0o600).catch(() => {});
return secretPath;
}
function serializeEnvValue(value) {
const text = `${value ?? ""}`;
if (/^[A-Za-z0-9_./:@?&=,+-]*$/.test(text)) {
return text;
}
return JSON.stringify(text);
}
async function buildProfileEnvText(payload) {
assertNoInlineProfileSecret(payload);
const name = `${payload?.name || ""}`.trim();
validateProfileName(name);
const listenPort = Number.parseInt(`${payload.listen_port || DEFAULT_CONFIG.listen_port}`, 10);
if (!Number.isInteger(listenPort) || listenPort < 1 || listenPort > 65535) {
throw new Error("监听端口必须是 1-65535 的整数");
}
const upstreamBaseUrl = `${payload.upstream_base_url || ""}`.trim();
if (!upstreamBaseUrl) {
throw new Error("上游 Base URL 不能为空");
}
try {
new URL(upstreamBaseUrl);
} catch {
throw new Error("上游 Base URL 必须是合法 URL");
}
const authMode = normalizeAuthMode(payload.auth_mode);
const reasoningEquals = normalizeIntegerList(payload.reasoning_equals, DEFAULT_CONFIG.reasoning_equals);
if (reasoningEquals.length === 0) {
throw new Error("reasoning_equals 不能为空");
}
const endpoints = normalizeStringList(payload.endpoints, DEFAULT_CONFIG.endpoints).map(normalizePath);
if (endpoints.length === 0) {
throw new Error("endpoints 不能为空");
}
const requestHistoryLimit = Number.parseInt(
`${payload.request_history_limit || DEFAULT_CONFIG.request_history_limit}`,
10,
);
if (!Number.isInteger(requestHistoryLimit) || requestHistoryLimit < 0) {
throw new Error("History Limit 必须是 0 或正整数;0 表示不裁剪");
}
const retryableStatusCodes = normalizeIntegerList(
payload.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
);
const retryableErrorMessages = normalizePhraseList(
payload.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
const upstreamFetchRetryAttempts = normalizePositiveInteger(
payload.upstream_fetch_retry_attempts,
DEFAULT_CONFIG.upstream_fetch_retry_attempts,
);
const upstreamFetchRetryBackoffMs = normalizeNonNegativeInteger(
payload.upstream_fetch_retry_backoff_ms,
DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms,
);
if (retryableStatusCodes.length === 0) {
throw new Error("retryable_status_codes 不能为空");
}
if (retryableErrorMessages.length === 0) {
throw new Error("retryable_error_messages 不能为空");
}
if (upstreamFetchRetryAttempts < 1) {
throw new Error("upstream_fetch_retry_attempts 必须是正整数");
}
if (upstreamFetchRetryBackoffMs < 0) {
throw new Error("upstream_fetch_retry_backoff_ms 不能为负数");
}
const envPairs = [
["CODEX_RETRY_GATEWAY_LISTEN_HOST", `${payload.listen_host || DEFAULT_CONFIG.listen_host}`.trim()],
["CODEX_RETRY_GATEWAY_LISTEN_PORT", `${listenPort}`],
["CODEX_RETRY_GATEWAY_UPSTREAM_BASE_URL", upstreamBaseUrl],
["CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE", authMode],
["CODEX_RETRY_GATEWAY_REASONING_EQUALS", reasoningEquals.join(",")],
["CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES", retryableStatusCodes.join(",")],
["CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES", retryableErrorMessages.join("\n")],
["CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS", `${upstreamFetchRetryAttempts}`],
["CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS", `${upstreamFetchRetryBackoffMs}`],
["CODEX_RETRY_GATEWAY_REQUEST_HISTORY_LIMIT", `${requestHistoryLimit}`],
["CODEX_RETRY_GATEWAY_ENDPOINTS", endpoints.join(",")],
];
const managementAccessKey = normalizeManagementAccessKey(payload.management_access_key);
if (managementAccessKey) {
envPairs.push(["CODEX_RETRY_GATEWAY_MANAGEMENT_ACCESS_KEY", managementAccessKey]);
}
const modelRemap = `${payload.model_remap || ""}`.trim();
if (modelRemap) {
envPairs.push(["CODEX_RETRY_GATEWAY_MODEL_REMAP", modelRemap]);
}
if (authMode === "manual_bearer") {
const secretPath =
(await writeManualSecret(name, payload.manual_secret)) ||
`${payload.manual_secret_file || ""}`.trim() ||
defaultManualSecretPath(name);
if (!fs.existsSync(secretPath)) {
throw new Error("manual_bearer 需要手动填入一次 token/password 后才能保存");
}
envPairs.push(["CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_FILE", secretPath]);
} else if (authMode === "fixed_bearer") {
envPairs.push([
"CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_ENV",
`${payload.auth_env || DEFAULT_CONFIG.upstream_auth_env}`.trim(),
]);
if (`${payload.auth_file || ""}`.trim()) {
envPairs.push(["CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_FILE", `${payload.auth_file}`.trim()]);
}
} else if (authMode === "auth_json") {
if (`${payload.auth_json_path || ""}`.trim()) {
envPairs.push(["CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_PATH", `${payload.auth_json_path}`.trim()]);
}
envPairs.push([
"CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_KEY",
`${payload.auth_json_key || DEFAULT_CONFIG.upstream_auth_json_key}`.trim(),
]);
}
const lines = [
"# Managed by codex-retry-gateway UI.",
"# Do not put raw sk-* secrets here; use env/file/auth.json references.",
...envPairs.map(([key, value]) => `${key}=${serializeEnvValue(value)}`),
"",
];
return {
name,
content: lines.join("\n"),
};
}
async function writeProfile(runtime, payload) {
const { name, content } = await buildProfileEnvText(payload);
await mkdir(runtime.paths.profilesDir, { recursive: true });
const profilePath = path.join(runtime.paths.profilesDir, `${name}.env`);
await writeFile(profilePath, content, { encoding: "utf8", mode: 0o600 });
return {
name,
file_path: profilePath,
};
}
function buildMetricsSnapshot(monitor) {
const reasoning516Count = monitor.observed_reasoning_counts["516"] || 0;
const inspectedResponseCount = monitor.inspected_response_count;
const reasoningEntries = Object.entries(monitor.observed_reasoning_counts).sort((left, right) => {
const countDelta = Number(right[1] || 0) - Number(left[1] || 0);
if (countDelta !== 0) {
return countDelta;
}
return Number(left[0] || 0) - Number(right[0] || 0);
});
const visibleReasoningEntries = reasoningEntries.slice(0, STATUS_REASONING_COUNT_LIMIT);
return {
started_at: monitor.started_at,
persistent_since: monitor.persistent_since,
total_proxy_request_count: monitor.total_proxy_request_count,
inspected_response_count: inspectedResponseCount,
matched_response_count: monitor.matched_response_count,
reasoning_516_count: reasoning516Count,
reasoning_516_ratio:
inspectedResponseCount === 0 ? 0 : reasoning516Count / inspectedResponseCount,
token_totals: { ...monitor.token_totals },
observed_reasoning_counts: Object.fromEntries(visibleReasoningEntries),
observed_reasoning_counts_total_keys: reasoningEntries.length,
observed_reasoning_counts_omitted: Math.max(
0,
reasoningEntries.length - visibleReasoningEntries.length,
),
};
}
function buildLogsSnapshot(monitor, sinceSeq = null) {
const entries = Number.isInteger(sinceSeq)
? monitor.log_entries.filter((entry) => entry.seq > sinceSeq)
: monitor.log_entries;
return {
total_entries: monitor.log_entries.length,
latest_seq: monitor.next_log_seq - 1,
entries,
};
}
async function buildPersistentLogsSnapshot(runtime, sinceSeq = null, limit = 500) {
const safeLimit = Number.isInteger(limit) && limit > 0 ? Math.min(limit, 1000) : 500;
const text = runtime.logPath ? await readOptionalText(runtime.logPath) : null;
if (!text) {
return buildLogsSnapshot(runtime.monitor, sinceSeq);
}
const allEntries = text
.split(/\r?\n/)
.filter(Boolean)
.map((line, index) => parseLogLine(line, index + 1));
const entries = Number.isInteger(sinceSeq)
? allEntries.filter((entry) => entry.seq > sinceSeq)
: allEntries.slice(-safeLimit);
return {
total_entries: allEntries.length,
latest_seq: allEntries.length,
entries,
};
}
function buildRequestsSnapshot(monitor, limit = 50) {
const safeLimit = Number.isInteger(limit) && limit > 0 ? Math.min(limit, 500) : 50;
const entries = monitor.request_entries.slice(-safeLimit).reverse();
return {
total_entries: monitor.request_entries.length,
latest_seq: monitor.next_request_seq - 1,
entries,
};
}
async function buildPersistentRequestsSnapshot(runtime, { limit = 50, offset = 0, query = "", filter = "all" } = {}) {
const safeLimit = Number.isInteger(limit) && limit > 0 ? Math.min(limit, 500) : 50;
const safeOffset = Number.isInteger(offset) && offset > 0 ? offset : 0;
if (!runtime.requestsDb) {
const entries = await readJsonlFile(runtime.paths.requestsPath);
return {
total_entries: entries.length,
latest_seq: entries.reduce((maxSeq, entry) => {
return Math.max(maxSeq, Number.isInteger(entry.seq) ? entry.seq : maxSeq);
}, 0),
entries: entries.slice(-safeLimit).reverse(),
};
}
const { whereSql, params } = buildRequestQueryFilters({ query, filter });
const totalRow = runtime.requestsDb.prepare(`SELECT COUNT(*) AS count FROM requests ${whereSql}`).get(params);
const latestRow = runtime.requestsDb.prepare("SELECT MAX(seq) AS latest_seq FROM requests").get();
const rows = runtime.requestsDb.prepare(`
SELECT payload_json
FROM requests
${whereSql}
ORDER BY seq DESC
LIMIT @limit OFFSET @offset
`).all({
...params,
limit: safeLimit,
offset: safeOffset,
});
return {
total_entries: totalRow?.count || 0,
latest_seq: latestRow?.latest_seq || 0,
entries: rows.map(parseRequestRowPayload).filter(Boolean),
};
}
function buildRequestEntry({ seq, startedAt, startedMs, req, pathname, requestJson, profileName }) {
return {
seq,
request_id: null,
response_id: null,
thread_id: firstNonEmptyString(
extractRequestThreadId(requestJson),
extractHeaderThreadId(req?.headers),
),
lifecycle_state: "sent",
started_at: startedAt.toISOString(),
first_response_at: null,
first_response_delay_ms: null,
last_activity_at: null,
finished_at: null,
duration_ms: null,
profile_name: profileName || "default",
method: req.method,
path: pathname,
request_body_bytes: null,
response_bytes_received: 0,
model: requestJson?.model || null,
requested_model: requestJson?.model || null,
forwarded_model: requestJson?.model || null,
request_stream: Boolean(requestJson?.stream),
response_stream: false,
stream_chunk_count: 0,
usage_last_updated_at: null,
upstream_attempt_count: 0,
inspected: false,
matched: false,
status_code: null,
upstream_status_code: null,
upstream: null,
reasoning_tokens: null,
usage: null,
error: null,
_started_ms: startedMs,
};
}
function markRequestFirstResponse(entry, at = new Date()) {
if (entry.first_response_at) {
return false;
}
const firstAt = at instanceof Date ? at : new Date(at);
const firstMs = firstAt.getTime();
entry.first_response_at = firstAt.toISOString();
entry.first_response_delay_ms = Number.isFinite(entry._started_ms)
? Math.max(0, firstMs - entry._started_ms)
: null;
entry.last_activity_at = firstAt.toISOString();
entry.lifecycle_state = "receive_first";
return true;
}
const STREAM_PROGRESS_PERSIST_INTERVAL_MS = 1000;
function updateStreamingProgress(entry, { chunkBytes = 0, usage = null, reasoning = null, at = new Date() } = {}) {
const observedAt = at instanceof Date ? at : new Date(at);
const observedAtIso = observedAt.toISOString();
entry.last_activity_at = observedAtIso;
entry.lifecycle_state = "streaming";
if (chunkBytes > 0) {
entry.stream_chunk_count = (entry.stream_chunk_count || 0) + 1;
entry.response_bytes_received = (entry.response_bytes_received || 0) + chunkBytes;
}
if (usage) {
entry.usage = mergeUsageSnapshots(entry.usage, usage);
entry.usage_last_updated_at = observedAtIso;
}
if (Number.isInteger(reasoning)) {
entry.reasoning_tokens = reasoning;
}
}
function persistStreamingProgress(runtime, entry, { force = false, usageUpdated = false } = {}, at = new Date()) {
const observedAt = at instanceof Date ? at : new Date(at);
const nowMs = observedAt.getTime();
const lastPersistedMs = Number.isFinite(entry._last_stream_persisted_ms)
? entry._last_stream_persisted_ms
: 0;
if (!force && !usageUpdated && nowMs - lastPersistedMs < STREAM_PROGRESS_PERSIST_INTERVAL_MS) {
return false;
}
entry._last_stream_persisted_ms = nowMs;
upsertRequestEntry(runtime, entry);
return true;
}
function finalizeRequestEntry(entry, result = {}) {
const finishedAt = new Date();
const startedMs = entry._started_ms;
delete entry._started_ms;
delete entry._last_stream_persisted_ms;
return {
...entry,
...result,
lifecycle_state: "finish",
finished_at: finishedAt.toISOString(),
duration_ms: Number.isFinite(startedMs) ? Math.max(0, Date.now() - startedMs) : null,
};
}
async function loadConfig(configPath) {
const content = await readFile(configPath, "utf8");
const loaded = JSON.parse(content);
const config = { ...DEFAULT_CONFIG, ...loaded };
config.model_remap_map = parseModelRemapMap(config.model_remap);
config.endpoints = normalizeStringList(config.endpoints, DEFAULT_CONFIG.endpoints).map(normalizePath);
config.reasoning_equals = normalizeIntegerList(
config.reasoning_equals,
DEFAULT_CONFIG.reasoning_equals,
);
config.retryable_status_codes = normalizeIntegerList(
config.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
);
config.retryable_error_messages = normalizePhraseList(
config.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
config.management_access_key = normalizeManagementAccessKey(config.management_access_key);
config.upstream_fetch_retry_attempts = normalizePositiveInteger(
config.upstream_fetch_retry_attempts,
DEFAULT_CONFIG.upstream_fetch_retry_attempts,
);
config.upstream_fetch_retry_backoff_ms = normalizeNonNegativeInteger(
config.upstream_fetch_retry_backoff_ms,
DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms,
);
if (!config.upstream_base_url) {
throw new Error("配置缺少 upstream_base_url");
}
return config;
}
function buildRuntimePaths(configPath, logPath) {
const configDirectory = path.dirname(configPath);
const stateRoot = path.dirname(configDirectory);
const homeDir = process.env.HOME || "";
return {
stateRoot,
statePath: path.join(stateRoot, "state.json"),
pidPath: path.join(stateRoot, "gateway.pid"),
profilesDir: path.join(homeDir, ".config", "codex-retry-gateway", "profiles"),
configPath,
logPath,
requestsPath: path.join(stateRoot, "logs", "requests.jsonl"),
requestsDbPath: path.join(stateRoot, "logs", "requests.sqlite"),
};
}
async function readOptionalJson(jsonPath) {
try {
const content = await readFile(jsonPath, "utf8");
return JSON.parse(content);
} catch {
return null;
}
}
async function readOptionalText(textPath) {
try {
return await readFile(textPath, "utf8");
} catch {
return null;
}
}
function normalizeAuthMode(value) {
const mode = `${value || "passthrough"}`.trim().toLowerCase();
if (["passthrough", "fixed_bearer", "manual_bearer", "auth_json"].includes(mode)) {
return mode;
}
return "passthrough";
}
function sanitizeConfigForStatus(config) {
const {
management_access_key,
upstream_auth_file,
upstream_auth_json_path,
upstream_auth_env,
upstream_auth_json_key,
model_remap_map,
...rest
} = config;
return {
...rest,
management_access_key_configured: Boolean(management_access_key),
upstream_auth_env: upstream_auth_env || null,
upstream_auth_file: upstream_auth_file ? "[configured]" : "",
upstream_auth_json_path: upstream_auth_json_path ? "[configured]" : "",
upstream_auth_json_key: upstream_auth_json_key || null,
};
}
function remapRequestModel(config, requestJson) {
if (!requestJson || typeof requestJson !== "object") {
return { requestJson, remapped: false, forwardedModel: null };
}
const requestedModel = `${requestJson.model || ""}`.trim();
if (!requestedModel) {
return { requestJson, remapped: false, forwardedModel: null };
}
const forwardedModel = config.model_remap_map?.[requestedModel];
if (!forwardedModel || forwardedModel === requestedModel) {
return { requestJson, remapped: false, forwardedModel: requestedModel };
}
return {
requestJson: {
...requestJson,
model: forwardedModel,
},
remapped: true,
forwardedModel,
};
}
async function resolveUpstreamAuth(config) {
const mode = normalizeAuthMode(config.upstream_auth_mode);
if (mode === "passthrough") {
return { mode, authorization: null, source: "passthrough" };
}
let token = "";
let source = "missing";
if (mode === "fixed_bearer") {
const envName = config.upstream_auth_env || DEFAULT_CONFIG.upstream_auth_env;
token = envName ? `${process.env[envName] || ""}`.trim() : "";
source = token ? "env" : "missing";
if (!token && config.upstream_auth_file) {
token = `${(await readOptionalText(config.upstream_auth_file)) || ""}`.trim();
source = token ? "file" : "file_empty";
}
} else if (mode === "manual_bearer") {
if (config.upstream_auth_file) {
token = `${(await readOptionalText(config.upstream_auth_file)) || ""}`.trim();
source = token ? "manual_file" : "manual_file_empty";
}
} else if (mode === "auth_json") {
const authPath = config.upstream_auth_json_path || path.join(process.env.HOME || "", ".codex", "auth.json");
const key = config.upstream_auth_json_key || DEFAULT_CONFIG.upstream_auth_json_key;
const authJson = await readOptionalJson(authPath);
token = `${authJson?.[key] || ""}`.trim();
source = token ? "auth_json" : "auth_json_missing";
}
if (!token) {
throw new Error(`upstream_auth_mode=${mode} requires a configured token source`);
}
const authorization = token.toLowerCase().startsWith("bearer ") ? token : `Bearer ${token}`;
return { mode, authorization, source };
}
async function writeConfig(configPath, config) {
await writeFile(configPath, `${JSON.stringify(config, null, 2)}\n`, "utf8");
}
async function updateRuntimeState(runtime, updates) {
const current = await readOptionalJson(runtime.paths.statePath);
if (!current) {
return;
}
await writeFile(
runtime.paths.statePath,
`${JSON.stringify({ ...current, ...updates }, null, 2)}\n`,
"utf8",
);
}
async function listProfiles(runtime) {
let files = [];
try {
files = await readdir(runtime.paths.profilesDir, { withFileTypes: true });
} catch {
files = [];
}
const activeProfile = runtime.config.profile_name || "default";
const profiles = [];
for (const file of files) {
if (!file.isFile()) {
continue;
}
const name = getProfileNameFromFile(file.name);
if (!name) {
continue;
}
const filePath = path.join(runtime.paths.profilesDir, file.name);
const env = parseEnvText((await readOptionalText(filePath)) || "");
const form = buildProfileFormModel(env);
profiles.push({
name,
active: name === activeProfile,
file_path: filePath,
summary: {
listen_host: form.listen_host,
listen_port: form.listen_port,
upstream_base_url: form.upstream_base_url,
auth_mode: form.auth_mode,
auth_env: redactProfileValue(
"CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_ENV",
form.auth_env,
),
auth_file: form.auth_file ? "[configured]" : "",
auth_json_path: form.auth_json_path ? "[configured]" : "",
auth_json_key: redactProfileValue(
"CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_JSON_KEY",
form.auth_json_key,
),
request_history_limit: form.request_history_limit,
model_remap: form.model_remap || "",
auth_source: summarizeProfileAuthSource(env),
reasoning_equals: form.reasoning_equals,
},
form,
});
}
profiles.sort((left, right) => left.name.localeCompare(right.name));
return profiles;
}
function runDetached(command, args) {
const child = spawn(command, args, {
detached: true,
stdio: "ignore",
windowsHide: true,
});
child.unref();
return child.pid;
}
async function triggerProfileSwitch(runtime, profileName) {
if (!/^[A-Za-z0-9_.-]+$/.test(profileName)) {
throw new Error("profile 名称只能包含字母、数字、下划线、点和短横线");
}
const profilePath = path.join(runtime.paths.profilesDir, `${profileName}.env`);
if (!fs.existsSync(profilePath)) {
throw new Error(`profile 不存在: ${profileName}`);
}
const currentUnit = `codex-retry-gateway@${runtime.config.profile_name || "default"}.service`;
const targetUnit = `codex-retry-gateway@${profileName}.service`;
const switchScript = [
"set -e",
`systemctl --user disable --now ${currentUnit}`,
`systemctl --user enable --now ${targetUnit}`,
].join("\n");
const unitName = `codex-retry-gateway-switch-${Date.now()}`;
const pid = runDetached("systemd-run", [
"--user",
"--collect",
`--unit=${unitName}`,
"--on-active=1s",
"/usr/bin/env",
"bash",
"-lc",
switchScript,
]);
return {
profile: profileName,
current_unit: currentUnit,
target_unit: targetUnit,
switch_unit: `${unitName}.service`,
pid,
};
}
async function applyProfileConfig(runtime, profileName) {
const { profilePath, config } = await loadProfileConfigForProbe(runtime, profileName);
if (!canHotSwapProfile(runtime.config, config)) {
throw new Error("该 profile 的监听地址或端口与当前实例不同,暂不支持无重启热切换");
}
runtime.config = {
...config,
model_remap_map: parseModelRemapMap(config.model_remap),
};
await writeConfig(runtime.configPath, runtime.config);
await updateRuntimeState(runtime, {
profile_name: runtime.config.profile_name || "default",
profile_env_path: profilePath,
gateway_base_url: `http://${runtime.config.listen_host}:${runtime.config.listen_port}`,
last_started_at: new Date().toISOString(),
});
runtime.logger(
`[profile] hot-swapped profile=${runtime.config.profile_name || "default"} auth=${normalizeAuthMode(runtime.config.upstream_auth_mode)} upstream=${runtime.config.upstream_base_url}`,
);
return {
profile: runtime.config.profile_name || "default",
profile_env_path: profilePath,
hot_swapped: true,
listen: `${runtime.config.listen_host}:${runtime.config.listen_port}`,
upstream_base_url: runtime.config.upstream_base_url,
};
}
async function loadProfileConfigForProbe(runtime, profileName) {
validateProfileName(profileName);
const profilePath = path.join(runtime.paths.profilesDir, `${profileName}.env`);
if (!fs.existsSync(profilePath)) {
throw new Error(`profile 不存在: ${profileName}`);
}
const env = parseEnvText((await readOptionalText(profilePath)) || "");
const config = buildConfigFromProfileEnv(profileName, env);
if (!config.upstream_base_url) {
throw new Error(`profile ${profileName} 缺少 upstream_base_url`);
}
return { profilePath, env, config };
}
async function deleteProfile(runtime, profileName) {
validateProfileName(profileName);
const activeProfile = runtime.config.profile_name || "default";
if (profileName === activeProfile) {
throw new Error("不能删除当前正在运行的 profile;请先切换到其他 profile");
}
const profilePath = path.join(runtime.paths.profilesDir, `${profileName}.env`);
if (!fs.existsSync(profilePath)) {
throw new Error(`profile 不存在: ${profileName}`);
}
await rm(profilePath, { force: true });
return {
name: profileName,
file_path: profilePath,
};
}
async function readProbeBodySummary(response, maxChars = 400) {
const contentType = response.headers.get("content-type") || "";
const text = await response.text();
let summary = text.slice(0, maxChars);
if (contentType.includes("application/json")) {
try {
const payload = JSON.parse(text);
if (Array.isArray(payload?.data)) {
summary = JSON.stringify(payload.data.slice(0, 8).map((item) => item.id ?? item.display_name ?? item), null, 2);
} else if (payload?.error) {
summary = JSON.stringify(payload.error);
} else {
summary = text.slice(0, maxChars);
}
} catch {
summary = text.slice(0, maxChars);
}
}
return {
content_type: contentType,
body_preview: summary,
};
}
async function probeProfile(runtime, payload) {
const profileName = `${payload?.profile || ""}`.trim();
if (!profileName) {
throw new Error("缺少 profile");
}
const { config } = await loadProfileConfigForProbe(runtime, profileName);
const upstreamAuth = await resolveUpstreamAuth(config);
const result = {
profile: profileName,
upstream_base_url: config.upstream_base_url,
auth_mode: upstreamAuth.mode,
auth_source: upstreamAuth.source,
authorization_configured: Boolean(upstreamAuth.authorization),
model_remap: config.model_remap || "",
probes: [],
};
const modelsUrl = buildUpstreamUrl(config.upstream_base_url, new URL("http://local/v1/models"));
const { response: modelsResponse } = await fetchUpstreamWithRetry(modelsUrl, {
method: "GET",
headers: cloneHeadersForUpstream({}, upstreamAuth),
}, config, runtime.logger, { method: "GET", pathname: "/v1/models" });
result.probes.push({
kind: "models",
target: "/v1/models",
status: modelsResponse.status,
...await readProbeBodySummary(modelsResponse, 600),
});
const requestedModel = `${payload?.model || ""}`.trim();
if (requestedModel) {
const { requestJson, forwardedModel } = remapRequestModel(config, {
model: requestedModel,
input: `${payload?.input || "ping"}`,
max_output_tokens: 1,
stream: false,
});
const responsesUrl = buildUpstreamUrl(config.upstream_base_url, new URL("http://local/v1/responses"));
const { response: responseProbe } = await fetchUpstreamWithRetry(responsesUrl, {
method: "POST",
headers: cloneHeadersForUpstream({ "content-type": "application/json" }, upstreamAuth),
body: JSON.stringify(requestJson),
}, config, runtime.logger, { method: "POST", pathname: "/v1/responses" });
result.probes.push({
kind: "responses",
target: "/v1/responses",
requested_model: requestedModel,
forwarded_model: forwardedModel || requestedModel,
status: responseProbe.status,
...await readProbeBodySummary(responseProbe, 600),
});
}
return result;
}
function extractProviderBaseUrl(content, providerName) {
if (!content || !providerName) {
return null;
}
const sectionPattern = new RegExp(
String.raw`^\[model_providers\.${providerName.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}\]\s*$[\s\S]*?(?=^\[|\Z)`,
"m",
);
const sectionMatch = content.match(sectionPattern);
if (!sectionMatch) {
return null;
}
const baseUrlMatch = sectionMatch[0].match(/^\s*base_url\s*=\s*"([^"]+)"\s*$/m);
return baseUrlMatch ? baseUrlMatch[1] : null;
}
async function readRuntimeState(runtime) {
const state = await readOptionalJson(runtime.paths.statePath);
if (!state) {
return null;
}
let codexCurrentBaseUrl = null;
if (state.codex_config_path && state.provider_name) {
try {
const codexConfig = await readFile(state.codex_config_path, "utf8");
codexCurrentBaseUrl = extractProviderBaseUrl(codexConfig, state.provider_name);
} catch {
codexCurrentBaseUrl = null;
}
}
return {
...state,
codex_current_base_url: codexCurrentBaseUrl,
};
}
async function restoreRuntimeState(runtime, state) {
const backupPath = state?.latest_backup_path;
const codexConfigPath = state?.codex_config_path;
if (!backupPath || !fs.existsSync(backupPath)) {
throw new Error(`未找到可恢复备份: ${backupPath || "unknown"}`);
}
if (!codexConfigPath) {
throw new Error("安装状态里缺少 codex_config_path");
}
await copyFile(backupPath, codexConfigPath);
await Promise.all([
rm(runtime.paths.statePath, { force: true }),
rm(runtime.paths.pidPath, { force: true }),
]);
}
function pickContentEncoding(acceptEncoding = "") {
const value = `${acceptEncoding}`.toLowerCase();
if (value.includes("br")) {
return "br";
}
if (value.includes("gzip")) {
return "gzip";
}
return null;
}
function shouldCompressContent(headers = {}) {
const contentType = `${headers["content-type"] || headers["Content-Type"] || ""}`.toLowerCase();
return (
contentType.startsWith("text/") ||
contentType.includes("javascript") ||
contentType.includes("json") ||
contentType.includes("xml") ||
contentType.includes("svg")
);
}
function maybeCompressBody(body, headers = {}, acceptEncoding = "") {
if (!body || body.length < 1024 || !shouldCompressContent(headers)) {
return { body, encoding: null };
}
const encoding = pickContentEncoding(acceptEncoding);
if (encoding === "br") {
return { body: zlib.brotliCompressSync(body), encoding };
}
if (encoding === "gzip") {
return { body: zlib.gzipSync(body), encoding };
}
return { body, encoding: null };
}
function respondBuffer(res, statusCode, body, headers = {}, acceptEncoding = "") {
const { body: responseBody, encoding } = maybeCompressBody(body, headers, acceptEncoding);
res.writeHead(statusCode, {
"content-length": responseBody.length,
vary: "accept-encoding",
...(encoding ? { "content-encoding": encoding } : {}),
...headers,
});
res.end(responseBody);
}
function jsonResponse(req, res, statusCode, payload, headers = {}) {
const body = Buffer.from(JSON.stringify(payload));
respondBuffer(
res,
statusCode,
body,
{
"content-type": "application/json; charset=utf-8",
...headers,
},
req?.headers?.["accept-encoding"] || "",
);
}
const STATIC_CONTENT_TYPES = {
".html": "text/html; charset=utf-8",
".js": "text/javascript; charset=utf-8",
".css": "text/css; charset=utf-8",
".json": "application/json; charset=utf-8",
".svg": "image/svg+xml",
".png": "image/png",
".ico": "image/x-icon",
".woff": "font/woff",
".woff2": "font/woff2",
};
function contentTypeForFile(filePath) {
return STATIC_CONTENT_TYPES[path.extname(filePath).toLowerCase()] || "application/octet-stream";
}
function safeJoinStatic(root, requestPath) {
const decoded = decodeURIComponent(requestPath);
const relative = decoded.replace(/^\/+/, "");
const fullPath = path.resolve(root, relative);
const rootPath = path.resolve(root);
if (fullPath !== rootPath && !fullPath.startsWith(`${rootPath}${path.sep}`)) {
return null;
}
return fullPath;
}
async function serveStaticFile(req, res, filePath) {
try {
const body = await readFile(filePath);
respondBuffer(
res,
200,
body,
{
"content-type": contentTypeForFile(filePath),
"cache-control": filePath.includes(`${path.sep}assets${path.sep}`)
? "public, max-age=31536000, immutable"
: "no-cache",
},
req?.headers?.["accept-encoding"] || "",
);
return true;
} catch {
return false;
}
}
async function serveManagementUi(req, res, requestPathname) {
const uiPrefix = `${UI_PATH}/`;
if (requestPathname === UI_PATH || requestPathname === `${UI_PATH}/`) {
return serveStaticFile(req, res, path.join(UI_STATIC_ROOT, "index.html"));
}
if (!requestPathname.startsWith(uiPrefix)) {
return false;
}
const staticPath = safeJoinStatic(UI_STATIC_ROOT, requestPathname.slice(uiPrefix.length));
if (!staticPath) {
jsonResponse(req, res, 403, {
error: {
message: "invalid static path",
code: "invalid_static_path",
},
});
return true;
}
if (await serveStaticFile(req, res, staticPath)) {
return true;
}
return serveStaticFile(req, res, path.join(UI_STATIC_ROOT, "index.html"));
}
async function serveManagementUiWithOptionalCookie(req, res, requestPathname, cookieHeaders) {
const uiPrefix = `${UI_PATH}/`;
if (requestPathname === UI_PATH || requestPathname === `${UI_PATH}/`) {
try {
const body = await readFile(path.join(UI_STATIC_ROOT, "index.html"));
respondBuffer(
res,
200,
body,
{
"content-type": "text/html; charset=utf-8",
"cache-control": "no-cache",
...(cookieHeaders?.length ? { "set-cookie": cookieHeaders } : {}),
},
req?.headers?.["accept-encoding"] || "",
);
return true;
} catch {
return false;
}
}
if (!requestPathname.startsWith(uiPrefix)) {
return false;
}
return serveManagementUi(req, res, requestPathname);
}
function renderManagementAccessPage(requestPathname, errorMessage = "") {
const escapedAction = `${requestPathname || UI_PATH}`.replace(/&/g, "&amp;").replace(/"/g, "&quot;");
const escapedError = `${errorMessage || ""}`
.replace(/&/g, "&amp;")
.replace(/</g, "&lt;")
.replace(/>/g, "&gt;");
return `<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Codex Retry Gateway</title>
<style>
body { margin: 0; font-family: ui-sans-serif, system-ui, sans-serif; background: #0b1020; color: #eef2ff; }
main { min-height: 100vh; display: grid; place-items: center; padding: 24px; }
form { width: min(420px, 100%); background: #121933; border: 1px solid #26304f; padding: 24px; border-radius: 8px; }
h1 { margin: 0 0 8px; font-size: 20px; }
p { margin: 0 0 16px; color: #aab4d6; line-height: 1.5; }
label { display: block; margin-bottom: 8px; font-size: 14px; color: #cbd5f5; }
input { width: 100%; box-sizing: border-box; padding: 12px; border-radius: 6px; border: 1px solid #33406a; background: #0f1630; color: #eef2ff; }
button { margin-top: 16px; width: 100%; padding: 12px; border: 0; border-radius: 6px; background: #7c9cff; color: #09101f; font-weight: 600; cursor: pointer; }
.error { min-height: 20px; margin-top: 12px; color: #fca5a5; font-size: 14px; }
</style>
</head>
<body>
<main>
<form method="get" action="${escapedAction}">
<h1>Codex Retry Gateway</h1>
<p>此页面已启用访问 key。输入后会在当前浏览器保存后台访问状态。</p>
<label for="key">Access key</label>
<input id="key" name="key" type="password" autofocus autocomplete="current-password">
<button type="submit">进入后台</button>
<div class="error">${escapedError}</div>
</form>
</main>
</body>
</html>`;
}
function buildEditableConfig(currentConfig, payload) {
const nextReasoning = normalizeIntegerList(payload.reasoning_equals, currentConfig.reasoning_equals);
const nextRetryableStatusCodes = normalizeIntegerList(
payload.retryable_status_codes,
currentConfig.retryable_status_codes,
);
const nextRetryableErrorMessages = normalizePhraseList(
payload.retryable_error_messages,
currentConfig.retryable_error_messages,
);
const nextEndpoints = normalizeStringList(payload.endpoints, currentConfig.endpoints).map(normalizePath);
const nextUpstreamFetchRetryAttempts =
payload.upstream_fetch_retry_attempts === undefined
? normalizePositiveInteger(
currentConfig.upstream_fetch_retry_attempts,
DEFAULT_CONFIG.upstream_fetch_retry_attempts,
)
: normalizePositiveInteger(
payload.upstream_fetch_retry_attempts,
DEFAULT_CONFIG.upstream_fetch_retry_attempts,
);
const nextUpstreamFetchRetryBackoffMs =
payload.upstream_fetch_retry_backoff_ms === undefined
? normalizeNonNegativeInteger(
currentConfig.upstream_fetch_retry_backoff_ms,
DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms,
)
: normalizeNonNegativeInteger(
payload.upstream_fetch_retry_backoff_ms,
DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms,
);
const nextStatusCode =
payload.non_stream_status_code === undefined
? currentConfig.non_stream_status_code
: Number.parseInt(`${payload.non_stream_status_code}`, 10);
if (nextReasoning.length === 0) {
throw new Error("reasoning_equals 不能为空");
}
if (nextRetryableStatusCodes.length === 0) {
throw new Error("retryable_status_codes 不能为空");
}
if (nextRetryableErrorMessages.length === 0) {
throw new Error("retryable_error_messages 不能为空");
}
if (nextEndpoints.length === 0) {
throw new Error("endpoints 不能为空");
}
if (nextUpstreamFetchRetryAttempts < 1) {
throw new Error("upstream_fetch_retry_attempts 必须是正整数");
}
if (nextUpstreamFetchRetryBackoffMs < 0) {
throw new Error("upstream_fetch_retry_backoff_ms 不能为负数");
}
if (!Number.isInteger(nextStatusCode) || nextStatusCode < 100 || nextStatusCode > 599) {
throw new Error("non_stream_status_code 必须是 100-599 的整数");
}
return {
...currentConfig,
reasoning_equals: nextReasoning,
retryable_status_codes: nextRetryableStatusCodes,
retryable_error_messages: nextRetryableErrorMessages,
management_access_key:
payload.management_access_key === undefined
? currentManagementAccessKey(currentConfig)
: normalizeManagementAccessKey(payload.management_access_key),
endpoints: nextEndpoints,
upstream_fetch_retry_attempts: nextUpstreamFetchRetryAttempts,
upstream_fetch_retry_backoff_ms: nextUpstreamFetchRetryBackoffMs,
non_stream_status_code: nextStatusCode,
log_match: payload.log_match === undefined ? currentConfig.log_match : Boolean(payload.log_match),
};
}
async function handleManagementRequest(runtime, req, res, requestUrl) {
const pathname = normalizePath(requestUrl.pathname);
const accessEnabled = managementAccessEnabled(runtime.config);
const accessGranted = hasManagementAccess(req, requestUrl, runtime.config);
if (pathname === UI_PATH || pathname.startsWith(`${UI_PATH}/`)) {
if (accessEnabled && !accessGranted) {
if (
pathname === UI_PATH ||
pathname === `${UI_PATH}/` ||
pathname === `${UI_PATH}/index.html`
) {
const body = Buffer.from(
renderManagementAccessPage(pathname, requestUrl.searchParams.get("key") ? "access key 不正确" : ""),
"utf8",
);
return respondBuffer(
res,
401,
body,
{
"content-type": "text/html; charset=utf-8",
"cache-control": "no-store",
"set-cookie": clearManagementAccessCookieHeaders(),
},
req?.headers?.["accept-encoding"] || "",
);
}
jsonResponse(req, res, 401, managementUnauthorizedPayload(), {
"cache-control": "no-store",
"set-cookie": clearManagementAccessCookieHeaders(),
});
return true;
}
const cookieHeaders = accessEnabled
? buildManagementAccessCookieHeaders(runtime.config, requestManagementAccessKey(req, requestUrl))
: [];
if (!(await serveManagementUiWithOptionalCookie(req, res, requestUrl.pathname, cookieHeaders))) {
jsonResponse(req, res, 503, {
error: {
message: "UI assets were not built. Run: npm run build:ui",
code: "ui_not_built",
},
});
}
return true;
}
if (accessEnabled && !accessGranted && pathname.startsWith(`${ADMIN_BASE_PATH}/`)) {
jsonResponse(req, res, 401, managementUnauthorizedPayload(), {
"cache-control": "no-store",
"set-cookie": clearManagementAccessCookieHeaders(),
});
return true;
}
if (pathname === STATUS_API_PATH && req.method === "GET") {
const state = await readRuntimeState(runtime);
jsonResponse(req, res, 200, {
ok: true,
listen: `${runtime.config.listen_host}:${runtime.config.listen_port}`,
config: sanitizeConfigForStatus(runtime.config),
state,
paths: {
config_path: runtime.configPath,
state_path: runtime.paths.statePath,
state_root: runtime.paths.stateRoot,
log_path: runtime.logPath,
requests_path: runtime.paths.requestsPath,
profiles_dir: runtime.paths.profilesDir,
},
metrics: buildMetricsSnapshot(runtime.monitor),
}, accessEnabled && accessGranted ? { "set-cookie": buildManagementAccessCookieHeaders(runtime.config, requestManagementAccessKey(req, requestUrl)) } : {});
return true;
}
if (pathname === LOGS_API_PATH && req.method === "GET") {
const sinceSeqRaw = requestUrl.searchParams.get("since_seq");
const limitRaw = requestUrl.searchParams.get("limit");
const sinceSeq = sinceSeqRaw === null ? null : Number.parseInt(sinceSeqRaw, 10);
const limit = limitRaw === null ? 500 : Number.parseInt(limitRaw, 10);
jsonResponse(req, res, 200, {
ok: true,
...await buildPersistentLogsSnapshot(
runtime,
Number.isInteger(sinceSeq) ? sinceSeq : null,
Number.isInteger(limit) ? limit : 500,
),
});
return true;
}
if (pathname === REQUESTS_API_PATH && req.method === "GET") {
const limitRaw = requestUrl.searchParams.get("limit");
const offsetRaw = requestUrl.searchParams.get("offset");
const query = requestUrl.searchParams.get("query") || "";
const filter = requestUrl.searchParams.get("filter") || "all";
const limit = limitRaw === null ? 50 : Number.parseInt(limitRaw, 10);
const offset = offsetRaw === null ? 0 : Number.parseInt(offsetRaw, 10);
jsonResponse(req, res, 200, {
ok: true,
...await buildPersistentRequestsSnapshot(runtime, {
limit: Number.isInteger(limit) ? limit : 50,
offset: Number.isInteger(offset) ? offset : 0,
query,
filter,
}),
});
return true;
}
if (pathname === PROFILES_API_PATH && req.method === "GET") {
jsonResponse(req, res, 200, {
ok: true,
profiles_dir: runtime.paths.profilesDir,
active_profile: runtime.config.profile_name || "default",
profiles: await listProfiles(runtime),
});
return true;
}
if (pathname === PROFILES_API_PATH && req.method === "POST") {
const body = await readRequestBody(req, runtime.config.request_body_limit_bytes);
const payload = parseJsonSafely(body);
if (!payload) {
jsonResponse(req, res, 400, {
error: {
message: "profile 保存请求必须是有效 JSON",
code: "invalid_json",
},
});
return true;
}
const result = await writeProfile(runtime, payload);
let applied = null;
if (result.name === (runtime.config.profile_name || "default")) {
applied = await applyProfileConfig(runtime, result.name);
}
runtime.logger(`[profile] saved name=${result.name} path=${result.file_path}`);
jsonResponse(req, res, 200, {
ok: true,
message: applied ? "profile 已保存并已热应用" : "profile 已保存",
saved_profile: result,
applied_profile: applied,
profiles_dir: runtime.paths.profilesDir,
active_profile: runtime.config.profile_name || "default",
profiles: await listProfiles(runtime),
});
return true;
}
if (pathname === PROFILE_PROBE_API_PATH && req.method === "POST") {
const body = await readRequestBody(req, runtime.config.request_body_limit_bytes);
const payload = parseJsonSafely(body);
if (!payload) {
jsonResponse(req, res, 400, {
error: {
message: "profile probe 请求必须是有效 JSON",
code: "invalid_json",
},
});
return true;
}
const result = await probeProfile(runtime, payload);
runtime.logger(
`[profile-probe] profile=${result.profile} auth=${result.auth_mode}/${result.auth_source} upstream=${result.upstream_base_url}`,
);
jsonResponse(req, res, 200, {
ok: true,
...result,
});
return true;
}
if (pathname === PROFILE_SWITCH_API_PATH && req.method === "POST") {
const body = await readRequestBody(req, runtime.config.request_body_limit_bytes);
const payload = parseJsonSafely(body);
const profileName = `${payload?.profile || ""}`.trim();
if (!profileName) {
jsonResponse(req, res, 400, {
error: {
message: "缺少 profile",
code: "profile_required",
},
});
return true;
}
const result = await applyProfileConfig(runtime, profileName);
jsonResponse(req, res, 200, {
ok: true,
message: "profile 已热切换,无需重启 gateway",
...result,
});
return true;
}
if (pathname.startsWith(PROFILE_ITEM_API_PREFIX) && req.method === "DELETE") {
const rawName = pathname.slice(PROFILE_ITEM_API_PREFIX.length);
if (!rawName || rawName.includes("/")) {
jsonResponse(req, res, 400, {
error: {
message: "无效的 profile 名称",
code: "invalid_profile_name",
},
});
return true;
}
const profileName = decodeURIComponent(rawName);
const result = await deleteProfile(runtime, profileName);
runtime.logger(`[profile] deleted name=${result.name} path=${result.file_path}`);
jsonResponse(req, res, 200, {
ok: true,
message: "profile 已删除",
deleted_profile: result,
profiles_dir: runtime.paths.profilesDir,
active_profile: runtime.config.profile_name || "default",
profiles: await listProfiles(runtime),
});
return true;
}
if (pathname === CONFIG_API_PATH && req.method === "POST") {
const body = await readRequestBody(req, runtime.config.request_body_limit_bytes);
const payload = parseJsonSafely(body);
if (!payload) {
jsonResponse(req, res, 400, {
error: {
message: "配置保存请求必须是有效 JSON",
code: "invalid_json",
},
});
return true;
}
const nextConfig = buildEditableConfig(runtime.config, payload);
await writeConfig(runtime.configPath, nextConfig);
runtime.config = nextConfig;
runtime.logger(
`[config] updated reasoning_equals=${nextConfig.reasoning_equals.join(",")} retryable_status_codes=${nextConfig.retryable_status_codes.join(",")} endpoints=${nextConfig.endpoints.join(",")}`,
);
const state = await readRuntimeState(runtime);
jsonResponse(req, res, 200, {
ok: true,
message: "配置已保存并立即生效",
config: sanitizeConfigForStatus(runtime.config),
state,
paths: {
config_path: runtime.configPath,
state_path: runtime.paths.statePath,
state_root: runtime.paths.stateRoot,
log_path: runtime.logPath,
},
metrics: buildMetricsSnapshot(runtime.monitor),
});
return true;
}
if (pathname === RESTORE_API_PATH && req.method === "POST") {
const state = await readRuntimeState(runtime);
if (!state) {
jsonResponse(req, res, 409, {
error: {
message: "当前未检测到安装状态,无法恢复 Codex 原设置",
code: "state_not_found",
},
});
return true;
}
await restoreRuntimeState(runtime, state);
runtime.logger(`[restore] restored via UI state_root=${runtime.paths.stateRoot}`);
jsonResponse(req, res, 202, {
ok: true,
message: "原设置已恢复,gateway 即将关闭",
});
res.on("finish", () => {
const exitTimer = setTimeout(() => {
if (runtime.server) {
runtime.server.close(() => {
process.exit(0);
});
} else {
process.exit(0);
}
const hardExitTimer = setTimeout(() => {
process.exit(0);
}, 600);
hardExitTimer.unref();
}, 120);
exitTimer.unref();
});
return true;
}
return false;
}
function buildUpstreamUrl(baseUrl, requestUrl) {
const upstream = new URL(baseUrl);
const normalizedBasePath = upstream.pathname.endsWith("/")
? upstream.pathname.slice(0, -1)
: upstream.pathname;
const incomingPath = requestUrl.pathname;
let finalPath = incomingPath;
if (normalizedBasePath && normalizedBasePath !== "/") {
if (incomingPath.startsWith(`${normalizedBasePath}/`) || incomingPath === normalizedBasePath) {
finalPath = incomingPath;
} else if (normalizedBasePath.endsWith("/v1") && incomingPath.startsWith("/v1/")) {
finalPath = `${normalizedBasePath}${incomingPath.slice(3)}`;
} else {
finalPath = `${normalizedBasePath}${incomingPath}`;
}
}
upstream.pathname = finalPath;
upstream.search = requestUrl.search;
return upstream.toString();
}
function buildUpstreamSnapshot({ upstreamUrl, upstreamAuth = null, upstreamResponse = null }) {
const parsedUrl = new URL(upstreamUrl);
const snapshot = {
origin: parsedUrl.origin,
path: parsedUrl.pathname,
auth_mode: upstreamAuth?.mode || "unknown",
auth_source: upstreamAuth?.source || "unknown",
authorization_configured: Boolean(upstreamAuth?.authorization),
};
if (upstreamResponse) {
snapshot.status = upstreamResponse.status;
snapshot.content_type = upstreamResponse.headers.get("content-type") || "";
}
return snapshot;
}
function cloneHeadersForUpstream(headers, upstreamAuth = null) {
const outgoing = new Headers();
for (const [key, value] of Object.entries(headers)) {
if (value === undefined) {
continue;
}
const lowerKey = key.toLowerCase();
if (
lowerKey === "host" ||
lowerKey === "content-length" ||
lowerKey === "connection" ||
lowerKey === "transfer-encoding"
) {
continue;
}
if (Array.isArray(value)) {
for (const item of value) {
outgoing.append(key, item);
}
} else {
outgoing.set(key, value);
}
}
if (upstreamAuth?.authorization) {
outgoing.set("authorization", upstreamAuth.authorization);
}
return outgoing;
}
function copyHeadersToClient(sourceHeaders, target) {
for (const [key, value] of sourceHeaders.entries()) {
const lowerKey = key.toLowerCase();
if (
lowerKey === "content-length" ||
lowerKey === "transfer-encoding" ||
lowerKey === "content-encoding" ||
lowerKey === "connection"
) {
continue;
}
target.setHeader(key, value);
}
}
async function readRequestBody(req, limitBytes) {
const chunks = [];
let total = 0;
for await (const chunk of req) {
total += chunk.length;
if (total > limitBytes) {
throw new Error(`请求体超过限制: ${limitBytes} bytes`);
}
chunks.push(chunk);
}
return Buffer.concat(chunks);
}
function parseJsonSafely(buffer) {
try {
return JSON.parse(buffer.toString("utf8"));
} catch {
return null;
}
}
function rebuildBufferedResponse(response, bodyBuffer) {
return new Response(bodyBuffer, {
status: response.status,
statusText: response.statusText,
headers: new Headers(response.headers),
});
}
function matchPath(config, pathname) {
return config.endpoints.includes(normalizePath(pathname));
}
function reasoningMatched(config, reasoning) {
return reasoning !== null && config.reasoning_equals.includes(reasoning);
}
function collectRetryableMessageCandidates(value, state = { seen: new Set(), results: [] }, depth = 0) {
if (value === null || value === undefined || depth > 5 || state.results.length >= 64) {
return state.results;
}
if (typeof value === "string") {
const trimmed = value.trim();
if (trimmed) {
state.results.push(trimmed);
}
return state.results;
}
if (typeof value !== "object") {
return state.results;
}
if (state.seen.has(value)) {
return state.results;
}
state.seen.add(value);
if (Array.isArray(value)) {
for (const item of value) {
collectRetryableMessageCandidates(item, state, depth + 1);
}
return state.results;
}
const preferredKeys = ["message", "error", "detail", "details", "description", "title"];
for (const key of preferredKeys) {
if (Object.hasOwn(value, key)) {
collectRetryableMessageCandidates(value[key], state, depth + 1);
}
}
for (const [key, nested] of Object.entries(value)) {
if (preferredKeys.includes(key)) {
continue;
}
collectRetryableMessageCandidates(nested, state, depth + 1);
}
return state.results;
}
function truncateRetryableMessage(value, maxLength = 280) {
const text = `${value || ""}`.trim();
if (!text) {
return "";
}
return text.length > maxLength ? `${text.slice(0, maxLength - 3)}...` : text;
}
function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, bodyText) {
const retryableMessages = normalizePhraseList(
config.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
if (retryableMessages.length === 0) {
return null;
}
const retryableStatusCodes = normalizeIntegerList(
config.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
);
const statusEligible = isUpstreamErrorStatus(upstreamStatusCode) && (
retryableStatusCodes.length === 0 ||
retryableStatusCodes.includes(upstreamStatusCode)
);
if (!statusEligible && !isRetryableErrorPayloadShape(parsedBody)) {
return null;
}
return matchRetryableMessage(config, parsedBody, bodyText);
}
function matchRetryableMessage(config, parsedBody, bodyText) {
const retryableMessages = normalizePhraseList(
config.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
if (retryableMessages.length === 0) {
return null;
}
const normalizedPatterns = retryableMessages.map((message) => ({
original: message,
normalized: message.toLowerCase(),
}));
const candidateMessages = [];
if (parsedBody) {
candidateMessages.push(...collectRetryableMessageCandidates(parsedBody));
}
const trimmedBodyText = `${bodyText || ""}`.trim();
if (trimmedBodyText) {
candidateMessages.push(trimmedBodyText);
}
for (const candidate of candidateMessages) {
const normalizedCandidate = candidate.toLowerCase();
for (const pattern of normalizedPatterns) {
if (normalizedCandidate.includes(pattern.normalized)) {
return {
matched_pattern: pattern.original,
matched_message: truncateRetryableMessage(candidate),
};
}
}
}
return null;
}
function hasRetryableFailureKeyword(value) {
const normalized = `${value || ""}`.trim().toLowerCase();
if (!normalized) {
return false;
}
return (
normalized.includes("error") ||
normalized.includes("failed") ||
normalized.includes("failure") ||
normalized.includes("cancelled") ||
normalized.includes("canceled") ||
normalized.includes("rate_limit") ||
normalized.includes("overloaded") ||
normalized.includes("unavailable")
);
}
function hasRetryableFailureShape(value, depth = 0) {
if (!value || depth > 4) {
return false;
}
if (Array.isArray(value)) {
return value.some((item) => hasRetryableFailureShape(item, depth + 1));
}
if (typeof value !== "object") {
return false;
}
if (Object.hasOwn(value, "error")) {
return true;
}
if (Number.isInteger(value.status) && value.status >= 400) {
return true;
}
for (const key of ["type", "event", "status", "state", "result", "code"]) {
if (typeof value[key] === "string" && hasRetryableFailureKeyword(value[key])) {
return true;
}
}
for (const key of ["response", "data", "meta", "details", "detail"]) {
if (hasRetryableFailureShape(value[key], depth + 1)) {
return true;
}
}
return false;
}
function isRetryableErrorPayloadShape(parsedBody, eventName = "") {
if (hasRetryableFailureKeyword(eventName)) {
return true;
}
if (!parsedBody || typeof parsedBody !== "object" || Array.isArray(parsedBody)) {
return false;
}
return hasRetryableFailureShape(parsedBody);
}
function findRetryableStreamErrorMatch(config, parsedBody, bodyText, eventName = "") {
if (!isRetryableErrorPayloadShape(parsedBody, eventName)) {
return null;
}
return matchRetryableMessage(config, parsedBody, bodyText);
}
function findRetryableStreamTerminationMatch(config, error) {
const message = `${error?.message || error || ""}`.trim();
if (!message) {
return null;
}
return matchRetryableMessage(config, null, message);
}
function isExpectedStreamTermination(error) {
if (!error) {
return false;
}
const message = `${error?.message || ""}`.trim().toLowerCase();
if (error.name === "AbortError") {
return true;
}
return error instanceof TypeError && (
message === "terminated" ||
message.includes("stream disconnected before completion")
);
}
function isRetryableUpstreamFetchError(error) {
if (!error) {
return false;
}
return error instanceof TypeError && error.message === "fetch failed";
}
function isUpstreamErrorStatus(statusCode) {
return Number.isInteger(statusCode) && statusCode >= 400;
}
function computeRetryBackoffMs(config, attempt) {
const baseDelay = normalizeNonNegativeInteger(
config?.upstream_fetch_retry_backoff_ms,
DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms,
);
if (baseDelay <= 0 || attempt <= 1) {
return baseDelay;
}
const multiplier = Math.min(attempt - 1, 4);
return baseDelay * multiplier;
}
async function fetchUpstreamWithRetry(upstreamUrl, init, config, logger, requestContext = {}) {
const maxAttempts = normalizePositiveInteger(
config?.upstream_fetch_retry_attempts,
DEFAULT_CONFIG.upstream_fetch_retry_attempts,
);
const method = `${requestContext.method || init?.method || "GET"}`.toUpperCase();
const pathname = requestContext.pathname || "";
let lastError = null;
let lastResponse = null;
let retryableUpstreamError = null;
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
try {
const response = await fetch(upstreamUrl, init);
lastResponse = response;
if (method === "GET") {
return {
response,
attempt_count: attempt,
retryable_upstream_error: null,
};
}
const contentType = response.headers.get("content-type");
if (!isJsonContentType(contentType)) {
return {
response,
attempt_count: attempt,
retryable_upstream_error: null,
};
}
const bodyBuffer = Buffer.from(await response.arrayBuffer());
const parsed = parseJsonSafely(bodyBuffer);
const bodyText = bodyBuffer.toString("utf8");
retryableUpstreamError = findRetryableUpstreamErrorMatch(
config,
response.status,
parsed,
bodyText,
);
if (!retryableUpstreamError) {
return {
response: rebuildBufferedResponse(response, bodyBuffer),
attempt_count: attempt,
retryable_upstream_error: null,
};
}
if (attempt === maxAttempts) {
return {
response: rebuildBufferedResponse(response, bodyBuffer),
attempt_count: attempt,
retryable_upstream_error: {
...retryableUpstreamError,
upstream_status_code: response.status,
},
};
}
const backoffMs = computeRetryBackoffMs(config, attempt);
logger?.(
`[retry] upstream retryable error attempt=${attempt} next_attempt=${attempt + 1} status=${response.status} path=${pathname || "-"} reason=${JSON.stringify(retryableUpstreamError.matched_pattern)} backoff_ms=${backoffMs}`,
);
if (backoffMs > 0) {
await sleep(backoffMs);
}
} catch (error) {
lastError = error;
if (error && typeof error === "object") {
error.gatewayAttemptCount = attempt;
}
if (!isRetryableUpstreamFetchError(error) || attempt === maxAttempts) {
break;
}
const backoffMs = computeRetryBackoffMs(config, attempt);
logger?.(
`[retry] upstream fetch failed attempt=${attempt} next_attempt=${attempt + 1} method=${method} path=${pathname || "-"} url=${upstreamUrl} backoff_ms=${backoffMs}`,
);
if (backoffMs > 0) {
await sleep(backoffMs);
}
}
}
if (lastError) {
throw lastError;
}
return {
response: lastResponse,
attempt_count: maxAttempts,
retryable_upstream_error: retryableUpstreamError,
};
}
function inspectSseChunk(state, chunk, config) {
const decoded = state.decoder.decode(chunk, { stream: true });
state.buffer += decoded;
const result = {
reasoning: null,
usage: null,
response_id: null,
thread_id: null,
retryable_upstream_error: null,
};
const blocks = state.buffer.split(/\r?\n\r?\n/);
state.buffer = blocks.pop() ?? "";
for (const block of blocks) {
const lines = block
.split(/\r?\n/)
.map((line) => line.trimEnd())
.filter(Boolean);
const eventName = lines
.filter((line) => line.startsWith("event:"))
.map((line) => line.replace(/^event:\s?/, "").trim())
.find(Boolean) || "";
const dataLines = lines
.filter((line) => line.startsWith("data:"))
.map((line) => line.replace(/^data:\s?/, ""));
if (dataLines.length === 0) {
continue;
}
const payloadText = dataLines.join("\n");
if (payloadText === "[DONE]") {
continue;
}
let parsed = null;
try {
parsed = JSON.parse(payloadText);
const reasoning = extractReasoningTokens(parsed);
if (reasoning !== null) {
result.reasoning = reasoning;
}
result.usage = mergeUsageSnapshots(result.usage, normalizeUsageSnapshot(parsed));
result.response_id = result.response_id || extractStreamingResponseId(parsed);
result.thread_id = result.thread_id || extractResponseThreadId(parsed);
} catch {
// ignore malformed SSE payloads
}
const retryableUpstreamError = findRetryableStreamErrorMatch(
config,
parsed,
payloadText,
eventName,
);
if (retryableUpstreamError) {
result.retryable_upstream_error = retryableUpstreamError;
}
}
return result;
}
async function handleNonStreaming({
runtime,
config,
logger,
monitor,
pathname,
upstreamResponse,
res,
requestEntry,
terminalRetryableUpstreamError = null,
}) {
markAndPersistFirstResponse(runtime, requestEntry);
const bodyBuffer = Buffer.from(await upstreamResponse.arrayBuffer());
const bodyText = bodyBuffer.toString("utf8");
const parsed = isJsonContentType(upstreamResponse.headers.get("content-type"))
? parseJsonSafely(bodyBuffer)
: null;
const reasoning = parsed ? extractReasoningTokens(parsed) : null;
const usage = parsed ? normalizeUsageSnapshot(parsed) : null;
const responseId = parsed ? extractNonStreamingResponseId(parsed) : null;
const threadId = firstNonEmptyString(
requestEntry.thread_id,
parsed ? extractResponseThreadId(parsed) : null,
);
const matched = reasoningMatched(config, reasoning);
const retryableUpstreamError = terminalRetryableUpstreamError || findRetryableUpstreamErrorMatch(
config,
upstreamResponse.status,
parsed,
bodyText,
);
requestEntry.response_id = responseId || requestEntry.response_id || null;
requestEntry.thread_id = threadId || requestEntry.thread_id || null;
recordInspectedResponse(monitor, reasoning, matched || Boolean(retryableUpstreamError));
if (matched) {
if (config.log_match) {
logger(
`[match] non-stream path=${pathname} reasoning_tokens=${reasoning} action=status_${config.non_stream_status_code}`,
);
}
const blockedBody = buildBlockedBody(pathname, reasoning, config.non_stream_status_code);
res.writeHead(config.non_stream_status_code, {
"content-type": "application/json; charset=utf-8",
"x-codex-retry-gateway-reason": "reasoning-guard-triggered",
});
res.end(blockedBody);
return {
inspected: true,
matched,
status_code: config.non_stream_status_code,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: reasoning,
usage,
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
};
}
if (retryableUpstreamError) {
if (config.log_match) {
logger(
`[match] non-stream path=${pathname} upstream_status=${upstreamResponse.status} retryable_error=${JSON.stringify(retryableUpstreamError.matched_pattern)} action=status_${config.non_stream_status_code}`,
);
}
const blockedBody = buildRetryableUpstreamErrorBody(
pathname,
upstreamResponse.status,
retryableUpstreamError.matched_message || retryableUpstreamError.matched_pattern,
config.non_stream_status_code,
);
res.writeHead(config.non_stream_status_code, {
"content-type": "application/json; charset=utf-8",
"x-codex-retry-gateway-reason": "upstream-error-retry-triggered",
});
res.end(blockedBody);
return {
inspected: true,
matched: true,
status_code: config.non_stream_status_code,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: reasoning,
usage,
error: `retryable upstream error: ${retryableUpstreamError.matched_pattern}`,
match_reason: "retryable_upstream_error",
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
};
}
copyHeadersToClient(upstreamResponse.headers, res);
res.writeHead(upstreamResponse.status);
res.end(bodyBuffer);
return {
inspected: true,
matched,
status_code: upstreamResponse.status,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: reasoning,
usage,
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
};
}
async function handleStreaming({
runtime,
config,
logger,
monitor,
pathname,
upstreamResponse,
res,
abortController,
requestEntry,
}) {
const strict502Mode = config.stream_action !== "disconnect";
const reader = upstreamResponse.body.getReader();
const sseState = {
decoder: new TextDecoder("utf8"),
buffer: "",
};
let wroteAnyChunk = false;
let observedReasoning = null;
let observedUsage = null;
const bufferedChunks = [];
if (!strict502Mode) {
copyHeadersToClient(upstreamResponse.headers, res);
res.writeHead(upstreamResponse.status);
}
while (true) {
let readResult;
try {
readResult = await reader.read();
} catch (error) {
if (isExpectedStreamTermination(error)) {
const retryableTerminationError = findRetryableStreamTerminationMatch(config, error);
if (retryableTerminationError) {
return {
inspected: true,
matched: true,
retry_requested: strict502Mode || !wroteAnyChunk,
retryable_upstream_error: retryableTerminationError,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: observedReasoning,
usage: observedUsage,
error: `retryable upstream error: ${retryableTerminationError.matched_pattern}`,
match_reason: "retryable_upstream_error",
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
}
recordInspectedResponse(monitor, observedReasoning, false);
persistStreamingProgress(runtime, requestEntry, { force: true }, new Date());
if (strict502Mode) {
logger?.(`[stream] upstream terminated before completion path=${pathname} action=status_502`);
res.writeHead(502, { "content-type": "application/json; charset=utf-8" });
res.end(buildGatewayErrorBody("upstream stream terminated before completion"));
return {
inspected: true,
matched: false,
status_code: 502,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: observedReasoning,
usage: observedUsage,
error: "upstream stream terminated before completion",
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
} else {
res.end();
return {
inspected: true,
matched: false,
status_code: upstreamResponse.status,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: observedReasoning,
usage: observedUsage,
error: "upstream stream terminated before completion",
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
}
}
throw error;
}
const { done, value } = readResult;
if (done) {
recordInspectedResponse(monitor, observedReasoning, false);
persistStreamingProgress(runtime, requestEntry, { force: true }, new Date());
if (strict502Mode) {
copyHeadersToClient(upstreamResponse.headers, res);
res.writeHead(upstreamResponse.status);
res.end(Buffer.concat(bufferedChunks));
} else {
res.end();
}
return {
inspected: true,
matched: false,
status_code: upstreamResponse.status,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: observedReasoning,
usage: observedUsage,
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
}
const chunkBuffer = Buffer.from(value);
const now = new Date();
const inspection = inspectSseChunk(sseState, value, config);
const retryableUpstreamError = inspection.retryable_upstream_error;
if (retryableUpstreamError) {
abortController.abort();
reader.cancel().catch(() => {});
return {
inspected: true,
matched: true,
retry_requested: strict502Mode || !wroteAnyChunk,
retryable_upstream_error: retryableUpstreamError,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: observedReasoning,
usage: observedUsage,
error: `retryable upstream error: ${retryableUpstreamError.matched_pattern}`,
match_reason: "retryable_upstream_error",
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
}
markAndPersistFirstResponse(runtime, requestEntry, now);
const reasoning = inspection.reasoning;
const usageUpdated = Boolean(inspection.usage);
observedUsage = mergeUsageSnapshots(observedUsage, inspection.usage);
if (Number.isInteger(reasoning)) {
observedReasoning = reasoning;
}
if (inspection.response_id) {
requestEntry.response_id = inspection.response_id;
}
if (inspection.thread_id) {
requestEntry.thread_id = inspection.thread_id;
}
updateStreamingProgress(requestEntry, {
chunkBytes: chunkBuffer.length,
usage: inspection.usage,
reasoning,
at: now,
});
persistStreamingProgress(
runtime,
requestEntry,
{ usageUpdated, force: requestEntry.stream_chunk_count === 1 },
now,
);
if (reasoningMatched(config, reasoning)) {
recordInspectedResponse(monitor, reasoning, true);
if (config.log_match) {
logger(
`[match] stream path=${pathname} reasoning_tokens=${reasoning} action=${config.stream_action}`,
);
}
if (strict502Mode || !wroteAnyChunk) {
abortController.abort();
reader.cancel().catch(() => {});
const blockedBody = buildBlockedBody(pathname, reasoning, config.non_stream_status_code);
res.writeHead(config.non_stream_status_code, {
"content-type": "application/json; charset=utf-8",
"x-codex-retry-gateway-reason": "reasoning-guard-triggered",
});
res.end(blockedBody);
} else {
abortController.abort();
reader.cancel().catch(() => {});
res.socket?.destroy();
}
return {
inspected: true,
matched: true,
status_code: config.non_stream_status_code,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: reasoning,
usage: observedUsage,
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
}
if (strict502Mode) {
bufferedChunks.push(chunkBuffer);
} else {
wroteAnyChunk = true;
res.write(chunkBuffer);
}
}
}
async function proxyRequest(runtime, req, res) {
const { logger } = runtime;
const config = runtime.config;
const maxUpstreamAttempts = normalizePositiveInteger(
config.upstream_fetch_retry_attempts,
DEFAULT_CONFIG.upstream_fetch_retry_attempts,
);
const requestStartedAt = new Date();
const requestStartedMs = Date.now();
const incomingUrl = new URL(req.url, `http://${req.headers.host || "127.0.0.1"}`);
const pathname = normalizePath(incomingUrl.pathname);
if (pathname === "/favicon.ico") {
res.writeHead(204, { "cache-control": "public, max-age=86400" });
res.end();
return;
}
if (pathname === config.health_path) {
res.writeHead(200, { "content-type": "application/json; charset=utf-8" });
res.end(
JSON.stringify({
ok: true,
listen: `${config.listen_host}:${config.listen_port}`,
upstream_base_url: config.upstream_base_url,
ui_path: UI_PATH,
}),
);
return;
}
if (await handleManagementRequest(runtime, req, res, incomingUrl)) {
return;
}
runtime.monitor.total_proxy_request_count += 1;
const requestSeq = runtime.monitor.next_request_seq;
runtime.monitor.next_request_seq += 1;
const requestEntry = buildRequestEntry({
seq: requestSeq,
startedAt: requestStartedAt,
startedMs: requestStartedMs,
req,
pathname,
requestJson: null,
profileName: config.profile_name || "default",
});
upsertRequestEntry(runtime, requestEntry);
try {
const rawRequestBody = await readRequestBody(req, config.request_body_limit_bytes);
const parsedRequestJson = isJsonContentType(req.headers["content-type"])
? parseJsonSafely(rawRequestBody)
: null;
const { requestJson, remapped, forwardedModel } = remapRequestModel(config, parsedRequestJson);
const requestBody = remapped ? Buffer.from(JSON.stringify(requestJson)) : rawRequestBody;
const requestIsStream = Boolean(requestJson?.stream);
let totalUpstreamAttempts = 0;
requestEntry.request_body_bytes = rawRequestBody.length;
requestEntry.request_id = computeRequestId(pathname, rawRequestBody);
requestEntry.thread_id = firstNonEmptyString(
extractRequestThreadId(parsedRequestJson),
extractHeaderThreadId(req.headers),
);
requestEntry.model = requestJson?.model || null;
requestEntry.requested_model = parsedRequestJson?.model || null;
requestEntry.forwarded_model = forwardedModel || parsedRequestJson?.model || null;
requestEntry.model = requestEntry.forwarded_model;
requestEntry.request_stream = requestIsStream;
upsertRequestEntry(runtime, requestEntry);
const upstreamUrl = buildUpstreamUrl(config.upstream_base_url, incomingUrl);
const upstreamAuth = await resolveUpstreamAuth(config);
requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth });
upsertRequestEntry(runtime, requestEntry);
if (remapped && requestEntry.requested_model && requestEntry.forwarded_model) {
logger?.(
`[model-remap] profile=${config.profile_name || "default"} requested=${requestEntry.requested_model} forwarded=${requestEntry.forwarded_model}`,
);
}
while (totalUpstreamAttempts < maxUpstreamAttempts) {
const abortController = new AbortController();
const {
response: upstreamResponse,
attempt_count: upstreamAttemptCount,
retryable_upstream_error: terminalRetryableUpstreamError,
} = await fetchUpstreamWithRetry(upstreamUrl, {
method: req.method,
headers: cloneHeadersForUpstream(req.headers, upstreamAuth),
body: requestBody.length > 0 ? requestBody : undefined,
signal: abortController.signal,
}, {
...config,
upstream_fetch_retry_attempts: Math.max(1, maxUpstreamAttempts - totalUpstreamAttempts),
}, logger, { method: req.method, pathname });
totalUpstreamAttempts += upstreamAttemptCount;
const shouldInspect = matchPath(config, pathname);
const responseContentType = upstreamResponse.headers.get("content-type");
const responseIsStream = isSseContentType(responseContentType) || (
requestIsStream &&
!isJsonContentType(responseContentType) &&
!isUpstreamErrorStatus(upstreamResponse.status)
);
requestEntry.response_stream = responseIsStream;
requestEntry.inspected = shouldInspect;
requestEntry.upstream_status_code = upstreamResponse.status;
requestEntry.upstream_attempt_count = totalUpstreamAttempts;
requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth, upstreamResponse });
upsertRequestEntry(runtime, requestEntry);
if (isUpstreamErrorStatus(upstreamResponse.status)) {
logger?.(
`[upstream] status=${upstreamResponse.status} profile=${config.profile_name || "default"} path=${requestEntry.upstream.path} auth=${requestEntry.upstream.auth_mode}/${requestEntry.upstream.auth_source} content_type=${requestEntry.upstream.content_type || "-"}`,
);
}
if (!shouldInspect) {
markRequestFirstResponse(requestEntry);
upsertRequestEntry(runtime, requestEntry);
copyHeadersToClient(upstreamResponse.headers, res);
res.writeHead(upstreamResponse.status);
const body = Buffer.from(await upstreamResponse.arrayBuffer());
const parsed = isJsonContentType(upstreamResponse.headers.get("content-type"))
? parseJsonSafely(body)
: null;
requestEntry.response_id = requestEntry.response_id || (parsed ? extractNonStreamingResponseId(parsed) : null);
requestEntry.thread_id = requestEntry.thread_id || (parsed ? extractResponseThreadId(parsed) : null);
res.end(body);
recordRequestEntry(
runtime,
finalizeRequestEntry(requestEntry, {
status_code: upstreamResponse.status,
upstream_status_code: upstreamResponse.status,
inspected: false,
}),
config.request_history_limit,
);
return;
}
if (responseIsStream) {
const result = await handleStreaming({
runtime,
config,
logger,
monitor: runtime.monitor,
pathname,
upstreamResponse,
res,
abortController,
requestEntry,
});
if (result.retry_requested) {
if (totalUpstreamAttempts < maxUpstreamAttempts) {
const backoffMs = computeRetryBackoffMs(config, totalUpstreamAttempts);
logger?.(
`[retry] upstream retryable stream error attempt=${totalUpstreamAttempts} next_attempt=${totalUpstreamAttempts + 1} status=${upstreamResponse.status} path=${pathname || "-"} reason=${JSON.stringify(result.retryable_upstream_error?.matched_pattern)} backoff_ms=${backoffMs}`,
);
if (backoffMs > 0) {
await sleep(backoffMs);
}
continue;
}
recordInspectedResponse(runtime.monitor, result.reasoning_tokens, true);
if (config.log_match) {
logger?.(
`[match] stream path=${pathname} upstream_status=${upstreamResponse.status} retryable_error=${JSON.stringify(result.retryable_upstream_error?.matched_pattern)} action=status_${config.non_stream_status_code}`,
);
}
const blockedBody = buildRetryableUpstreamErrorBody(
pathname,
upstreamResponse.status,
result.retryable_upstream_error?.matched_message || result.retryable_upstream_error?.matched_pattern,
config.non_stream_status_code,
);
res.writeHead(config.non_stream_status_code, {
"content-type": "application/json; charset=utf-8",
"x-codex-retry-gateway-reason": "upstream-error-retry-triggered",
});
res.end(blockedBody);
recordRequestEntry(
runtime,
finalizeRequestEntry(requestEntry, {
response_stream: true,
...result,
matched: true,
status_code: config.non_stream_status_code,
upstream_status_code: upstreamResponse.status,
}),
config.request_history_limit,
);
return;
}
recordRequestEntry(
runtime,
finalizeRequestEntry(requestEntry, {
response_stream: true,
...result,
}),
config.request_history_limit,
);
return;
}
const result = await handleNonStreaming({
runtime,
config,
logger,
monitor: runtime.monitor,
pathname,
upstreamResponse,
res,
requestEntry,
terminalRetryableUpstreamError,
});
recordRequestEntry(
runtime,
finalizeRequestEntry(requestEntry, {
response_stream: false,
...result,
}),
config.request_history_limit,
);
return;
}
} catch (error) {
if (Number.isInteger(error?.gatewayAttemptCount)) {
requestEntry.upstream_attempt_count = (requestEntry.upstream_attempt_count || 0) + error.gatewayAttemptCount;
}
recordRequestEntry(
runtime,
finalizeRequestEntry(requestEntry, {
status_code: res.headersSent ? null : 502,
error: `${error?.message || error}`,
}),
config.request_history_limit,
);
throw error;
}
}
async function main() {
const args = parseArgs(process.argv);
const configPath = args.config || path.join(__dirname, "config.json");
const config = await loadConfig(configPath);
const monitor = createMonitor();
if (args.log) {
await mkdir(path.dirname(args.log), { recursive: true });
}
const logger = createLogger(args.log, createMonitorRecorder(monitor));
const requestsDb = openRequestsDatabase(buildRuntimePaths(configPath, args.log || null).requestsDbPath);
const runtime = {
config,
configPath,
logPath: args.log || null,
logger,
monitor,
paths: buildRuntimePaths(configPath, args.log || null),
requestsDb,
server: null,
};
const importedCount = await importRequestsJsonlToDb(requestsDb, runtime.paths.requestsPath);
await hydrateMonitorFromDisk(monitor, runtime.paths, config.request_history_limit, requestsDb);
logger(`[start] hydrated persistent request metrics from db path=${runtime.paths.requestsDbPath}`);
logger(`[start] requests db ready path=${runtime.paths.requestsDbPath} imported_jsonl_rows=${importedCount}`);
const server = http.createServer(async (req, res) => {
try {
await proxyRequest(runtime, req, res);
} catch (error) {
logger(`[error] ${error?.stack || error}`);
if (!res.headersSent) {
res.writeHead(502, { "content-type": "application/json; charset=utf-8" });
res.end(
JSON.stringify({
error: {
message: `${error?.message || error}`,
type: "codex_retry_gateway_error",
code: "gateway_error",
},
}),
);
} else {
res.socket?.destroy();
}
}
});
runtime.server = server;
let shuttingDown = false;
const shutdown = (signal) => {
if (shuttingDown) {
return;
}
shuttingDown = true;
logger(`[stop] received ${signal}, closing gateway`);
server.close(() => {
process.exit(0);
});
const hardExitTimer = setTimeout(() => {
logger("[stop] forced exit after graceful shutdown timeout");
process.exit(signal === "SIGTERM" || signal === "SIGINT" ? 0 : 1);
}, 5000);
hardExitTimer.unref();
};
process.on("SIGTERM", () => shutdown("SIGTERM"));
process.on("SIGINT", () => shutdown("SIGINT"));
server.listen(config.listen_port, config.listen_host, () => {
updateRuntimeState(runtime, {
last_started_at: new Date().toISOString(),
profile_name: config.profile_name || "default",
gateway_base_url: `http://${config.listen_host}:${config.listen_port}`,
}).catch((error) => logger(`[state] failed to update runtime state: ${error?.message || error}`));
logger(
`[start] codex retry gateway profile=${config.profile_name || "default"} auth=${normalizeAuthMode(config.upstream_auth_mode)} listening on http://${config.listen_host}:${config.listen_port} -> ${config.upstream_base_url}`,
);
});
}
main().catch((error) => {
process.stderr.write(`${error?.stack || error}\n`);
process.exit(1);
});