diff --git a/config.example.json b/config.example.json index 885dfb6..0604f0e 100644 --- a/config.example.json +++ b/config.example.json @@ -11,7 +11,8 @@ "request_body_limit_bytes": 1073741824, "request_history_limit": 200, "endpoints": ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"], - "reasoning_equals": [516], + "reasoning_match_mode": "formula_518n_minus_2", + "reasoning_equals": [516, 1034, 1552], "retryable_status_codes": [429, 503], "retryable_error_messages": [ "Selected model is at capacity. Please try a different model.", diff --git a/gateway.mjs b/gateway.mjs index 8027c46..fe767ab 100644 --- a/gateway.mjs +++ b/gateway.mjs @@ -27,6 +27,9 @@ const PROFILE_ITEM_API_PREFIX = `${ADMIN_BASE_PATH}/api/profiles/`; const RESTORE_API_PATH = `${ADMIN_BASE_PATH}/api/restore`; const STATUS_REASONING_COUNT_LIMIT = 24; const MANAGEMENT_ACCESS_COOKIE = "codex_retry_gateway_access"; +const RESPONSES_REASONING_RETRY_PATHS = new Set(["/responses", "/v1/responses"]); +const REASONING_RETRY_ABORT_CLIENT = "reasoning_retry_client_disconnected"; +const REASONING_RETRY_ABORT_WINNER = "reasoning_retry_winner_selected"; const DEFAULT_CONFIG = { profile_name: "default", @@ -42,7 +45,8 @@ const DEFAULT_CONFIG = { request_history_limit: 0, model_remap: "", endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"], - reasoning_equals: [516], + reasoning_match_mode: "formula_518n_minus_2", + reasoning_equals: [516, 1034, 1552], retryable_status_codes: [429, 503], retryable_error_messages: [ "Selected model is at capacity. Please try a different model.", @@ -85,6 +89,16 @@ const REQUEST_THREAD_ID_POINTERS = [ "/x-codex-turn-metadata/conversationId", "/previous_response_id", ]; +const REQUEST_REASONING_EFFORT_POINTERS = [ + "/reasoning/effort", + "/reasoning_effort", + "/reasoningEffort", +]; +const REQUEST_REASONING_SUMMARY_POINTERS = [ + "/reasoning/summary", + "/reasoning_summary", + "/reasoningSummary", +]; const RESPONSE_THREAD_ID_POINTERS = [ "/thread_id", "/client_metadata/thread_id", @@ -142,7 +156,7 @@ function printHelp() { "", "说明:", " 独立 Codex 本地重试网关。", - " 非流式命中 reasoning_tokens=516 时返回 502。", + " 默认按 518n-2 公式拦截 reasoning_tokens（516, 1034, 1552, ...），非流式返回 502。", " 流式命中时默认缓存并返回 502，避免半截流返回。", "", ].join("\n"), @@ -335,6 +349,14 @@ function extractRequestThreadId(payload) { return extractStringByPointers(payload, REQUEST_THREAD_ID_POINTERS); } +function extractRequestReasoningEffort(payload) { + return extractStringByPointers(payload, REQUEST_REASONING_EFFORT_POINTERS); +} + +function extractRequestReasoningSummary(payload) { + return extractStringByPointers(payload, REQUEST_REASONING_SUMMARY_POINTERS); +} + function extractResponseThreadId(payload) { return extractStringByPointers(payload, RESPONSE_THREAD_ID_POINTERS); } @@ -436,6 +458,21 @@ function mergeUsageSnapshots(current, next) { }; } +function sumUsageSnapshots(current, next) { + if (!next) { + return current || null; + } + const result = { ...(current || {}) }; + for (const key of ["input_tokens", "output_tokens", "total_tokens", "reasoning_tokens", "cached_tokens"]) { + const nextValue = next[key]; + if (!Number.isInteger(nextValue)) { + continue; + } + result[key] = (Number.isInteger(result[key]) ? result[key] : 0) + nextValue; + } + return Object.keys(result).length > 0 ? result : null; +} + function normalizeIntegerList(values, fallback = []) { const source = values === undefined || values === null ? fallback : values; const normalized = flattenValues(source) @@ -476,6 +513,19 @@ function normalizePhraseList(values, fallback = []) { return [...new Set(normalized)]; } +function normalizeReasoningMatchMode(value) { + const mode = `${value || DEFAULT_CONFIG.reasoning_match_mode}`.trim().toLowerCase(); + if (["formula_518n_minus_2", "manual"].includes(mode)) { + return mode; + } + return DEFAULT_CONFIG.reasoning_match_mode; +} + +function normalizeReasoningEquals(values, fallback = DEFAULT_CONFIG.reasoning_equals) { + const normalized = normalizeIntegerList(values, fallback); + return normalized.length > 0 ? normalized : [...fallback]; +} + function normalizePositiveInteger(value, fallback) { const parsed = Number.parseInt(`${value ?? ""}`, 10); if (Number.isInteger(parsed) && parsed > 0) { @@ -698,6 +748,8 @@ function openRequestsDatabase(dbPath) { model TEXT, requested_model TEXT, forwarded_model TEXT, + reasoning_effort TEXT, + reasoning_summary TEXT, request_stream INTEGER, response_stream INTEGER, inspected INTEGER, @@ -733,9 +785,16 @@ function openRequestsDatabase(dbPath) { if (!requestColumnNames.has("thread_id")) { db.exec("ALTER TABLE requests ADD COLUMN thread_id TEXT"); } + if (!requestColumnNames.has("reasoning_effort")) { + db.exec("ALTER TABLE requests ADD COLUMN reasoning_effort TEXT"); + } + if (!requestColumnNames.has("reasoning_summary")) { + db.exec("ALTER TABLE requests ADD COLUMN reasoning_summary TEXT"); + } db.exec("CREATE INDEX IF NOT EXISTS idx_requests_request_id ON requests(request_id)"); db.exec("CREATE INDEX IF NOT EXISTS idx_requests_response_id ON requests(response_id)"); db.exec("CREATE INDEX IF NOT EXISTS idx_requests_thread_id ON requests(thread_id)"); + db.exec("CREATE INDEX IF NOT EXISTS idx_requests_reasoning_effort ON requests(reasoning_effort)"); return db; } @@ -778,6 +837,8 @@ function requestRowFromEntry(entry) { model: entry.model || null, requested_model: entry.requested_model || null, forwarded_model: entry.forwarded_model || null, + reasoning_effort: entry.reasoning_effort || null, + reasoning_summary: entry.reasoning_summary || null, request_stream: boolToInt(Boolean(entry.request_stream)), response_stream: boolToInt(Boolean(entry.response_stream)), inspected: boolToInt(Boolean(entry.inspected)), @@ -802,13 +863,13 @@ function insertRequestRow(db, row) { db.prepare(` INSERT OR REPLACE INTO requests ( seq, request_id, response_id, thread_id, started_at, finished_at, duration_ms, profile_name, method, path, model, - requested_model, forwarded_model, request_stream, response_stream, inspected, matched, + requested_model, forwarded_model, reasoning_effort, reasoning_summary, request_stream, response_stream, inspected, matched, status_code, upstream_status_code, reasoning_tokens, input_tokens, output_tokens, total_tokens, cached_tokens, error, upstream_origin, upstream_path, upstream_auth_mode, upstream_auth_source, payload_json ) VALUES ( @seq, @request_id, @response_id, @thread_id, @started_at, @finished_at, @duration_ms, @profile_name, @method, @path, @model, - @requested_model, @forwarded_model, @request_stream, @response_stream, @inspected, @matched, + @requested_model, @forwarded_model, @reasoning_effort, @reasoning_summary, @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 @@ -873,6 +934,8 @@ function buildRequestQueryFilters({ query, filter }) { lower(coalesce(model, '')) LIKE @query OR lower(coalesce(requested_model, '')) LIKE @query OR lower(coalesce(forwarded_model, '')) LIKE @query OR + lower(coalesce(reasoning_effort, '')) LIKE @query OR + lower(coalesce(reasoning_summary, '')) LIKE @query OR lower(coalesce(error, '')) LIKE @query OR lower(coalesce(upstream_origin, '')) LIKE @query OR lower(coalesce(upstream_path, '')) LIKE @query OR @@ -947,6 +1010,15 @@ async function hydrateMonitorFromDisk(monitor, paths, requestHistoryLimit, reque } monitor.observed_reasoning_counts[`${row.reasoning_tokens}`] = row.count || 0; } + + const persistedRows = requestsDb.prepare("SELECT payload_json FROM requests").all(); + for (const row of persistedRows) { + const entry = parseRequestRowPayload(row); + if (!entry) { + continue; + } + applyPersistedRetryExtras(monitor, entry); + } return; } @@ -959,6 +1031,7 @@ async function hydrateMonitorFromDisk(monitor, paths, requestHistoryLimit, reque if (entry.inspected) { recordInspectedResponse(monitor, entry.reasoning_tokens ?? entry.usage?.reasoning_tokens ?? null, Boolean(entry.matched)); } + applyPersistedRetryExtras(monitor, entry); addTokenTotals(monitor, entry.usage); } const firstStartedAt = requestEntries @@ -1014,6 +1087,32 @@ function recordInspectedResponse(monitor, reasoning, matched) { } } +function applyPersistedRetryExtras(monitor, entry) { + const extraInspectedCount = Number.isInteger(entry?.reasoning_retry_extra_inspected_count) + ? entry.reasoning_retry_extra_inspected_count + : 0; + const extraMatchedCount = Number.isInteger(entry?.reasoning_retry_extra_matched_count) + ? entry.reasoning_retry_extra_matched_count + : 0; + monitor.inspected_response_count += extraInspectedCount; + monitor.matched_response_count += extraMatchedCount; + + const extraReasoningCounts = entry?.reasoning_retry_extra_reasoning_counts; + if (extraReasoningCounts && typeof extraReasoningCounts === "object") { + for (const [reasoningKey, count] of Object.entries(extraReasoningCounts)) { + const parsedReasoning = Number.parseInt(reasoningKey, 10); + if (!Number.isInteger(parsedReasoning) || !Number.isInteger(count) || count <= 0) { + continue; + } + monitor.observed_reasoning_counts[`${parsedReasoning}`] = ( + monitor.observed_reasoning_counts[`${parsedReasoning}`] || 0 + ) + count; + } + } + + addTokenTotals(monitor, entry?.reasoning_retry_extra_usage || null); +} + function addTokenTotals(monitor, usage) { if (!usage) { return; @@ -1029,6 +1128,7 @@ function addTokenTotals(monitor, usage) { function upsertRequestEntry(runtime, entry, { persistJsonl = false, includeUsage = false } = {}) { if (includeUsage) { addTokenTotals(runtime.monitor, entry.usage); + addTokenTotals(runtime.monitor, entry.reasoning_retry_extra_usage || null); } if (runtime.requestsDb) { insertRequestRow(runtime.requestsDb, requestRowFromEntry(entry)); @@ -1055,6 +1155,16 @@ function markAndPersistFirstResponse(runtime, entry, at = new Date()) { return true; } +function markRequestEntryFirstResponse(runtime, entry, { persistEntry = true, at = new Date() } = {}) { + if (!markRequestFirstResponse(entry, at)) { + return false; + } + if (persistEntry) { + upsertRequestEntry(runtime, entry); + } + return true; +} + function summarizeProfileAuthSource(env) { const mode = env.CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE || "passthrough"; if (mode === "auth_json") { @@ -1099,7 +1209,11 @@ function buildProfileFormModel(env) { 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 || "", + reasoning_match_mode: normalizeReasoningMatchMode(env.CODEX_RETRY_GATEWAY_REASONING_MATCH_MODE), + reasoning_equals: normalizeReasoningEquals( + env.CODEX_RETRY_GATEWAY_REASONING_EQUALS || DEFAULT_CONFIG.reasoning_equals, + DEFAULT_CONFIG.reasoning_equals, + ).join(","), 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, @@ -1114,6 +1228,7 @@ function buildProfileFormModel(env) { } function buildConfigFromProfileEnv(profileName, env) { + const reasoningMatchMode = normalizeReasoningMatchMode(env.CODEX_RETRY_GATEWAY_REASONING_MATCH_MODE); const config = { ...DEFAULT_CONFIG, profile_name: profileName, @@ -1135,7 +1250,11 @@ function buildConfigFromProfileEnv(profileName, env) { : 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), + reasoning_match_mode: reasoningMatchMode, + reasoning_equals: normalizeReasoningEquals( + 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, @@ -1237,10 +1356,8 @@ async function buildProfileEnvText(payload) { } 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 reasoningMatchMode = normalizeReasoningMatchMode(payload.reasoning_match_mode); + const reasoningEquals = normalizeReasoningEquals(payload.reasoning_equals, DEFAULT_CONFIG.reasoning_equals); const endpoints = normalizeStringList(payload.endpoints, DEFAULT_CONFIG.endpoints).map(normalizePath); if (endpoints.length === 0) { @@ -1289,6 +1406,7 @@ async function buildProfileEnvText(payload) { ["CODEX_RETRY_GATEWAY_LISTEN_PORT", `${listenPort}`], ["CODEX_RETRY_GATEWAY_UPSTREAM_BASE_URL", upstreamBaseUrl], ["CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE", authMode], + ["CODEX_RETRY_GATEWAY_REASONING_MATCH_MODE", reasoningMatchMode], ["CODEX_RETRY_GATEWAY_REASONING_EQUALS", reasoningEquals.join(",")], ["CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES", retryableStatusCodes.join(",")], ["CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES", retryableErrorMessages.join("\n")], @@ -1493,11 +1611,27 @@ function buildRequestEntry({ seq, startedAt, startedMs, req, pathname, requestJs model: requestJson?.model || null, requested_model: requestJson?.model || null, forwarded_model: requestJson?.model || null, + reasoning_effort: extractRequestReasoningEffort(requestJson), + reasoning_summary: extractRequestReasoningSummary(requestJson), request_stream: Boolean(requestJson?.stream), response_stream: false, stream_chunk_count: 0, usage_last_updated_at: null, upstream_attempt_count: 0, + reasoning_retry_enabled: false, + reasoning_retry_query_count: 0, + reasoning_retry_round_count: 0, + reasoning_retry_current_round: null, + reasoning_retry_current_width: 0, + reasoning_retry_current_firsts: [], + reasoning_retry_winner_round: null, + reasoning_retry_winner_slot: null, + reasoning_retry_stop_reason: null, + reasoning_retry_thread_mode: "disabled", + reasoning_retry_extra_inspected_count: 0, + reasoning_retry_extra_matched_count: 0, + reasoning_retry_extra_usage: null, + reasoning_retry_extra_reasoning_counts: {}, inspected: false, matched: false, status_code: null, @@ -1522,6 +1656,10 @@ function markRequestFirstResponse(entry, at = new Date()) { : null; entry.last_activity_at = firstAt.toISOString(); entry.lifecycle_state = "receive_first"; + entry._first_response_observer?.({ + first_response_at: entry.first_response_at, + first_response_delay_ms: entry.first_response_delay_ms, + }); return true; } @@ -1583,7 +1721,8 @@ async function loadConfig(configPath) { 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_match_mode = normalizeReasoningMatchMode(config.reasoning_match_mode); + config.reasoning_equals = normalizeReasoningEquals( config.reasoning_equals, DEFAULT_CONFIG.reasoning_equals, ); @@ -1790,6 +1929,7 @@ async function listProfiles(runtime) { request_history_limit: form.request_history_limit, model_remap: form.model_remap || "", auth_source: summarizeProfileAuthSource(env), + reasoning_match_mode: form.reasoning_match_mode, reasoning_equals: form.reasoning_equals, }, form, @@ -2256,7 +2396,15 @@ function renderManagementAccessPage(requestPathname, errorMessage = "") { } function buildEditableConfig(currentConfig, payload) { - const nextReasoning = normalizeIntegerList(payload.reasoning_equals, currentConfig.reasoning_equals); + const nextReasoningMatchMode = normalizeReasoningMatchMode( + payload.reasoning_match_mode === undefined + ? currentConfig.reasoning_match_mode + : payload.reasoning_match_mode, + ); + const nextReasoning = normalizeReasoningEquals( + payload.reasoning_equals === undefined ? currentConfig.reasoning_equals : payload.reasoning_equals, + currentConfig.reasoning_equals, + ); const nextRetryableStatusCodes = normalizeIntegerList( payload.retryable_status_codes, currentConfig.retryable_status_codes, @@ -2291,9 +2439,6 @@ function buildEditableConfig(currentConfig, payload) { ? 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 不能为空"); } @@ -2315,6 +2460,7 @@ function buildEditableConfig(currentConfig, payload) { return { ...currentConfig, + reasoning_match_mode: nextReasoningMatchMode, reasoning_equals: nextReasoning, retryable_status_codes: nextRetryableStatusCodes, retryable_error_messages: nextRetryableErrorMessages, @@ -2573,7 +2719,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { 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(",")}`, + `[config] updated reasoning_match_mode=${nextConfig.reasoning_match_mode} 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, { @@ -2718,6 +2864,128 @@ function copyHeadersToClient(sourceHeaders, target) { } } +function cloneResponseHeaders(sourceHeaders) { + const headers = new Headers(); + for (const [key, value] of sourceHeaders.entries()) { + headers.set(key, value); + } + return headers; +} + +function writeCapturedResponse(res, delivery) { + if (!delivery) { + throw new Error("missing captured response delivery"); + } + copyHeadersToClient(delivery.headers, res); + res.writeHead(delivery.status_code); + res.end(delivery.body); +} + +function buildCapturedDelivery(statusCode, headers, body) { + return { + status_code: statusCode, + headers: cloneResponseHeaders(headers), + body: Buffer.isBuffer(body) ? body : Buffer.from(body || ""), + }; +} + +function createAbortReason(code, message) { + const error = new Error(message); + error.name = "AbortError"; + error.code = code; + return error; +} + +function abortReasonCode(value) { + return value?.reason?.code || value?.code || null; +} + +function createLinkedAbortController(signals = []) { + const controller = new AbortController(); + const cleanups = []; + const linkSignal = (signal) => { + if (!signal) { + return; + } + const abort = () => { + if (!controller.signal.aborted) { + controller.abort(signal.reason || createAbortReason(REASONING_RETRY_ABORT_CLIENT, "request aborted")); + } + }; + if (signal.aborted) { + abort(); + return; + } + signal.addEventListener("abort", abort, { once: true }); + cleanups.push(() => signal.removeEventListener("abort", abort)); + }; + for (const signal of signals) { + linkSignal(signal); + if (controller.signal.aborted) { + break; + } + } + return { + controller, + cleanup() { + for (const cleanup of cleanups) { + cleanup(); + } + }, + }; +} + +function createClientAbortContext(req, res) { + const controller = new AbortController(); + const abort = () => { + if (!controller.signal.aborted) { + controller.abort( + createAbortReason(REASONING_RETRY_ABORT_CLIENT, "client disconnected before reasoning retry completed"), + ); + } + }; + const onAborted = () => abort(); + const onReqClose = () => { + if (req.destroyed && !res.writableEnded) { + abort(); + } + }; + const onResClose = () => { + if (!res.writableEnded) { + abort(); + } + }; + req.on("aborted", onAborted); + req.on("close", onReqClose); + res.on("close", onResClose); + return { + signal: controller.signal, + cleanup() { + req.off("aborted", onAborted); + req.off("close", onReqClose); + res.off("close", onResClose); + }, + }; +} + +function isResponsesReasoningRetryPath(pathname) { + return RESPONSES_REASONING_RETRY_PATHS.has(normalizePath(pathname)); +} + +function isResponsesReasoningRetryEligible(pathname, requestEntry) { + return isResponsesReasoningRetryPath(pathname) && Boolean(requestEntry?.thread_id); +} + +function reasoningRetryWaveWidth(round) { + if (round <= 2) { + return 1; + } + if (round <= 4) { + return 2; + } + return 4; +} + async function readRequestBody(req, limitBytes) { const chunks = []; let total = 0; @@ -2751,8 +3019,18 @@ function matchPath(config, pathname) { return config.endpoints.includes(normalizePath(pathname)); } +function reasoningMatchesFormula518nMinus2(reasoning) { + return Number.isInteger(reasoning) && reasoning >= 516 && (reasoning + 2) % 518 === 0; +} + function reasoningMatched(config, reasoning) { - return reasoning !== null && config.reasoning_equals.includes(reasoning); + if (!Number.isInteger(reasoning)) { + return false; + } + if (normalizeReasoningMatchMode(config.reasoning_match_mode) === "manual") { + return config.reasoning_equals.includes(reasoning); + } + return reasoningMatchesFormula518nMinus2(reasoning); } function collectRetryableMessageCandidates(value, state = { seen: new Set(), results: [] }, depth = 0) { @@ -3148,8 +3426,10 @@ async function handleNonStreaming({ res, requestEntry, terminalRetryableUpstreamError = null, + captureOnly = false, + persistEntry = true, }) { - markAndPersistFirstResponse(runtime, requestEntry); + markRequestEntryFirstResponse(runtime, requestEntry, { persistEntry }); const bodyBuffer = Buffer.from(await upstreamResponse.arrayBuffer()); const bodyText = bodyBuffer.toString("utf8"); const parsed = isJsonContentType(upstreamResponse.headers.get("content-type")) @@ -3180,15 +3460,18 @@ async function handleNonStreaming({ `[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); + if (!captureOnly) { + 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, + match_reason: "reasoning_guard", status_code: config.non_stream_status_code, upstream_status_code: upstreamResponse.status, reasoning_tokens: reasoning, @@ -3210,11 +3493,13 @@ async function handleNonStreaming({ 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); + if (!captureOnly) { + 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, @@ -3226,12 +3511,24 @@ async function handleNonStreaming({ match_reason: "retryable_upstream_error", response_id: requestEntry.response_id, thread_id: requestEntry.thread_id, + delivery: captureOnly + ? buildCapturedDelivery( + config.non_stream_status_code, + new Headers({ + "content-type": "application/json; charset=utf-8", + "x-codex-retry-gateway-reason": "upstream-error-retry-triggered", + }), + blockedBody, + ) + : null, }; } - copyHeadersToClient(upstreamResponse.headers, res); - res.writeHead(upstreamResponse.status); - res.end(bodyBuffer); + if (!captureOnly) { + copyHeadersToClient(upstreamResponse.headers, res); + res.writeHead(upstreamResponse.status); + res.end(bodyBuffer); + } return { inspected: true, matched, @@ -3241,6 +3538,9 @@ async function handleNonStreaming({ usage, response_id: requestEntry.response_id, thread_id: requestEntry.thread_id, + delivery: captureOnly + ? buildCapturedDelivery(upstreamResponse.status, upstreamResponse.headers, bodyBuffer) + : null, }; } @@ -3254,8 +3554,11 @@ async function handleStreaming({ res, abortController, requestEntry, + requestAbortSignal = null, + captureOnly = false, + persistEntry = true, }) { - const strict502Mode = config.stream_action !== "disconnect"; + const strict502Mode = captureOnly || config.stream_action !== "disconnect"; const reader = upstreamResponse.body.getReader(); const sseState = { decoder: new TextDecoder("utf8"), @@ -3267,7 +3570,7 @@ async function handleStreaming({ let observedUsage = null; const bufferedChunks = []; - if (!strict502Mode) { + if (!strict502Mode && !captureOnly) { copyHeadersToClient(upstreamResponse.headers, res); res.writeHead(upstreamResponse.status); } @@ -3277,6 +3580,20 @@ async function handleStreaming({ try { readResult = await reader.read(); } catch (error) { + if (requestAbortSignal?.aborted) { + const abortCode = abortReasonCode(requestAbortSignal); + if (abortCode === REASONING_RETRY_ABORT_WINNER) { + return { + cancelled: true, + cancel_reason: abortCode, + response_id: requestEntry.response_id, + thread_id: requestEntry.thread_id, + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, + }; + } + throw requestAbortSignal.reason || error; + } if (isExpectedStreamTermination(error)) { const retryableTerminationError = findRetryableStreamTerminationMatch(config, error); if (retryableTerminationError) { @@ -3297,11 +3614,16 @@ async function handleStreaming({ }; } recordInspectedResponse(monitor, observedReasoning, false); - persistStreamingProgress(runtime, requestEntry, { force: true }, new Date()); + if (persistEntry) { + 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")); + const gatewayErrorBody = buildGatewayErrorBody("upstream stream terminated before completion"); + if (!captureOnly) { + res.writeHead(502, { "content-type": "application/json; charset=utf-8" }); + res.end(gatewayErrorBody); + } return { inspected: true, matched: false, @@ -3314,9 +3636,18 @@ async function handleStreaming({ thread_id: requestEntry.thread_id, response_bytes_received: requestEntry.response_bytes_received, stream_chunk_count: requestEntry.stream_chunk_count, + delivery: captureOnly + ? buildCapturedDelivery( + 502, + new Headers({ "content-type": "application/json; charset=utf-8" }), + gatewayErrorBody, + ) + : null, }; } else { - res.end(); + if (!captureOnly) { + res.end(); + } return { inspected: true, matched: false, @@ -3338,26 +3669,48 @@ async function handleStreaming({ 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(); + if (persistEntry) { + persistStreamingProgress(runtime, requestEntry, { force: true }, new Date()); + } + if (strict502Mode) { + const bufferedBody = Buffer.concat(bufferedChunks); + if (!captureOnly) { + copyHeadersToClient(upstreamResponse.headers, res); + res.writeHead(upstreamResponse.status); + res.end(bufferedBody); + } + return { + inspected: true, + matched: false, + status_code: upstreamResponse.status, + upstream_status_code: upstreamResponse.status, + reasoning_tokens: observedReasoning, + usage: observedUsage, + response_id: requestEntry.response_id, + thread_id: requestEntry.thread_id, + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, + delivery: captureOnly + ? buildCapturedDelivery(upstreamResponse.status, upstreamResponse.headers, bufferedBody) + : null, + }; + } else { + if (!captureOnly) { + res.end(); + } + return { + inspected: true, + matched: false, + status_code: upstreamResponse.status, + upstream_status_code: upstreamResponse.status, + reasoning_tokens: observedReasoning, + usage: observedUsage, + response_id: requestEntry.response_id, + thread_id: requestEntry.thread_id, + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, + }; } - return { - inspected: true, - matched: false, - status_code: upstreamResponse.status, - upstream_status_code: upstreamResponse.status, - reasoning_tokens: observedReasoning, - usage: observedUsage, - response_id: requestEntry.response_id, - thread_id: requestEntry.thread_id, - response_bytes_received: requestEntry.response_bytes_received, - stream_chunk_count: requestEntry.stream_chunk_count, - }; } const chunkBuffer = Buffer.from(value); @@ -3384,7 +3737,7 @@ async function handleStreaming({ }; } - markAndPersistFirstResponse(runtime, requestEntry, now); + markRequestEntryFirstResponse(runtime, requestEntry, { persistEntry, at: now }); const reasoning = inspection.reasoning; const usageUpdated = Boolean(inspection.usage); observedUsage = mergeUsageSnapshots(observedUsage, inspection.usage); @@ -3403,12 +3756,14 @@ async function handleStreaming({ reasoning, at: now, }); - persistStreamingProgress( - runtime, - requestEntry, - { usageUpdated, force: requestEntry.stream_chunk_count === 1 }, - now, - ); + if (persistEntry) { + persistStreamingProgress( + runtime, + requestEntry, + { usageUpdated, force: requestEntry.stream_chunk_count === 1 }, + now, + ); + } if (reasoningMatched(config, reasoning)) { recordInspectedResponse(monitor, reasoning, true); if (config.log_match) { @@ -3417,7 +3772,10 @@ async function handleStreaming({ ); } - if (strict502Mode || !wroteAnyChunk) { + if (captureOnly) { + abortController.abort(createAbortReason(REASONING_RETRY_ABORT_WINNER, "reasoning retry wave advanced")); + reader.cancel().catch(() => {}); + } else if (strict502Mode || !wroteAnyChunk) { abortController.abort(); reader.cancel().catch(() => {}); const blockedBody = buildBlockedBody(pathname, reasoning, config.non_stream_status_code); @@ -3434,6 +3792,7 @@ async function handleStreaming({ return { inspected: true, matched: true, + match_reason: "reasoning_guard", status_code: config.non_stream_status_code, upstream_status_code: upstreamResponse.status, reasoning_tokens: reasoning, @@ -3449,18 +3808,588 @@ async function handleStreaming({ bufferedChunks.push(chunkBuffer); } else { wroteAnyChunk = true; - res.write(chunkBuffer); + if (!captureOnly) { + res.write(chunkBuffer); + } } } } +function buildAttemptRequestEntry(baseEntry) { + return { + ...baseEntry, + lifecycle_state: "sent", + first_response_at: null, + first_response_delay_ms: null, + last_activity_at: null, + finished_at: null, + duration_ms: null, + response_bytes_received: 0, + 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, + reasoning_tokens: null, + usage: null, + error: null, + _started_ms: Date.now(), + }; +} + +function classifyGatewayQueryResult(result) { + if (result?.cancelled) { + return "cancelled"; + } + if (result?.match_reason === "reasoning_guard") { + return "reasoning_retry"; + } + if ( + Number.isInteger(result?.status_code) && + result.status_code >= 200 && + result.status_code < 400 && + !result?.error + ) { + return "success"; + } + return "fatal"; +} + +async function executeGatewayQuery({ + runtime, + config, + logger, + monitor, + pathname, + req, + res, + requestBody, + requestIsStream, + upstreamUrl, + upstreamAuth, + requestEntry, + captureOnly = false, + persistEntry = true, + externalAbortSignals = [], +}) { + const maxUpstreamAttempts = normalizePositiveInteger( + config.upstream_fetch_retry_attempts, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ); + const shouldInspect = matchPath(config, pathname); + let totalUpstreamAttempts = 0; + const queryAbortLink = createLinkedAbortController(externalAbortSignals); + + try { + while (totalUpstreamAttempts < maxUpstreamAttempts) { + if (queryAbortLink.controller.signal.aborted) { + throw queryAbortLink.controller.signal.reason || createAbortReason( + REASONING_RETRY_ABORT_CLIENT, + "request aborted", + ); + } + + const upstreamAbortLink = createLinkedAbortController([queryAbortLink.controller.signal]); + try { + 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: upstreamAbortLink.controller.signal, + }, { + ...config, + upstream_fetch_retry_attempts: Math.max(1, maxUpstreamAttempts - totalUpstreamAttempts), + }, logger, { method: req.method, pathname }); + + totalUpstreamAttempts += upstreamAttemptCount; + requestEntry.upstream_attempt_count = totalUpstreamAttempts; + + 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 = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth, upstreamResponse }); + if (persistEntry) { + 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) { + markRequestEntryFirstResponse(runtime, requestEntry, { persistEntry }); + const body = Buffer.from(await upstreamResponse.arrayBuffer()); + const parsed = isJsonContentType(responseContentType) + ? parseJsonSafely(body) + : null; + requestEntry.response_id = requestEntry.response_id || (parsed ? extractNonStreamingResponseId(parsed) : null); + requestEntry.thread_id = requestEntry.thread_id || (parsed ? extractResponseThreadId(parsed) : null); + if (!captureOnly) { + copyHeadersToClient(upstreamResponse.headers, res); + res.writeHead(upstreamResponse.status); + res.end(body); + } + return { + response_stream: responseIsStream, + inspected: false, + matched: false, + status_code: upstreamResponse.status, + upstream_status_code: upstreamResponse.status, + response_id: requestEntry.response_id, + thread_id: requestEntry.thread_id, + total_upstream_attempts: totalUpstreamAttempts, + delivery: captureOnly + ? buildCapturedDelivery(upstreamResponse.status, upstreamResponse.headers, body) + : null, + }; + } + + const result = responseIsStream + ? await handleStreaming({ + runtime, + config, + logger, + monitor, + pathname, + upstreamResponse, + res, + abortController: upstreamAbortLink.controller, + requestEntry, + requestAbortSignal: queryAbortLink.controller.signal, + captureOnly, + persistEntry, + }) + : await handleNonStreaming({ + runtime, + config, + logger, + monitor, + pathname, + upstreamResponse, + res, + requestEntry, + terminalRetryableUpstreamError, + captureOnly, + persistEntry, + }); + + 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, + ); + if (!captureOnly) { + 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 { + response_stream: true, + ...result, + matched: true, + status_code: config.non_stream_status_code, + upstream_status_code: upstreamResponse.status, + total_upstream_attempts: totalUpstreamAttempts, + delivery: captureOnly + ? buildCapturedDelivery( + config.non_stream_status_code, + new Headers({ + "content-type": "application/json; charset=utf-8", + "x-codex-retry-gateway-reason": "upstream-error-retry-triggered", + }), + blockedBody, + ) + : null, + }; + } + + return { + response_stream: responseIsStream, + ...result, + total_upstream_attempts: totalUpstreamAttempts, + }; + } finally { + upstreamAbortLink.cleanup(); + } + } + } catch (error) { + const accumulatedAttempts = Number.isInteger(error?.gatewayAttemptCount) + ? totalUpstreamAttempts + error.gatewayAttemptCount + : totalUpstreamAttempts; + if (error && typeof error === "object") { + error.gatewayAttemptCount = accumulatedAttempts; + } + throw error; + } finally { + queryAbortLink.cleanup(); + } + + throw new Error("gateway query exhausted unexpectedly"); +} + +async function executeCapturedGatewayQuery(args) { + try { + const result = await executeGatewayQuery({ + ...args, + captureOnly: true, + persistEntry: false, + }); + return { + kind: classifyGatewayQueryResult(result), + result, + total_upstream_attempts: result.total_upstream_attempts || 0, + }; + } catch (error) { + const abortCode = abortReasonCode(error); + if (abortCode === REASONING_RETRY_ABORT_CLIENT) { + throw error; + } + if (abortCode === REASONING_RETRY_ABORT_WINNER) { + return { + kind: "cancelled", + result: { + cancelled: true, + cancel_reason: abortCode, + }, + total_upstream_attempts: Number.isInteger(error?.gatewayAttemptCount) ? error.gatewayAttemptCount : 0, + }; + } + + const gatewayErrorBody = buildGatewayErrorBody(`${error?.message || error}`); + return { + kind: "fatal", + result: { + inspected: false, + matched: false, + status_code: 502, + upstream_status_code: null, + error: `${error?.message || error}`, + delivery: buildCapturedDelivery( + 502, + new Headers({ "content-type": "application/json; charset=utf-8" }), + gatewayErrorBody, + ), + }, + total_upstream_attempts: Number.isInteger(error?.gatewayAttemptCount) ? error.gatewayAttemptCount : 0, + }; + } +} + +function createReasoningRetryExtraState() { + return { + inspected_count: 0, + matched_count: 0, + usage: null, + reasoning_counts: {}, + }; +} + +function accumulateReasoningRetryExtras(state, outcomes, excludedResult = null) { + for (const outcome of outcomes) { + if (outcome?.kind === "cancelled") { + continue; + } + const result = outcome?.result; + if (!result?.inspected || result === excludedResult) { + continue; + } + state.inspected_count += 1; + if (result.matched) { + state.matched_count += 1; + } + incrementReasoningCount(state.reasoning_counts, result.reasoning_tokens); + state.usage = sumUsageSnapshots(state.usage, result.usage); + } +} + +function buildReasoningRetryFirstSlots(round, width) { + return Array.from({ length: width }, (_, index) => ({ + round, + slot: index + 1, + first_response_at: null, + first_response_delay_ms: null, + outcome: "pending", + })); +} + +function updateReasoningRetryFirstSlot(requestEntry, round, slot, fields = {}) { + if (requestEntry.reasoning_retry_current_round !== round) { + return false; + } + const currentFirsts = Array.isArray(requestEntry.reasoning_retry_current_firsts) + ? requestEntry.reasoning_retry_current_firsts + : []; + const index = slot - 1; + if (index < 0 || index >= currentFirsts.length) { + return false; + } + const existing = currentFirsts[index] || { round, slot }; + currentFirsts[index] = { + ...existing, + round, + slot, + ...fields, + }; + requestEntry.reasoning_retry_current_firsts = currentFirsts; + return true; +} + +function updateReasoningRetryFirstFromAttempt(requestEntry, outcome) { + const attemptEntry = outcome?.requestEntry || {}; + const result = outcome?.result || {}; + return updateReasoningRetryFirstSlot(requestEntry, outcome.round, outcome.slot, { + first_response_at: attemptEntry.first_response_at || null, + first_response_delay_ms: Number.isInteger(attemptEntry.first_response_delay_ms) + ? attemptEntry.first_response_delay_ms + : null, + outcome: outcome.kind || "unknown", + status_code: Number.isInteger(result.status_code) ? result.status_code : null, + upstream_status_code: Number.isInteger(result.upstream_status_code) ? result.upstream_status_code : null, + reasoning_tokens: Number.isInteger(result.reasoning_tokens) ? result.reasoning_tokens : null, + matched: Boolean(result.matched), + }); +} + +function markReasoningRetryCancelledSlots(requestEntry, round, slots = []) { + let changed = false; + for (const slot of slots) { + changed = updateReasoningRetryFirstSlot(requestEntry, round, slot, { outcome: "cancelled" }) || changed; + } + return changed; +} + +async function runResponsesReasoningRetry({ + runtime, + config, + logger, + monitor, + pathname, + req, + requestBody, + requestIsStream, + upstreamUrl, + upstreamAuth, + requestEntry, + clientAbortSignal, +}) { + let round = 1; + let totalUpstreamAttempts = 0; + const retryExtraState = createReasoningRetryExtraState(); + + while (true) { + if (clientAbortSignal?.aborted) { + throw clientAbortSignal.reason || createAbortReason(REASONING_RETRY_ABORT_CLIENT, "client disconnected"); + } + + const width = reasoningRetryWaveWidth(round); + requestEntry.reasoning_retry_round_count = round; + requestEntry.reasoning_retry_current_round = round; + requestEntry.reasoning_retry_current_width = width; + requestEntry.reasoning_retry_current_firsts = buildReasoningRetryFirstSlots(round, width); + requestEntry.reasoning_retry_query_count += width; + if (requestEntry.reasoning_retry_enabled) { + upsertRequestEntry(runtime, requestEntry); + } + logger?.( + `[reasoning-retry] round=${round} width=${width} path=${pathname} thread_id=${requestEntry.thread_id}`, + ); + + const slotControllers = Array.from({ length: width }, () => new AbortController()); + let winner = null; + const attempts = slotControllers.map((slotController, index) => { + const slot = index + 1; + const attemptEntry = buildAttemptRequestEntry(requestEntry); + attemptEntry._first_response_observer = (firstFields) => { + if (updateReasoningRetryFirstSlot(requestEntry, round, slot, firstFields)) { + upsertRequestEntry(runtime, requestEntry); + } + }; + const rawPromise = executeCapturedGatewayQuery({ + runtime, + config, + logger, + monitor, + pathname, + req, + requestBody, + requestIsStream, + upstreamUrl, + upstreamAuth, + requestEntry: attemptEntry, + externalAbortSignals: [clientAbortSignal, slotController.signal], + }); + return new Promise((resolve, reject) => { + let settled = false; + const finish = (fn, value) => { + if (settled) { + return; + } + settled = true; + slotController.signal.removeEventListener("abort", onAbort); + fn(value); + }; + const onAbort = () => { + if (abortReasonCode(slotController.signal) !== REASONING_RETRY_ABORT_WINNER) { + return; + } + finish(resolve, { + kind: "cancelled", + result: { + cancelled: true, + cancel_reason: REASONING_RETRY_ABORT_WINNER, + }, + total_upstream_attempts: 0, + round, + slot, + requestEntry: attemptEntry, + }); + }; + slotController.signal.addEventListener("abort", onAbort, { once: true }); + rawPromise.then( + (outcome) => { + const attemptOutcome = { + ...outcome, + round, + slot, + requestEntry: attemptEntry, + }; + if (!winner && attemptOutcome.kind === "success") { + winner = attemptOutcome; + const cancelledSlots = []; + slotControllers.forEach((otherController, otherIndex) => { + const otherSlot = otherIndex + 1; + if (otherController !== slotController && !otherController.signal.aborted) { + cancelledSlots.push(otherSlot); + otherController.abort( + createAbortReason(REASONING_RETRY_ABORT_WINNER, "reasoning retry winner selected"), + ); + } + }); + if (cancelledSlots.length > 0) { + markReasoningRetryCancelledSlots(requestEntry, round, cancelledSlots); + upsertRequestEntry(runtime, requestEntry); + } + } + finish(resolve, attemptOutcome); + }, + (error) => finish(reject, error), + ); + }); + }); + + const outcomes = await Promise.all(attempts); + totalUpstreamAttempts += outcomes.reduce((sum, outcome) => { + return sum + (Number.isInteger(outcome.total_upstream_attempts) ? outcome.total_upstream_attempts : 0); + }, 0); + const nonCancelledOutcomes = outcomes.filter((outcome) => outcome.kind !== "cancelled"); + for (const outcome of outcomes) { + updateReasoningRetryFirstFromAttempt(requestEntry, outcome); + } + if (requestEntry.reasoning_retry_enabled) { + upsertRequestEntry(runtime, requestEntry); + } + + if (winner) { + accumulateReasoningRetryExtras(retryExtraState, nonCancelledOutcomes, winner.result); + requestEntry.reasoning_retry_extra_inspected_count = retryExtraState.inspected_count; + requestEntry.reasoning_retry_extra_matched_count = retryExtraState.matched_count; + requestEntry.reasoning_retry_extra_usage = retryExtraState.usage; + requestEntry.reasoning_retry_extra_reasoning_counts = retryExtraState.reasoning_counts; + requestEntry.reasoning_retry_winner_round = winner.round; + requestEntry.reasoning_retry_winner_slot = winner.slot; + requestEntry.reasoning_retry_stop_reason = "success"; + return { + ...winner.result, + response_stream: winner.result.response_stream, + total_upstream_attempts: totalUpstreamAttempts, + }; + } + + const allReasoningRetry = nonCancelledOutcomes.length > 0 && nonCancelledOutcomes.every( + (outcome) => outcome.kind === "reasoning_retry", + ); + if (allReasoningRetry) { + accumulateReasoningRetryExtras(retryExtraState, nonCancelledOutcomes); + round += 1; + continue; + } + + const fatalOutcome = nonCancelledOutcomes.find((outcome) => outcome.kind === "fatal"); + if (fatalOutcome) { + accumulateReasoningRetryExtras(retryExtraState, nonCancelledOutcomes, fatalOutcome.result); + requestEntry.reasoning_retry_extra_inspected_count = retryExtraState.inspected_count; + requestEntry.reasoning_retry_extra_matched_count = retryExtraState.matched_count; + requestEntry.reasoning_retry_extra_usage = retryExtraState.usage; + requestEntry.reasoning_retry_extra_reasoning_counts = retryExtraState.reasoning_counts; + requestEntry.reasoning_retry_stop_reason = fatalOutcome.result?.match_reason || "fatal"; + return { + ...fatalOutcome.result, + response_stream: fatalOutcome.result.response_stream, + total_upstream_attempts: totalUpstreamAttempts, + }; + } + + accumulateReasoningRetryExtras(retryExtraState, nonCancelledOutcomes); + requestEntry.reasoning_retry_extra_inspected_count = retryExtraState.inspected_count; + requestEntry.reasoning_retry_extra_matched_count = retryExtraState.matched_count; + requestEntry.reasoning_retry_extra_usage = retryExtraState.usage; + requestEntry.reasoning_retry_extra_reasoning_counts = retryExtraState.reasoning_counts; + requestEntry.reasoning_retry_stop_reason = "exhausted_without_winner"; + return { + inspected: false, + matched: false, + status_code: 502, + upstream_status_code: null, + error: "reasoning retry ended without a successful response", + response_stream: requestIsStream, + total_upstream_attempts: totalUpstreamAttempts, + delivery: buildCapturedDelivery( + 502, + new Headers({ "content-type": "application/json; charset=utf-8" }), + buildGatewayErrorBody("reasoning retry ended without a successful response"), + ), + }; + } +} + 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"}`); @@ -3511,7 +4440,6 @@ async function proxyRequest(runtime, req, res) { const { requestJson, remapped, forwardedModel } = remapRequestModel(config, parsedRequestJson); const requestBody = remapped ? Buffer.from(JSON.stringify(requestJson)) : rawRequestBody; const requestIsStream = Boolean(requestJson?.stream); - let totalUpstreamAttempts = 0; requestEntry.request_body_bytes = rawRequestBody.length; requestEntry.request_id = computeRequestId(pathname, rawRequestBody); requestEntry.thread_id = firstNonEmptyString( @@ -3522,7 +4450,16 @@ async function proxyRequest(runtime, req, res) { requestEntry.requested_model = parsedRequestJson?.model || null; requestEntry.forwarded_model = forwardedModel || parsedRequestJson?.model || null; requestEntry.model = requestEntry.forwarded_model; + requestEntry.reasoning_effort = extractRequestReasoningEffort(requestJson); + requestEntry.reasoning_summary = extractRequestReasoningSummary(requestJson); requestEntry.request_stream = requestIsStream; + requestEntry.reasoning_retry_enabled = isResponsesReasoningRetryPath(pathname); + requestEntry.reasoning_retry_thread_mode = requestEntry.reasoning_retry_enabled + ? (requestEntry.thread_id ? "thread_id" : "missing_thread_id") + : "disabled"; + if (requestEntry.reasoning_retry_thread_mode === "missing_thread_id") { + requestEntry.reasoning_retry_stop_reason = "missing_thread_id"; + } upsertRequestEntry(runtime, requestEntry); const upstreamUrl = buildUpstreamUrl(config.upstream_base_url, incomingUrl); @@ -3534,158 +4471,67 @@ async function proxyRequest(runtime, req, res) { `[model-remap] profile=${config.profile_name || "default"} requested=${requestEntry.requested_model} forwarded=${requestEntry.forwarded_model}`, ); } - - while (totalUpstreamAttempts < maxUpstreamAttempts) { - const abortController = new AbortController(); - const { - response: upstreamResponse, - attempt_count: upstreamAttemptCount, - retryable_upstream_error: terminalRetryableUpstreamError, - } = await fetchUpstreamWithRetry(upstreamUrl, { - method: req.method, - headers: cloneHeadersForUpstream(req.headers, upstreamAuth), - body: requestBody.length > 0 ? requestBody : undefined, - signal: abortController.signal, - }, { - ...config, - upstream_fetch_retry_attempts: Math.max(1, maxUpstreamAttempts - totalUpstreamAttempts), - }, logger, { method: req.method, pathname }); - - totalUpstreamAttempts += upstreamAttemptCount; - - const shouldInspect = matchPath(config, pathname); - const responseContentType = upstreamResponse.headers.get("content-type"); - const responseIsStream = isSseContentType(responseContentType) || ( - requestIsStream && - !isJsonContentType(responseContentType) && - !isUpstreamErrorStatus(upstreamResponse.status) - ); - requestEntry.response_stream = responseIsStream; - requestEntry.inspected = shouldInspect; - requestEntry.upstream_status_code = upstreamResponse.status; - requestEntry.upstream_attempt_count = totalUpstreamAttempts; - requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth, upstreamResponse }); - upsertRequestEntry(runtime, requestEntry); - if (isUpstreamErrorStatus(upstreamResponse.status)) { - logger?.( - `[upstream] status=${upstreamResponse.status} profile=${config.profile_name || "default"} path=${requestEntry.upstream.path} auth=${requestEntry.upstream.auth_mode}/${requestEntry.upstream.auth_source} content_type=${requestEntry.upstream.content_type || "-"}`, - ); - } - - if (!shouldInspect) { - markRequestFirstResponse(requestEntry); - upsertRequestEntry(runtime, requestEntry); - copyHeadersToClient(upstreamResponse.headers, res); - res.writeHead(upstreamResponse.status); - const body = Buffer.from(await upstreamResponse.arrayBuffer()); - const parsed = isJsonContentType(upstreamResponse.headers.get("content-type")) - ? parseJsonSafely(body) - : null; - requestEntry.response_id = requestEntry.response_id || (parsed ? extractNonStreamingResponseId(parsed) : null); - requestEntry.thread_id = requestEntry.thread_id || (parsed ? extractResponseThreadId(parsed) : null); - res.end(body); - recordRequestEntry( - runtime, - finalizeRequestEntry(requestEntry, { - status_code: upstreamResponse.status, - upstream_status_code: upstreamResponse.status, - inspected: false, - }), - config.request_history_limit, - ); - return; - } - - if (responseIsStream) { - const result = await handleStreaming({ + const clientAbortContext = createClientAbortContext(req, res); + try { + const result = isResponsesReasoningRetryEligible(pathname, requestEntry) + ? await runResponsesReasoningRetry({ runtime, config, logger, monitor: runtime.monitor, pathname, - upstreamResponse, - res, - abortController, + req, + requestBody, + requestIsStream, + upstreamUrl, + upstreamAuth, 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( + clientAbortSignal: clientAbortContext.signal, + }) + : await executeGatewayQuery({ runtime, - finalizeRequestEntry(requestEntry, { - response_stream: true, - ...result, - }), - config.request_history_limit, - ); - return; - } + config, + logger, + monitor: runtime.monitor, + pathname, + req, + res, + requestBody, + requestIsStream, + upstreamUrl, + upstreamAuth, + requestEntry, + externalAbortSignals: [clientAbortContext.signal], + }); - const result = await handleNonStreaming({ - runtime, - config, - logger, - monitor: runtime.monitor, - pathname, - upstreamResponse, - res, - requestEntry, - terminalRetryableUpstreamError, - }); + const { delivery, total_upstream_attempts: observedUpstreamAttempts, ...resultFields } = result; + if (delivery && !res.headersSent) { + writeCapturedResponse(res, delivery); + } + if (Number.isInteger(observedUpstreamAttempts)) { + requestEntry.upstream_attempt_count = observedUpstreamAttempts; + } + if (requestEntry.reasoning_retry_enabled && !requestEntry.reasoning_retry_stop_reason) { + requestEntry.reasoning_retry_stop_reason = "completed_without_retry"; + } recordRequestEntry( runtime, finalizeRequestEntry(requestEntry, { - response_stream: false, - ...result, + response_stream: Boolean(result.response_stream), + ...resultFields, }), config.request_history_limit, ); return; + } finally { + clientAbortContext.cleanup(); } } catch (error) { if (Number.isInteger(error?.gatewayAttemptCount)) { - requestEntry.upstream_attempt_count = (requestEntry.upstream_attempt_count || 0) + error.gatewayAttemptCount; + requestEntry.upstream_attempt_count = Math.max( + requestEntry.upstream_attempt_count || 0, + error.gatewayAttemptCount, + ); } recordRequestEntry( runtime, @@ -3775,7 +4621,7 @@ async function main() { 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}`, + `[start] codex retry gateway profile=${config.profile_name || "default"} auth=${normalizeAuthMode(config.upstream_auth_mode)} reasoning_match_mode=${normalizeReasoningMatchMode(config.reasoning_match_mode)} listening on http://${config.listen_host}:${config.listen_port} -> ${config.upstream_base_url}`, ); }); } diff --git a/scripts/admin-lib.mjs b/scripts/admin-lib.mjs index 4137a2e..64b95f3 100644 --- a/scripts/admin-lib.mjs +++ b/scripts/admin-lib.mjs @@ -9,10 +9,12 @@ import path from "node:path"; export const DEFAULT_STATE_ROOT = path.join(os.homedir(), ".codex-retry-gateway"); const DEFAULT_REQUEST_BODY_LIMIT_BYTES = 1024 * 1024 * 1024; const LEGACY_DEFAULT_REQUEST_BODY_LIMIT_BYTES = 10 * 1024 * 1024; -export const DEFAULT_CODEX_CONFIG_PATH = path.join(os.homedir(), ".codex", "config.toml"); -export const DEFAULT_LISTEN_HOST = "127.0.0.1"; -export const DEFAULT_LISTEN_PORT = 4610; -export const DEFAULT_HEALTH_PATH = "/__codex_retry_gateway/health"; +export const DEFAULT_CODEX_CONFIG_PATH = path.join(os.homedir(), ".codex", "config.toml"); +export const DEFAULT_LISTEN_HOST = "127.0.0.1"; +export const DEFAULT_LISTEN_PORT = 4610; +export const DEFAULT_HEALTH_PATH = "/__codex_retry_gateway/health"; +export const DEFAULT_REASONING_MATCH_MODE = "formula_518n_minus_2"; +export const DEFAULT_REASONING_EQUALS = [516, 1034, 1552]; function escapeRegExp(value) { return `${value}`.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); @@ -167,10 +169,10 @@ export async function setCodexProviderBaseUrl({ codexConfigPath, providerName, n await writeUtf8File(codexConfigPath, updatedContent); } -export function normalizeIntArray(values, fallback = [516]) { - const source = values === undefined || values === null ? fallback : values; - const queue = Array.isArray(source) ? source.flat(Infinity) : [source]; - const normalized = queue +export function normalizeIntArray(values, fallback = DEFAULT_REASONING_EQUALS) { + const source = values === undefined || values === null ? fallback : values; + const queue = Array.isArray(source) ? source.flat(Infinity) : [source]; + const normalized = queue .map((value) => (typeof value === "string" ? value.split(/[\s,]+/).filter(Boolean) : [value])) .flat() .map((value) => Number.parseInt(`${value}`, 10)) @@ -190,6 +192,14 @@ export function normalizeStringArray(values, fallback = []) { return normalized.length > 0 ? [...new Set(normalized)] : [...fallback]; } +export function normalizeReasoningMatchMode(value) { + const mode = `${value || DEFAULT_REASONING_MATCH_MODE}`.trim().toLowerCase(); + if (["formula_518n_minus_2", "manual"].includes(mode)) { + return mode; + } + return DEFAULT_REASONING_MATCH_MODE; +} + export function normalizePhraseArray(values, fallback = []) { const source = values === undefined || values === null ? fallback : values; const queue = Array.isArray(source) ? source.flat(Infinity) : [source]; @@ -430,7 +440,8 @@ export async function installForCurrentProvider({ : Number.parseInt(`${existingGatewayConfig.request_body_limit_bytes}`, 10) ), endpoints: mergedEndpoints, - reasoning_equals: normalizeIntArray(existingGatewayConfig?.reasoning_equals, [516]), + reasoning_match_mode: normalizeReasoningMatchMode(existingGatewayConfig?.reasoning_match_mode), + reasoning_equals: normalizeIntArray(existingGatewayConfig?.reasoning_equals, DEFAULT_REASONING_EQUALS), retryable_status_codes: normalizeIntArray(existingGatewayConfig?.retryable_status_codes, [429, 503]), retryable_error_messages: normalizePhraseArray(existingGatewayConfig?.retryable_error_messages, [ "Selected model is at capacity. Please try a different model.", diff --git a/scripts/common.ps1 b/scripts/common.ps1 index dc1cd67..5fbe099 100644 --- a/scripts/common.ps1 +++ b/scripts/common.ps1 @@ -221,7 +221,7 @@ function Wait-GatewayHealth { function Normalize-IntArray { param( $Values, - [int[]]$Default = @(516) + [int[]]$Default = @(516, 1034, 1552) ) if ($null -eq $Values) { diff --git a/scripts/install-for-current-provider.ps1 b/scripts/install-for-current-provider.ps1 index 7a48352..1b8c641 100644 --- a/scripts/install-for-current-provider.ps1 +++ b/scripts/install-for-current-provider.ps1 @@ -62,8 +62,13 @@ $gatewayConfig = [ordered]@{ if ([int]$existingGatewayConfig.request_body_limit_bytes -eq 10485760) { 1073741824 } else { [int]$existingGatewayConfig.request_body_limit_bytes } } else { 1073741824 } endpoints = @($mergedEndpoints) - reasoning_equals = Normalize-IntArray -Values $(if ($existingGatewayConfig) { $existingGatewayConfig.reasoning_equals } else { $null }) -Default @(516) - non_stream_status_code = if ($existingGatewayConfig -and $null -ne $existingGatewayConfig.non_stream_status_code) { [int]$existingGatewayConfig.non_stream_status_code } else { 502 } + reasoning_match_mode = if ( + $existingGatewayConfig -and + $existingGatewayConfig.reasoning_match_mode -and + @("formula_518n_minus_2", "manual") -contains ([string]$existingGatewayConfig.reasoning_match_mode) + ) { [string]$existingGatewayConfig.reasoning_match_mode } else { "formula_518n_minus_2" } + reasoning_equals = Normalize-IntArray -Values $(if ($existingGatewayConfig) { $existingGatewayConfig.reasoning_equals } else { $null }) -Default @(516, 1034, 1552) + non_stream_status_code = if ($existingGatewayConfig -and $null -ne $existingGatewayConfig.non_stream_status_code) { [int]$existingGatewayConfig.non_stream_status_code } else { 502 } stream_action = if ($existingGatewayConfig -and -not [string]::IsNullOrWhiteSpace([string]$existingGatewayConfig.stream_action)) { [string]$existingGatewayConfig.stream_action } else { "strict_502" } log_match = if ($existingGatewayConfig -and $null -ne $existingGatewayConfig.log_match) { [bool]$existingGatewayConfig.log_match } else { $true } health_path = if ($existingGatewayConfig -and -not [string]::IsNullOrWhiteSpace([string]$existingGatewayConfig.health_path)) { [string]$existingGatewayConfig.health_path } else { "/__codex_retry_gateway/health" } diff --git a/scripts/run-profile.mjs b/scripts/run-profile.mjs index 6d7b815..cdf5230 100644 --- a/scripts/run-profile.mjs +++ b/scripts/run-profile.mjs @@ -11,6 +11,8 @@ import { DEFAULT_HEALTH_PATH, DEFAULT_LISTEN_HOST, DEFAULT_LISTEN_PORT, + DEFAULT_REASONING_EQUALS, + DEFAULT_REASONING_MATCH_MODE, DEFAULT_STATE_ROOT, ensureDirectory, getCodexProviderContext, @@ -18,6 +20,7 @@ import { getGatewayStatePaths, normalizeIntArray, normalizePhraseArray, + normalizeReasoningMatchMode, normalizeStringArray, parseOptions, readJsonFile, @@ -183,6 +186,11 @@ function buildProfileConfig({ profileName, profileEnv, existingGatewayConfig, pr } const profileAuthConfig = buildProfileAuthConfig(profileEnv); + const reasoningMatchMode = normalizeReasoningMatchMode( + profileEnv.CODEX_RETRY_GATEWAY_REASONING_MATCH_MODE || + existingGatewayConfig?.reasoning_match_mode || + DEFAULT_REASONING_MATCH_MODE, + ); return { profile_name: profileName, @@ -203,9 +211,10 @@ function buildProfileConfig({ profileName, profileEnv, existingGatewayConfig, pr profileEnv.CODEX_RETRY_GATEWAY_ENDPOINTS || existingGatewayConfig?.endpoints, ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"], ), + reasoning_match_mode: reasoningMatchMode, reasoning_equals: normalizeIntArray( profileEnv.CODEX_RETRY_GATEWAY_REASONING_EQUALS || existingGatewayConfig?.reasoning_equals, - [516], + DEFAULT_REASONING_EQUALS, ), retryable_status_codes: normalizeIntArray( profileEnv.CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES || existingGatewayConfig?.retryable_status_codes, diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index bf70d45..74f782a 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/scripts/test-gateway-e2e.mjs @@ -43,13 +43,14 @@ function createJsonResponse(res, statusCode, body, extraHeaders = {}) { res.end(JSON.stringify(body)); } -function createSseResponse(res, chunks, intervalMs = 20) { +function createSseResponse(res, chunks, intervalMs = 20, options = {}) { res.writeHead(200, { "content-type": "text/event-stream; charset=utf-8", "cache-control": "no-cache", connection: "keep-alive", - "x-upstream-test": "sse", - }); + "x-upstream-test": "sse", + ...(options.headers || {}), + }); let index = 0; const timer = setInterval(() => { @@ -67,12 +68,13 @@ function createSseResponse(res, chunks, intervalMs = 20) { }); } -function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) { +function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20, options = {}) { res.writeHead(200, { "content-type": "text/event-stream; charset=utf-8", "cache-control": "no-cache", connection: "keep-alive", "x-upstream-test": "sse-terminated", + ...(options.headers || {}), }); for (const chunk of chunks) { @@ -120,12 +122,57 @@ function createCapacityErrorSseResponse( `data: ${JSON.stringify(payload)}\n\n`, ], intervalMs, + { headers: options.headers || {} }, ); } +function reasoningRetryKeyForRequest(url, parsed) { + return [ + url, + parsed.stream ? "stream" : "non-stream", + parsed.test_reasoning_retry_key || parsed.thread_id || "missing-thread", + ].join(":"); +} + +function beginReasoningRetryTrackedRequest(statsMap, key, res) { + const stats = statsMap.get(key) || { + totalRequests: 0, + activeRequests: 0, + maxConcurrent: 0, + cancelledRequests: 0, + }; + stats.totalRequests += 1; + stats.activeRequests += 1; + stats.maxConcurrent = Math.max(stats.maxConcurrent, stats.activeRequests); + statsMap.set(key, stats); + + let finished = false; + const finish = (cancelled = false) => { + if (finished) { + return; + } + finished = true; + stats.activeRequests = Math.max(0, stats.activeRequests - 1); + if (cancelled) { + stats.cancelledRequests += 1; + } + }; + + res.on("close", () => { + finish(!res.writableEnded); + }); + + return { + stats, + finish, + }; +} + function startFakeUpstream(port) { const failBeforeResponseCounts = new Map(); const capacityBeforeSuccessCounts = new Map(); + const reasoningBeforeSuccessCounts = new Map(); + const reasoningRetryStats = new Map(); const server = http.createServer((req, res) => { const responsePaths = new Set(["/responses", "/v1/responses"]); const chatCompletionPaths = new Set(["/chat/completions", "/v1/chat/completions"]); @@ -148,10 +195,25 @@ function startFakeUpstream(port) { req.setEncoding("utf8"); req.on("data", (chunk) => { body += chunk; - }); + }); req.on("end", () => { const parsed = JSON.parse(body || "{}"); - const reasoning = parsed.test_reasoning_tokens ?? 128; + let reasoning = parsed.test_reasoning_tokens ?? 128; + let reasoningAttempt = null; + const reasoningRetryKey = Number.isInteger(parsed.test_reasoning_before_success_times) + ? reasoningRetryKeyForRequest(req.url, parsed) + : null; + const reasoningRetryTracker = reasoningRetryKey + ? beginReasoningRetryTrackedRequest(reasoningRetryStats, reasoningRetryKey, res) + : null; + if (Number.isInteger(parsed.test_reasoning_before_success_times) && reasoningRetryKey) { + const currentCount = (reasoningBeforeSuccessCounts.get(reasoningRetryKey) || 0) + 1; + reasoningBeforeSuccessCounts.set(reasoningRetryKey, currentCount); + reasoningAttempt = currentCount; + reasoning = currentCount <= parsed.test_reasoning_before_success_times + ? 516 + : (parsed.test_reasoning_success_tokens ?? 128); + } if (parsed.test_fail_before_response_once) { const failKey = `${req.url}:fail-before-response-once`; const failCount = (failBeforeResponseCounts.get(failKey) || 0) + 1; @@ -236,32 +298,50 @@ function startFakeUpstream(port) { } if (parsed.stream) { createSseResponse(res, [ - `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "hello", response_id: "resp_stream", thread_id: parsed.thread_id || "thread_stream" })}\n\n`, + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "hello", response_id: "resp_stream", thread_id: parsed.thread_id || "thread_stream", retry_attempt: reasoningAttempt })}\n\n`, `data: {"response":{"usage":{"output_tokens_details":{"reasoning_tokens":${reasoning}}}}}\n\n`, "data: [DONE]\n\n", - ], parsed.test_stream_chunk_delay_ms ?? 20); + ], parsed.test_reasoning_response_delay_ms ?? parsed.test_stream_chunk_delay_ms ?? 20, { + headers: reasoningAttempt + ? { "x-upstream-reasoning-attempt": `${reasoningAttempt}` } + : {}, + }); return; } - createJsonResponse( - res, - 200, - { - id: "resp_test", - thread_id: parsed.thread_id || "thread_test", - retry_attempt: parsed.test_fail_before_response_once - ? failBeforeResponseCounts.get(`${req.url}:fail-before-response-once`) || 0 - : 0, - usage: { - output_tokens_details: { - reasoning_tokens: reasoning, + const sendJsonResponse = () => { + if (res.writableEnded || res.destroyed) { + reasoningRetryTracker?.finish(true); + return; + } + createJsonResponse( + res, + 200, + { + id: "resp_test", + thread_id: parsed.thread_id || "thread_test", + retry_attempt: parsed.test_fail_before_response_once + ? failBeforeResponseCounts.get(`${req.url}:fail-before-response-once`) || 0 + : reasoningAttempt || 0, + usage: { + output_tokens_details: { + reasoning_tokens: reasoning, + }, }, }, - }, - { "x-upstream-test": `responses-${reasoning}` }, - ); - }); - return; - } + { + "x-upstream-test": `responses-${reasoning}`, + ...(reasoningAttempt ? { "x-upstream-reasoning-attempt": `${reasoningAttempt}` } : {}), + }, + ); + }; + if (parsed.test_reasoning_response_delay_ms) { + setTimeout(sendJsonResponse, parsed.test_reasoning_response_delay_ms); + return; + } + sendJsonResponse(); + }); + return; + } if (req.method === "POST" && chatCompletionPaths.has(req.url)) { let body = ""; @@ -293,11 +373,21 @@ function startFakeUpstream(port) { createJsonResponse(res, 404, { error: "not found" }); }); - return new Promise((resolve, reject) => { - server.once("error", reject); - server.listen(port, "127.0.0.1", () => resolve(server)); - }); -} + return new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(port, "127.0.0.1", () => { + server.getReasoningRetryStat = (key) => { + return reasoningRetryStats.get(key) || { + totalRequests: 0, + activeRequests: 0, + maxConcurrent: 0, + cancelledRequests: 0, + }; + }; + resolve(server); + }); + }); +} async function waitForHealth(url, timeoutMs = 5000) { const startedAt = Date.now(); @@ -386,7 +476,8 @@ async function run() { upstream_base_url: `http://127.0.0.1:${upstreamPort}`, request_body_limit_bytes: 1024 * 1024 * 1024, endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"], - reasoning_equals: [516], + reasoning_match_mode: "formula_518n_minus_2", + reasoning_equals: [516, 1034, 1552], retryable_status_codes: [429, 503], retryable_error_messages: [ "Selected model is at capacity. Please try a different model.", @@ -450,9 +541,9 @@ async function run() { "/v1/models 未保留上游头", ); - for (const responsePath of ["/responses", "/v1/responses"]) { - const blockedResponse = await fetch(`http://127.0.0.1:${gatewayPort}${responsePath}`, { - method: "POST", + for (const responsePath of ["/responses", "/v1/responses"]) { + const blockedResponse = await fetch(`http://127.0.0.1:${gatewayPort}${responsePath}`, { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ test_reasoning_tokens: 516 }), }); @@ -473,10 +564,145 @@ async function run() { assert(okResponse.headers.get("x-upstream-test") === "responses-128", `${responsePath} 128 未保留头`); assert( okBody?.usage?.output_tokens_details?.reasoning_tokens === 128, - `${responsePath} 128 返回体异常`, - ); - } - + `${responsePath} 128 返回体异常`, + ); + } + + const blockedFormulaResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ test_reasoning_tokens: 2070 }), + }); + const blockedFormulaBody = await blockedFormulaResponse.json(); + assert(blockedFormulaResponse.status === 502, `/responses 2070 未按 518n-2 返回 502: ${blockedFormulaResponse.status}`); + assert( + blockedFormulaBody?.error?.code === "reasoning_guard_triggered", + "/responses 2070 返回体不正确", + ); + + const missingThreadRetryKey = reasoningRetryKeyForRequest("/responses", { + stream: false, + test_reasoning_retry_key: "missing-thread-fallback", + }); + const missingThreadRetryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + test_reasoning_before_success_times: 1, + test_reasoning_retry_key: "missing-thread-fallback", + }), + }); + const missingThreadRetryBody = await missingThreadRetryResponse.json(); + assert(missingThreadRetryResponse.status === 502, `无 thread_id 的 responses 请求不应自动重打: ${missingThreadRetryResponse.status}`); + assert( + missingThreadRetryBody?.error?.code === "reasoning_guard_triggered", + "无 thread_id 的 responses 请求返回体异常", + ); + const missingThreadRetryStats = upstream.getReasoningRetryStat(missingThreadRetryKey); + assert(missingThreadRetryStats.totalRequests === 1, "无 thread_id 的 responses 请求不应启动多轮重打"); + + const retryRound2ThreadId = "thread_retry_round2"; + const retryRound2Response = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + thread_id: retryRound2ThreadId, + test_reasoning_before_success_times: 1, + test_reasoning_retry_key: "round2", + }), + }); + const retryRound2Body = await retryRound2Response.json(); + assert(retryRound2Response.status === 200, `1,1 重打未恢复: ${retryRound2Response.status}`); + assert(retryRound2Body?.usage?.output_tokens_details?.reasoning_tokens === 128, "1,1 重打恢复后的 reasoning_tokens 异常"); + assert(retryRound2Response.headers.get("x-upstream-reasoning-attempt") === "2", "1,1 重打未命中第二次上游请求"); + const retryRound2EntryResponse = await fetch( + `http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(retryRound2ThreadId)}`, + { headers: adminHeaders }, + ); + const retryRound2EntryPayload = await retryRound2EntryResponse.json(); + const retryRound2Entry = (retryRound2EntryPayload?.entries || []).find((entry) => entry.thread_id === retryRound2ThreadId); + assert(retryRound2Entry?.reasoning_retry_query_count === 2, "1,1 重打未记录两次 query"); + assert(retryRound2Entry?.reasoning_retry_round_count === 2, "1,1 重打未记录两轮"); + assert(retryRound2Entry?.reasoning_retry_current_round === 2, "1,1 重打未记录当前轮次"); + assert(retryRound2Entry?.reasoning_retry_current_width === 1, "1,1 重打未记录当前轮并行数"); + assert( + Array.isArray(retryRound2Entry?.reasoning_retry_current_firsts) && + retryRound2Entry.reasoning_retry_current_firsts.length === 1 && + Number.isInteger(retryRound2Entry.reasoning_retry_current_firsts[0]?.first_response_delay_ms) && + retryRound2Entry.reasoning_retry_current_firsts.every((first) => first?.outcome !== "pending"), + "1,1 重打未记录当前轮 first 列表", + ); + assert(retryRound2Entry?.reasoning_retry_winner_round === 2, "1,1 重打赢家轮次异常"); + assert(retryRound2Entry?.reasoning_retry_winner_slot === 1, "1,1 重打赢家槽位异常"); + assert(retryRound2Entry?.reasoning_retry_stop_reason === "success", "1,1 重打 stop reason 异常"); + + const retryWave2ThreadId = "thread_retry_wave2"; + const retryWave2Key = reasoningRetryKeyForRequest("/responses", { + stream: false, + thread_id: retryWave2ThreadId, + test_reasoning_retry_key: "wave2", + }); + const retryWave2Response = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + thread_id: retryWave2ThreadId, + test_reasoning_before_success_times: 3, + test_reasoning_retry_key: "wave2", + test_reasoning_response_delay_ms: 80, + }), + }); + const retryWave2Body = await retryWave2Response.json(); + assert(retryWave2Response.status === 200, `1,1,2 重打未恢复: ${retryWave2Response.status}`); + assert(retryWave2Body?.usage?.output_tokens_details?.reasoning_tokens === 128, "1,1,2 重打恢复后的 reasoning_tokens 异常"); + assert(retryWave2Response.headers.get("x-upstream-reasoning-attempt") === "4", "1,1,2 重打未命中第四次上游请求"); + const retryWave2Stats = upstream.getReasoningRetryStat(retryWave2Key); + assert(retryWave2Stats.totalRequests === 4, "1,1,2 重打总请求数异常"); + assert(retryWave2Stats.maxConcurrent >= 2, "1,1,2 重打未出现第二轮并行"); + const retryWave2EntryResponse = await fetch( + `http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(retryWave2ThreadId)}`, + { headers: adminHeaders }, + ); + const retryWave2EntryPayload = await retryWave2EntryResponse.json(); + const retryWave2Entry = (retryWave2EntryPayload?.entries || []).find((entry) => entry.thread_id === retryWave2ThreadId); + assert(retryWave2Entry?.reasoning_retry_query_count === 4, "1,1,2 重打未记录四次 query"); + assert(retryWave2Entry?.reasoning_retry_round_count === 3, "1,1,2 重打未记录三轮"); + assert(retryWave2Entry?.reasoning_retry_current_round === 3, "1,1,2 重打未记录当前轮次"); + assert(retryWave2Entry?.reasoning_retry_current_width === 2, "1,1,2 重打未记录当前轮并行数"); + assert( + Array.isArray(retryWave2Entry?.reasoning_retry_current_firsts) && + retryWave2Entry.reasoning_retry_current_firsts.length === 2 && + retryWave2Entry.reasoning_retry_current_firsts.some((first) => Number.isInteger(first?.first_response_delay_ms)) && + retryWave2Entry.reasoning_retry_current_firsts.every((first) => first?.outcome !== "pending"), + "1,1,2 重打未记录当前两请求 first 列表", + ); + assert(retryWave2Entry?.reasoning_retry_winner_round === 3, "1,1,2 重打赢家轮次异常"); + assert([1, 2].includes(retryWave2Entry?.reasoning_retry_winner_slot), "1,1,2 重打赢家槽位异常"); + assert(retryWave2Entry?.reasoning_retry_stop_reason === "success", "1,1,2 重打 stop reason 异常"); + + const streamRetryThreadId = "thread_stream_retry_v1"; + const streamRetryResponse = await readSseUntilClose( + `http://127.0.0.1:${gatewayPort}/v1/responses`, + { + stream: true, + thread_id: streamRetryThreadId, + test_reasoning_before_success_times: 1, + test_reasoning_retry_key: "stream-v1-round2", + }, + ); + assert(streamRetryResponse.status === 200, `/v1/responses 流式 1,1 重打未恢复: ${streamRetryResponse.status}`); + assert(streamRetryResponse.text.includes("hello"), "/v1/responses 流式 1,1 重打未拿到正常 SSE 内容"); + assert(streamRetryResponse.text.includes("[DONE]"), "/v1/responses 流式 1,1 重打未完整结束"); + assert(streamRetryResponse.headers.get("x-upstream-reasoning-attempt") === "2", "/v1/responses 流式 1,1 重打未命中第二次上游请求"); + const streamRetryEntryResponse = await fetch( + `http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(streamRetryThreadId)}`, + { headers: adminHeaders }, + ); + const streamRetryEntryPayload = await streamRetryEntryResponse.json(); + const streamRetryEntry = (streamRetryEntryPayload?.entries || []).find((entry) => entry.thread_id === streamRetryThreadId); + assert(streamRetryEntry?.reasoning_retry_query_count === 2, "/v1/responses 流式 1,1 重打未记录两次 query"); + assert(streamRetryEntry?.response_stream === true, "/v1/responses 流式 1,1 重打未保留流式标记"); + const recoveredPayload = JSON.stringify({ test_fail_before_response_once: true }); const recoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { method: "POST", @@ -513,6 +739,30 @@ async function run() { assert(threadEntry?.response_id === "resp_test", "non-stream 请求记录未保留 response_id"); assert(threadEntry?.thread_id === "thread_nonstream", "non-stream 请求记录未保留 thread_id"); + const effortTrackedResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + test_reasoning_tokens: 128, + reasoning: { + effort: "xhigh", + summary: "auto", + }, + }), + }); + assert(effortTrackedResponse.status === 200, `reasoning.effort 请求失败: ${effortTrackedResponse.status}`); + await effortTrackedResponse.json(); + + const effortRequestsResponse = await fetch( + `http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent("xhigh")}`, + { headers: adminHeaders }, + ); + const effortRequestsPayload = await effortRequestsResponse.json(); + const effortEntry = (effortRequestsPayload?.entries || []).find((entry) => entry.reasoning_effort === "xhigh"); + assert(effortRequestsResponse.status === 200, `reasoning.effort 搜索失败: ${effortRequestsResponse.status}`); + assert(effortEntry?.reasoning_effort === "xhigh", "请求记录未保留 reasoning.effort"); + assert(effortEntry?.reasoning_summary === "auto", "请求记录未保留 reasoning.summary"); + const sameRequestPayload = JSON.stringify({ test_reasoning_tokens: 128, test_request_id_marker: "same" }); const sameRequestFirstResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { method: "POST", diff --git a/scripts/test-install-restore.mjs b/scripts/test-install-restore.mjs index 63a4a4f..753f8c5 100644 --- a/scripts/test-install-restore.mjs +++ b/scripts/test-install-restore.mjs @@ -149,6 +149,10 @@ async function run() { gatewayConfig.upstream_base_url === `http://127.0.0.1:${upstreamPort}`, "Gateway config did not preserve original upstream_base_url", ); + assert( + gatewayConfig.reasoning_match_mode === "formula_518n_minus_2", + "Gateway config did not default reasoning_match_mode to formula_518n_minus_2", + ); assert(Array.isArray(gatewayConfig.endpoints), "Gateway config endpoints must be an array"); assert( gatewayConfig.endpoints.includes("/responses") && @@ -195,6 +199,13 @@ async function run() { }); assert(blocked516Response.status === 502, `Default 516 block did not trigger: ${blocked516Response.status}`); + const blocked2070Response = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ test_reasoning_tokens: 2070 }), + }); + assert(blocked2070Response.status === 502, `Default 2070 formula block did not trigger: ${blocked2070Response.status}`); + const metricsStatusResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/status`); const metricsStatusPayload = await metricsStatusResponse.json(); assert(metricsStatusResponse.status === 200, `Status API failed after traffic: ${metricsStatusResponse.status}`); @@ -220,6 +231,7 @@ async function run() { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ + reasoning_match_mode: "manual", reasoning_equals: [1024], retryable_status_codes: [429, 503, 529], retryable_error_messages: ["Selected model is at capacity. Please try a different model."], @@ -231,10 +243,15 @@ async function run() { const saveConfigPayload = await saveConfigResponse.json(); assert(saveConfigResponse.status === 200, `Save config API failed: ${saveConfigResponse.status}`); assert(saveConfigPayload.config?.non_stream_status_code === 503, "Save config API did not return updated config"); + assert(saveConfigPayload.config?.reasoning_match_mode === "manual", "Save config API did not return updated reasoning_match_mode"); const updatedGatewayConfig = JSON.parse( await readFile(path.join(stateRoot, "config", "config.json"), "utf8"), ); + assert( + updatedGatewayConfig.reasoning_match_mode === "manual", + "Saved config file did not persist reasoning_match_mode", + ); assert( JSON.stringify(updatedGatewayConfig.reasoning_equals) === JSON.stringify([1024]), "Saved config file did not persist reasoning_equals", @@ -261,6 +278,13 @@ async function run() { }); assert(blockedAfterSave.status === 503, `Hot reloaded config did not take effect: ${blockedAfterSave.status}`); + const manualModePassthrough = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ test_reasoning_tokens: 2070 }), + }); + assert(manualModePassthrough.status === 200, `manual 模式下 2070 不应继续被拦截: ${manualModePassthrough.status}`); + const restoreViaUiResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/restore`, { method: "POST", headers: { "content-type": "application/json" }, diff --git a/scripts/test-launch-ui.mjs b/scripts/test-launch-ui.mjs index 42fa535..13c795b 100644 --- a/scripts/test-launch-ui.mjs +++ b/scripts/test-launch-ui.mjs @@ -155,6 +155,10 @@ async function run() { statusPayload.state?.original_base_url === upstreamBaseUrl, "First launch did not persist the original upstream base URL", ); + assert( + statusPayload.config?.reasoning_match_mode === "formula_518n_minus_2", + "First launch did not default reasoning_match_mode to formula_518n_minus_2", + ); const firstStateRaw = await readFile(path.join(stateRoot, "state.json"), "utf8"); const firstState = JSON.parse(firstStateRaw); @@ -194,6 +198,13 @@ async function run() { }); assert(blockedResponse.status === 502, `Default 516 interception was not active: ${blockedResponse.status}`); + const blockedFormulaResponse = await fetch(`${gatewayBaseUrl}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ test_reasoning_tokens: 2070 }), + }); + assert(blockedFormulaResponse.status === 502, `Default 2070 formula interception was not active: ${blockedFormulaResponse.status}`); + process.stdout.write("PASS launch-ui flow\n"); } finally { try { diff --git a/ui-src/src/App.tsx b/ui-src/src/App.tsx index 6e322be..f04bb39 100644 --- a/ui-src/src/App.tsx +++ b/ui-src/src/App.tsx @@ -2,6 +2,7 @@ import { FormEvent, startTransition, useEffect, useState } from "react"; type PageKey = "overview" | "requests" | "profiles" | "rules" | "logs"; type Tone = "" | "success" | "error"; +type ReasoningMatchMode = "formula_518n_minus_2" | "manual"; type GatewayConfig = { profile_name?: string; @@ -16,6 +17,7 @@ type GatewayConfig = { request_history_limit?: number; model_remap?: string; endpoints?: string[]; + reasoning_match_mode?: ReasoningMatchMode; reasoning_equals?: number[]; retryable_status_codes?: number[]; retryable_error_messages?: string[]; @@ -90,10 +92,36 @@ type RequestEntry = { model?: string | null; requested_model?: string | null; forwarded_model?: string | null; + reasoning_effort?: string | null; + reasoning_summary?: string | null; response_stream?: boolean; stream_chunk_count?: number | null; usage_last_updated_at?: string | null; upstream_attempt_count?: number | null; + reasoning_retry_enabled?: boolean; + reasoning_retry_query_count?: number | null; + reasoning_retry_round_count?: number | null; + reasoning_retry_current_round?: number | null; + reasoning_retry_current_width?: number | null; + reasoning_retry_current_firsts?: Array<{ + round?: number | null; + slot?: number | null; + first_response_at?: string | null; + first_response_delay_ms?: number | null; + outcome?: string | null; + status_code?: number | null; + upstream_status_code?: number | null; + reasoning_tokens?: number | null; + matched?: boolean | null; + }> | null; + reasoning_retry_winner_round?: number | null; + reasoning_retry_winner_slot?: number | null; + reasoning_retry_stop_reason?: string | null; + reasoning_retry_thread_mode?: string | null; + reasoning_retry_extra_inspected_count?: number | null; + reasoning_retry_extra_matched_count?: number | null; + reasoning_retry_extra_usage?: Usage | null; + reasoning_retry_extra_reasoning_counts?: Record | null; matched?: boolean; status_code?: number | null; upstream_status_code?: number | null; @@ -130,6 +158,7 @@ type ProfileFormModel = { auth_json_key?: string; request_history_limit?: string; model_remap?: string; + reasoning_match_mode?: ReasoningMatchMode; reasoning_equals?: string; retryable_status_codes?: string; retryable_error_messages?: string[]; @@ -150,6 +179,7 @@ type Profile = { auth_source?: string; request_history_limit?: string; model_remap?: string; + reasoning_match_mode?: ReasoningMatchMode; reasoning_equals?: string; }; form?: ProfileFormModel; @@ -199,6 +229,7 @@ type ProfileFormState = { auth_json_key: string; request_history_limit: string; model_remap: string; + reasoning_match_mode: ReasoningMatchMode; reasoning_equals: string; retryable_status_codes: string; retryable_error_messages: string; @@ -208,6 +239,7 @@ type ProfileFormState = { }; type RuleFormState = { + reasoning_match_mode: ReasoningMatchMode; reasoning_equals: string; retryable_status_codes: string; retryable_error_messages: string; @@ -307,6 +339,7 @@ const defaultProfileForm: ProfileFormState = { auth_json_key: "OPENAI_API_KEY", request_history_limit: "0", model_remap: "", + reasoning_match_mode: "formula_518n_minus_2", reasoning_equals: "516,1034,1552", retryable_status_codes: "429,503", retryable_error_messages: "Selected model is at capacity. Please try a different model.\nstream disconnected before completion: Concurrency limit exceeded for account, please retry later", @@ -333,6 +366,17 @@ function durationSeconds(value?: number | null) { return typeof value === "number" && Number.isFinite(value) ? `${(value / 1000).toFixed(2)} s` : "-"; } +function compactDurationSeconds(value?: number | null) { + if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) { + return "?"; + } + const seconds = value / 1000; + if (seconds >= 100) { + return `${seconds.toFixed(0)}s`; + } + return `${seconds.toFixed(1).replace(/\\.0$/, "")}s`; +} + function secondsSince(startedAt: string | null | undefined, updatedAt: string | null | undefined) { if (!startedAt || !updatedAt) { return "-"; @@ -392,6 +436,98 @@ function requestPrimaryId(entry: RequestEntry) { return entry.response_id || entry.request_id || "-"; } +function hasReasoningRetryInfo(entry: RequestEntry) { + return Boolean( + entry.reasoning_retry_enabled || + entry.reasoning_retry_thread_mode || + entry.reasoning_retry_query_count || + entry.reasoning_retry_round_count || + entry.reasoning_retry_stop_reason, + ); +} + +function retryRoundWidthText(entry: RequestEntry) { + const round = entry.reasoning_retry_current_round || entry.reasoning_retry_round_count || 0; + const width = entry.reasoning_retry_current_width || 0; + if (!round) { + return "-"; + } + return width ? `${numberFormat(round)}(${numberFormat(width)})` : numberFormat(round); +} + +function retryFirstLabel(first: NonNullable[number], fallbackRound?: number | null) { + const round = first.round ?? fallbackRound ?? null; + const slot = first.slot ?? null; + if (typeof round === "number" && Number.isInteger(round)) { + if (typeof slot === "number" && Number.isInteger(slot) && slot > 1) { + return `${numberFormat(round)}-${numberFormat(slot)}`; + } + return numberFormat(round); + } + if (typeof slot === "number" && Number.isInteger(slot)) { + return `#${numberFormat(slot)}`; + } + return "?"; +} + +function retryFirstText( + first: NonNullable[number], + fallbackRound?: number | null, +) { + const delay = durationSeconds(first.first_response_delay_ms); + const outcome = first.outcome || "pending"; + const reasoning = typeof first.reasoning_tokens === "number" ? ` r${numberFormat(first.reasoning_tokens)}` : ""; + const status = typeof first.status_code === "number" ? ` ${first.status_code}` : ""; + return `${retryFirstLabel(first, fallbackRound)} ${delay} ${outcome}${status}${reasoning}`; +} + +function retryFirstCompactText( + first: NonNullable[number], +) { + return compactDurationSeconds(first.first_response_delay_ms); +} + +function formatRetryStopReason(value?: string | null) { + const reason = `${value || ""}`.trim(); + if (!reason) { + return "-"; + } + const labels: Record = { + success: "成功", + missing_thread_id: "缺少 thread_id", + completed_without_retry: "未触发调度", + reasoning_guard: "reasoning 命中", + retryable_upstream_error: "上游可重试错误", + fatal: "致命错误", + exhausted_without_winner: "无赢家", + }; + return labels[reason] || reason; +} + +function formatRetryThreadMode(value?: string | null) { + const mode = `${value || ""}`.trim(); + if (!mode || mode === "disabled") { + return "未启用"; + } + if (mode === "thread_id") { + return "按 thread_id"; + } + if (mode === "missing_thread_id") { + return "缺少 thread_id"; + } + return mode; +} + +function sortedReasoningCounts(value?: Record | null) { + return Object.entries(value || {}).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); + }); +} + function splitList(value: string) { return value .split(/[\s,]+/) @@ -406,6 +542,21 @@ function splitLines(value: string) { .filter(Boolean); } +function normalizeReasoningMode(value: unknown): ReasoningMatchMode { + return value === "manual" ? "manual" : "formula_518n_minus_2"; +} + +function formatReasoningMode(mode: ReasoningMatchMode) { + return mode === "manual" ? "manual" : "518n-2"; +} + +function formatReasoningRule(mode: ReasoningMatchMode, reasoningEquals?: string) { + if (mode === "manual") { + return reasoningEquals || "-"; + } + return "516, 1034, 1552, ..."; +} + async function fetchJson(url: string, options?: RequestInit): Promise { const headers = new Headers(options?.headers || {}); const accessKey = window.localStorage.getItem(ACCESS_KEY_STORAGE_KEY)?.trim(); @@ -435,6 +586,7 @@ function profileFormFromStatus(status: StatusPayload | null): ProfileFormState { auth_json_key: config.upstream_auth_json_key || defaultProfileForm.auth_json_key, request_history_limit: String(config.request_history_limit ?? defaultProfileForm.request_history_limit), model_remap: config.model_remap || "", + reasoning_match_mode: normalizeReasoningMode(config.reasoning_match_mode), reasoning_equals: Array.isArray(config.reasoning_equals) ? config.reasoning_equals.join(",") : defaultProfileForm.reasoning_equals, @@ -472,7 +624,8 @@ function profileFormFromProfile(profile: Profile): ProfileFormState { auth_json_key: form.auth_json_key || "OPENAI_API_KEY", request_history_limit: form.request_history_limit || defaultProfileForm.request_history_limit, model_remap: form.model_remap || "", - reasoning_equals: form.reasoning_equals || "", + reasoning_match_mode: normalizeReasoningMode(form.reasoning_match_mode), + reasoning_equals: form.reasoning_equals || defaultProfileForm.reasoning_equals, retryable_status_codes: form.retryable_status_codes || defaultProfileForm.retryable_status_codes, retryable_error_messages: Array.isArray(form.retryable_error_messages) ? form.retryable_error_messages.join("\n") @@ -488,6 +641,7 @@ function profileFormFromProfile(profile: Profile): ProfileFormState { function ruleFormFromStatus(status: StatusPayload | null): RuleFormState { const config = status?.config || {}; return { + reasoning_match_mode: normalizeReasoningMode(config.reasoning_match_mode), reasoning_equals: Array.isArray(config.reasoning_equals) ? config.reasoning_equals.join(", ") : "", retryable_status_codes: Array.isArray(config.retryable_status_codes) ? config.retryable_status_codes.join(", ") @@ -724,6 +878,7 @@ export default function App() { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ + reasoning_match_mode: ruleForm.reasoning_match_mode, reasoning_equals: splitList(ruleForm.reasoning_equals) .map((value) => Number.parseInt(value, 10)) .filter((value) => Number.isInteger(value)), @@ -784,6 +939,7 @@ export default function App() { auth_json_key: profileForm.auth_json_key, request_history_limit: Number.parseInt(profileForm.request_history_limit, 10), model_remap: profileForm.model_remap, + reasoning_match_mode: profileForm.reasoning_match_mode, reasoning_equals: splitList(profileForm.reasoning_equals), retryable_status_codes: splitList(profileForm.retryable_status_codes), retryable_error_messages: splitLines(profileForm.retryable_error_messages), @@ -1089,6 +1245,11 @@ export default function App() { const statusTone = entry.error ? "error" : entry.matched ? "warn" : ""; const effectiveInput = effectiveInputTokens(usage.input_tokens, usage.cached_tokens); const cachedHitRatio = cachedRatio(usage.input_tokens, usage.cached_tokens); + const showReasoningRetry = hasReasoningRetryInfo(entry); + const retryReasoningCounts = sortedReasoningCounts(entry.reasoning_retry_extra_reasoning_counts); + const retryCurrentFirsts = Array.isArray(entry.reasoning_retry_current_firsts) + ? entry.reasoning_retry_current_firsts + : []; return (
@@ -1101,10 +1262,21 @@ export default function App() { 发 {timestamp(entry.started_at)} - - 首 - {durationSeconds(entry.first_response_delay_ms)} - + {retryCurrentFirsts.length > 0 ? ( + + 首 + {retryCurrentFirsts.map((first, index) => ( + + {retryFirstCompactText(first)} + + ))} + + ) : ( + + 首 + {durationSeconds(entry.first_response_delay_ms)} + + )} 总 {durationSeconds(entry.duration_ms)} @@ -1123,6 +1295,12 @@ export default function App() { ? `${numberFormat(entry.stream_chunk_count)} chunk / ${bytesFormat(entry.response_bytes_received)} / ${secondsSince(entry.started_at, entry.last_activity_at || entry.usage_last_updated_at || entry.finished_at)}` : "-"} + {showReasoningRetry ? ( + + 重打 + {`${retryRoundWidthText(entry)} / ${numberFormat(entry.reasoning_retry_query_count || 0)}q`} + + ) : null} 收 {timestamp(entry.finished_at)} @@ -1138,6 +1316,11 @@ export default function App() {
{entry.matched ? matched : pass} {entry.error ? error : null} + {showReasoningRetry ? ( + + retry {formatRetryStopReason(entry.reasoning_retry_stop_reason)} + + ) : null}
@@ -1150,6 +1333,12 @@ export default function App() { ? `转发为 ${entry.forwarded_model}` : entry.forwarded_model || "-"} + + {entry.reasoning_effort + ? `强度 ${entry.reasoning_effort}` + : "强度 -"} + {entry.reasoning_summary ? ` / summary ${entry.reasoning_summary}` : ""} +
@@ -1182,6 +1371,54 @@ export default function App() { {`request ${entry.request_id || "-"}`} {`thread ${entry.thread_id || "-"}`}
+ + {showReasoningRetry ? ( +
+ + {formatRetryThreadMode(entry.reasoning_retry_thread_mode)} + + schedule 1,1,2,2,4,4... / query {numberFormat(entry.reasoning_retry_query_count || 0)} + {" / "} + round {retryRoundWidthText(entry)} + + + winner {entry.reasoning_retry_winner_round && entry.reasoning_retry_winner_slot + ? `round ${entry.reasoning_retry_winner_round} slot ${entry.reasoning_retry_winner_slot}` + : "-"} + {" / "} + stop {formatRetryStopReason(entry.reasoning_retry_stop_reason)} + + + extra matched {numberFormat(entry.reasoning_retry_extra_matched_count || 0)} + {" / "} + inspected {numberFormat(entry.reasoning_retry_extra_inspected_count || 0)} + {" / "} + extra reasoning {numberFormat(entry.reasoning_retry_extra_usage?.reasoning_tokens)} + + {retryReasoningCounts.length > 0 ? ( +
+ {retryReasoningCounts.slice(0, 4).map(([reasoning, count]) => ( + + {reasoning}: {numberFormat(count)} + + ))} +
+ ) : null} + {retryCurrentFirsts.length > 0 ? ( +
+ {retryCurrentFirsts.map((first, index) => ( + + {retryFirstText(first, entry.reasoning_retry_current_round)} + + ))} +
+ ) : null} +
+ ) : null}
); @@ -1246,7 +1483,17 @@ export default function App() { - + + )) @@ -1323,6 +1570,17 @@ export default function App() { onChange={(event) => setProfileForm({ ...profileForm, request_history_limit: event.target.value })} /> + + + setProfileForm({ ...profileForm, reasoning_equals: event.target.value })} /> @@ -1396,6 +1654,17 @@ export default function App() {
+ + + setRuleForm({ ...ruleForm, reasoning_equals: event.target.value })} /> diff --git a/ui-src/src/styles.css b/ui-src/src/styles.css index 19a46f9..31b561b 100644 --- a/ui-src/src/styles.css +++ b/ui-src/src/styles.css @@ -450,6 +450,11 @@ a { background: #fff0d9; } +.badge.success { + color: var(--accent); + background: #d7eee6; +} + .badge.error { color: var(--red); background: #ffece7; @@ -622,9 +627,29 @@ a { background: #e5eaef; } +.meta-retry { + color: #0f5045; + background: #d2eee4; +} + +.meta-first-list-pill { + display: inline-flex; + flex-wrap: wrap; + gap: 4px; +} + +.meta-first-mini { + border-radius: 999px; + padding: 2px 6px; + background: rgba(198, 84, 37, 0.09); + color: #8d421e; + font-size: 11px; + font-weight: 700; +} + .request-grid { display: grid; - grid-template-columns: 1.05fr 1.35fr 1.6fr; + grid-template-columns: minmax(150px, 1fr) minmax(220px, 1.25fr) minmax(260px, 1.5fr) minmax(180px, 1.05fr); gap: 8px; } @@ -653,6 +678,46 @@ a { font-size: 12px; } +.reasoning-retry-block { + grid-column: span 2; + border-color: rgba(22, 107, 92, 0.16); + background: + radial-gradient(circle at 0% 0%, rgba(22, 107, 92, 0.1), transparent 42%), + rgba(255, 250, 240, 0.82); +} + +.retry-chip-row { + display: flex; + flex-wrap: wrap; + gap: 5px; +} + +.retry-first-list { + display: flex; + flex-wrap: wrap; + gap: 6px; +} + +.retry-first-chip { + border: 1px solid rgba(22, 107, 92, 0.18); + border-radius: 999px; + padding: 4px 8px; + color: #12463c; + background: rgba(22, 107, 92, 0.1); + font-size: 11px; + font-weight: 750; +} + +.retry-first-chip.pending { + color: var(--muted); + background: rgba(30, 33, 29, 0.05); +} + +.compact-chip { + padding: 3px 7px; + font-size: 11px; +} + table { width: 100%; min-width: 960px; @@ -946,9 +1011,11 @@ form { } .request-head, - .request-grid { + .request-grid, + .reasoning-retry-block { display: grid; grid-template-columns: 1fr; + grid-column: auto; } .info-row {