3344 lines
108 KiB
JavaScript
3344 lines
108 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 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: 10 * 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.",
|
|
],
|
|
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",
|
|
];
|
|
|
|
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 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,
|
|
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");
|
|
}
|
|
db.exec("CREATE INDEX IF NOT EXISTS idx_requests_request_id ON requests(request_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,
|
|
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, 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, @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(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,
|
|
),
|
|
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 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,
|
|
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.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 {
|
|
upstream_auth_file,
|
|
upstream_auth_json_path,
|
|
upstream_auth_env,
|
|
upstream_auth_json_key,
|
|
model_remap_map,
|
|
...rest
|
|
} = config;
|
|
|
|
return {
|
|
...rest,
|
|
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"));
|
|
}
|
|
|
|
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,
|
|
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);
|
|
|
|
if (pathname === UI_PATH || pathname.startsWith(`${UI_PATH}/`)) {
|
|
if (!(await serveManagementUi(req, res, requestUrl.pathname))) {
|
|
jsonResponse(req, res, 503, {
|
|
error: {
|
|
message: "UI assets were not built. Run: npm run build:ui",
|
|
code: "ui_not_built",
|
|
},
|
|
});
|
|
}
|
|
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),
|
|
});
|
|
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 isExpectedStreamTermination(error) {
|
|
if (!error) {
|
|
return false;
|
|
}
|
|
if (error.name === "AbortError") {
|
|
return true;
|
|
}
|
|
return error instanceof TypeError && error.message === "terminated";
|
|
}
|
|
|
|
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,
|
|
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));
|
|
} 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 matched = reasoningMatched(config, reasoning);
|
|
const retryableUpstreamError = terminalRetryableUpstreamError || findRetryableUpstreamErrorMatch(
|
|
config,
|
|
upstreamResponse.status,
|
|
parsed,
|
|
bodyText,
|
|
);
|
|
|
|
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,
|
|
};
|
|
}
|
|
|
|
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",
|
|
};
|
|
}
|
|
|
|
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,
|
|
};
|
|
}
|
|
|
|
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)) {
|
|
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_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_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_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_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;
|
|
}
|
|
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_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.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());
|
|
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);
|
|
});
|