diff --git a/README.md b/README.md index e9190fc..7fb0b07 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ - 保持 Codex 继续使用现有 `auth.json` - 只把 `config.toml` 的当前 provider `base_url` 改成本地网关 - 非流式命中 `reasoning_tokens = 516` 时返回 `502` -- 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`，以及 `stream disconnected before completion: Concurrency limit exceeded for account, please retry later`），也会自动重试；重试耗尽后转成本地 `502` +- 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`、`stream disconnected before completion: Concurrency limit exceeded for account, please retry later`，以及 Responses 错误码 `server_is_overloaded` / `slow_down`），也会自动重试；重试耗尽后转成本地 `502` - 除了 `429/503` JSON 错误响应，也会识别 `200` 但返回体本质是错误、以及流式失败事件里携带同样文案的情况 - 流式命中时默认先缓存并判断；一旦命中 `516`，统一返回 `502` - 流式成功响应在严格检查模式下仍会先缓存完成；成功后会按真实且规范化的 Responses 生命周期与输出顺序逐块回放给 Codex，每个 SSE 块之间至少间隔 5ms，并关闭 TCP 小包聚合，避免大量 delta 在同一事件循环突发到达；不会伪造缺少 response ID 的生命周期事件 diff --git a/gateway.mjs b/gateway.mjs index 9d004a7..66cdec6 100644 --- a/gateway.mjs +++ b/gateway.mjs @@ -4403,6 +4403,47 @@ function truncateRetryableMessage(value, maxLength = 280) { return text.length > maxLength ? `${text.slice(0, maxLength - 3)}...` : text; } +const RETRYABLE_OVERLOAD_ERROR_CODES = new Set([ + "server_is_overloaded", + "slow_down", +]); + +function findRetryableOverloadErrorCode(value, state = { seen: new Set() }, depth = 0) { + if (!value || typeof value !== "object" || depth > 6 || state.seen.has(value)) { + return null; + } + state.seen.add(value); + + if (Array.isArray(value)) { + for (const item of value) { + const match = findRetryableOverloadErrorCode(item, state, depth + 1); + if (match) { + return match; + } + } + return null; + } + + const code = typeof value.code === "string" ? value.code.trim().toLowerCase() : ""; + if (RETRYABLE_OVERLOAD_ERROR_CODES.has(code)) { + return { + matched_code: code, + matched_pattern: `error_code:${code}`, + matched_message: truncateRetryableMessage( + firstNonEmptyString(value.message, value.detail, value.description, code), + ), + }; + } + + for (const nested of Object.values(value)) { + const match = findRetryableOverloadErrorCode(nested, state, depth + 1); + if (match) { + return match; + } + } + return null; +} + function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, bodyText) { const retryableMessages = normalizePhraseList( config.retryable_error_messages, @@ -4420,10 +4461,16 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, retryableStatusCodes.length === 0 || retryableStatusCodes.includes(upstreamStatusCode) ); - if (!statusEligible && !isRetryableErrorPayloadShape(parsedBody)) { + const failurePayload = isRetryableErrorPayloadShape(parsedBody); + if (!statusEligible && !failurePayload) { return null; } + const overloadCodeMatch = findRetryableOverloadErrorCode(parsedBody); + if (overloadCodeMatch) { + return overloadCodeMatch; + } + return matchRetryableMessage(config, parsedBody, bodyText); } @@ -4525,6 +4572,10 @@ function findRetryableStreamErrorMatch(config, parsedBody, bodyText, eventName = if (!isRetryableErrorPayloadShape(parsedBody, eventName)) { return null; } + const overloadCodeMatch = findRetryableOverloadErrorCode(parsedBody); + if (overloadCodeMatch) { + return overloadCodeMatch; + } return matchRetryableMessage(config, parsedBody, bodyText); } diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index 0865ce2..93394b2 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/scripts/test-gateway-e2e.mjs @@ -88,25 +88,22 @@ function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20, options = }, destroyDelayMs); } -function buildCapacityStreamPayload(message, shape = "default") { +function buildCapacityStreamPayload(message, shape = "default", errorCode = "") { + const error = { + message, + type: "server_error", + ...(errorCode ? { code: errorCode } : {}), + }; if (shape === "response_failed") { return { type: "response.failed", response: { status: "failed", - error: { - message, - type: "server_error", - }, + error, }, }; } - return { - error: { - message, - type: "server_error", - }, - }; + return { error }; } function createCapacityErrorSseResponse( @@ -116,7 +113,11 @@ function createCapacityErrorSseResponse( options = {}, ) { const eventName = options.eventName || "error"; - const payload = options.payload || buildCapacityStreamPayload(message, options.payloadShape || "default"); + const payload = options.payload || buildCapacityStreamPayload( + message, + options.payloadShape || "default", + options.errorCode || "", + ); createSseResponse( res, [ @@ -231,7 +232,7 @@ function startFakeUpstream(port, options = {}) { ]); return; } - if (parsed.test_capacity_error) { + if (parsed.test_capacity_error && !parsed.stream) { const capacityMessage = parsed.test_capacity_message || "Selected model is at capacity. Please try a different model."; createJsonResponse( res, @@ -240,6 +241,7 @@ function startFakeUpstream(port, options = {}) { error: { message: capacityMessage, type: "server_error", + ...(parsed.test_capacity_code ? { code: parsed.test_capacity_code } : {}), }, }, { "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` }, @@ -255,6 +257,8 @@ function startFakeUpstream(port, options = {}) { parsed.test_capacity_status ?? "default", parsed.test_capacity_stream_event_name || "", parsed.test_capacity_stream_payload_shape || "", + parsed.test_capacity_code || "", + parsed.test_capacity_message || "", ].join(":"); const capacityCount = (capacityBeforeSuccessCounts.get(capacityKey) || 0) + 1; capacityBeforeSuccessCounts.set(capacityKey, capacityCount); @@ -268,6 +272,7 @@ function startFakeUpstream(port, options = {}) { { eventName: parsed.test_capacity_stream_event_name || "error", payloadShape: parsed.test_capacity_stream_payload_shape || "default", + errorCode: parsed.test_capacity_code || "", }, ); return; @@ -279,6 +284,7 @@ function startFakeUpstream(port, options = {}) { error: { message: capacityMessage, type: "server_error", + ...(parsed.test_capacity_code ? { code: parsed.test_capacity_code } : {}), }, }, { "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` }, @@ -294,6 +300,7 @@ function startFakeUpstream(port, options = {}) { { eventName: parsed.test_capacity_stream_event_name || "error", payloadShape: parsed.test_capacity_stream_payload_shape || "default", + errorCode: parsed.test_capacity_code || "", }, ); return; @@ -1482,6 +1489,28 @@ async function run() { "stream+response.failed capacity error 返回体未标记 retry trigger", ); + const streamSlowDownCodeResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + stream: true, + test_capacity_error: true, + test_capacity_message: "Raw provider asked the client to wait.", + test_capacity_code: "slow_down", + test_capacity_stream_event_name: "response.failed", + test_capacity_stream_payload_shape: "response_failed", + }), + }); + const streamSlowDownCodeBody = await streamSlowDownCodeResponse.json(); + assert( + streamSlowDownCodeResponse.status === 502, + `stream+slow_down 持续过载未返回 502: ${streamSlowDownCodeResponse.status}`, + ); + assert( + streamSlowDownCodeBody?.error?.code === "upstream_error_retry_triggered", + "stream+slow_down 持续过载未标记 retry trigger", + ); + const streamCapacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { method: "POST", headers: { "content-type": "application/json" }, @@ -1520,6 +1549,43 @@ async function run() { ); assert(streamCapacityResponseFailedRecoveredEntry, "stream response.failed capacity 恢复后的请求记录未保留重试次数"); + const overloadCodeThreadId = "thread-server-overloaded-code-retry"; + const streamOverloadCodeRecoveredResponse = await readSseUntilClose( + `http://127.0.0.1:${gatewayPort}/responses`, + { + stream: true, + thread_id: overloadCodeThreadId, + test_capacity_before_success_times: 2, + test_capacity_message: "Raw provider reported a temporary overload.", + test_capacity_code: "server_is_overloaded", + test_capacity_stream_event_name: "response.failed", + test_capacity_stream_payload_shape: "response_failed", + test_reasoning_tokens: 128, + }, + ); + assert( + streamOverloadCodeRecoveredResponse.status === 200, + `stream+server_is_overloaded 未自动恢复: ${streamOverloadCodeRecoveredResponse.status}`, + ); + assert( + streamOverloadCodeRecoveredResponse.text.includes("hello"), + "stream+server_is_overloaded 恢复后未拿到正常 SSE 内容", + ); + assert( + !streamOverloadCodeRecoveredResponse.text.includes("Raw provider reported a temporary overload."), + "stream+server_is_overloaded 自动恢复前不应向客户端泄漏失败轮次", + ); + const overloadCodeRequestsResponse = await fetch( + `http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(overloadCodeThreadId)}`, + { headers: adminHeaders }, + ); + const overloadCodeRequestsPayload = await overloadCodeRequestsResponse.json(); + const overloadCodeEntry = (overloadCodeRequestsPayload?.entries || []).find( + (entry) => entry.thread_id === overloadCodeThreadId, + ); + assert(overloadCodeEntry?.status_code === 200, "server_is_overloaded 恢复请求未记录最终 200"); + assert(overloadCodeEntry?.upstream_attempt_count === 3, "server_is_overloaded 恢复请求未记录 3 次上游尝试"); + const streamThreadResponse = await readSseUntilClose( `http://127.0.0.1:${gatewayPort}/responses`, { stream: true, test_reasoning_tokens: 128, thread_id: "thread_stream_ok" },