retry more capacity-style upstream failures

README.md

@@ -9,7 +9,8 @@ tg群:https://t.me/AI_INPUT_IM
 - 保持 Codex 继续使用现有 `auth.json`
 - 只把 `config.toml` 的当前 provider `base_url` 改成本地网关
 - 非流式命中 `reasoning_tokens = 516` 时返回 `502`
-- 上游若返回明确的容量错误（默认匹配 `429/503` 且错误文案包含 `Selected model is at capacity. Please try a different model.`），也会转成本地 `502`
+- 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`），也会自动重试；重试耗尽后转成本地 `502`
+- 除了 `429/503` JSON 错误响应，也会识别 `200` 但返回体本质是错误、以及流式失败事件里携带同样文案的情况
 - 流式命中时默认先缓存并判断；一旦命中 `516`，统一返回 `502`
 - 默认同时拦截 root 路径和 `/v1` 路径：
   - `/responses`

gateway.mjs

@@ -2419,10 +2419,6 @@ function truncateRetryableMessage(value, maxLength = 280) {
 }
 
 function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, bodyText) {
-  if (!isUpstreamErrorStatus(upstreamStatusCode)) {
-    return null;
-  }
-
   const retryableMessages = normalizePhraseList(
     config.retryable_error_messages,
     DEFAULT_CONFIG.retryable_error_messages,
@@ -2435,7 +2431,11 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody,
     config.retryable_status_codes,
     DEFAULT_CONFIG.retryable_status_codes,
   );
-  if (retryableStatusCodes.length > 0 && !retryableStatusCodes.includes(upstreamStatusCode)) {
+  const statusEligible = isUpstreamErrorStatus(upstreamStatusCode) && (
+    retryableStatusCodes.length === 0 ||
+    retryableStatusCodes.includes(upstreamStatusCode)
+  );
+  if (!statusEligible && !isRetryableErrorPayloadShape(parsedBody)) {
     return null;
   }
 
@@ -2480,27 +2480,64 @@ function matchRetryableMessage(config, parsedBody, bodyText) {
   return null;
 }
 
-function isRetryableStreamErrorShape(parsedBody, eventName = "") {
+function hasRetryableFailureKeyword(value) {
-  if (`${eventName || ""}`.trim().toLowerCase() === "error") {
+  const normalized = `${value || ""}`.trim().toLowerCase();
+  if (!normalized) {
+    return false;
+  }
+  return (
+    normalized.includes("error") ||
+    normalized.includes("failed") ||
+    normalized.includes("failure") ||
+    normalized.includes("cancelled") ||
+    normalized.includes("canceled") ||
+    normalized.includes("rate_limit") ||
+    normalized.includes("overloaded") ||
+    normalized.includes("unavailable")
+  );
+}
+
+function hasRetryableFailureShape(value, depth = 0) {
+  if (!value || depth > 4) {
+    return false;
+  }
+  if (Array.isArray(value)) {
+    return value.some((item) => hasRetryableFailureShape(item, depth + 1));
+  }
+  if (typeof value !== "object") {
+    return false;
+  }
+  if (Object.hasOwn(value, "error")) {
+    return true;
+  }
+  if (Number.isInteger(value.status) && value.status >= 400) {
+    return true;
+  }
+  for (const key of ["type", "event", "status", "state", "result", "code"]) {
+    if (typeof value[key] === "string" && hasRetryableFailureKeyword(value[key])) {
+      return true;
+    }
+  }
+  for (const key of ["response", "data", "meta", "details", "detail"]) {
+    if (hasRetryableFailureShape(value[key], depth + 1)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+function isRetryableErrorPayloadShape(parsedBody, eventName = "") {
+  if (hasRetryableFailureKeyword(eventName)) {
     return true;
   }
   if (!parsedBody || typeof parsedBody !== "object" || Array.isArray(parsedBody)) {
     return false;
   }
-  if (Object.hasOwn(parsedBody, "error")) {
+  return hasRetryableFailureShape(parsedBody);
-    return true;
-  }
-  if (typeof parsedBody.type === "string" && parsedBody.type.toLowerCase().includes("error")) {
-    return true;
-  }
-  if (typeof parsedBody.event === "string" && parsedBody.event.toLowerCase().includes("error")) {
-    return true;
-  }
-  return Number.isInteger(parsedBody.status) && parsedBody.status >= 400;
 }
 
 function findRetryableStreamErrorMatch(config, parsedBody, bodyText, eventName = "") {
-  if (!isRetryableStreamErrorShape(parsedBody, eventName)) {
+  if (!isRetryableErrorPayloadShape(parsedBody, eventName)) {
     return null;
   }
   return matchRetryableMessage(config, parsedBody, bodyText);
@@ -2564,14 +2601,6 @@ async function fetchUpstreamWithRetry(upstreamUrl, init, config, logger, request
         };
       }
 
-      if (!isUpstreamErrorStatus(response.status)) {
-        return {
-          response,
-          attempt_count: attempt,
-          retryable_upstream_error: null,
-        };
-      }
-
       const contentType = response.headers.get("content-type");
       if (!isJsonContentType(contentType)) {
         return {

scripts/test-gateway-e2e.mjs

@@ -80,16 +80,40 @@ function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) {
   }, destroyDelayMs);
 }
 
+function buildCapacityStreamPayload(message, shape = "default") {
+  if (shape === "response_failed") {
+    return {
+      type: "response.failed",
+      response: {
+        status: "failed",
+        error: {
+          message,
+          type: "server_error",
+        },
+      },
+    };
+  }
+  return {
+    error: {
+      message,
+      type: "server_error",
+    },
+  };
+}
+
 function createCapacityErrorSseResponse(
   res,
   message = "Selected model is at capacity. Please try a different model.",
   intervalMs = 20,
+  options = {},
 ) {
+  const eventName = options.eventName || "error";
+  const payload = options.payload || buildCapacityStreamPayload(message, options.payloadShape || "default");
   createSseResponse(
     res,
     [
-      'event: error\n',
+      `event: ${eventName}\n`,
-      `data: ${JSON.stringify({ error: { message, type: "server_error" } })}\n\n`,
+      `data: ${JSON.stringify(payload)}\n\n`,
     ],
     intervalMs,
   );
@@ -140,28 +164,43 @@ function startFakeUpstream(port) {
           return;
         }
         if (parsed.test_capacity_error) {
+          const capacityMessage = parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.";
           createJsonResponse(
             res,
             parsed.test_capacity_status ?? 503,
             {
               error: {
-                message: parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.",
+                message: capacityMessage,
                 type: "server_error",
               },
             },
-            { "x-upstream-test": "capacity-error" },
+            { "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` },
           );
           return;
         }
         if (parsed.test_capacity_before_success_times) {
-          const capacityKey = `${req.url}:capacity-before-success:${parsed.test_capacity_before_success_times}`;
+          const capacityKey = [
+            req.url,
+            "capacity-before-success",
+            parsed.test_capacity_before_success_times,
+            parsed.stream ? "stream" : "non-stream",
+            parsed.test_capacity_status ?? "default",
+            parsed.test_capacity_stream_event_name || "",
+            parsed.test_capacity_stream_payload_shape || "",
+          ].join(":");
           const capacityCount = (capacityBeforeSuccessCounts.get(capacityKey) || 0) + 1;
           capacityBeforeSuccessCounts.set(capacityKey, capacityCount);
           if (capacityCount <= parsed.test_capacity_before_success_times) {
+            const capacityMessage = parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.";
             if (parsed.stream) {
               createCapacityErrorSseResponse(
                 res,
-                parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.",
+                capacityMessage,
+                20,
+                {
+                  eventName: parsed.test_capacity_stream_event_name || "error",
+                  payloadShape: parsed.test_capacity_stream_payload_shape || "default",
+                },
               );
               return;
             }
@@ -170,11 +209,11 @@ function startFakeUpstream(port) {
               parsed.test_capacity_status ?? 503,
               {
                 error: {
-                  message: parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.",
+                  message: capacityMessage,
                   type: "server_error",
                 },
               },
-              { "x-upstream-test": "capacity-error" },
+              { "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` },
             );
             return;
           }
@@ -183,6 +222,11 @@ function startFakeUpstream(port) {
           createCapacityErrorSseResponse(
             res,
             parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.",
+            20,
+            {
+              eventName: parsed.test_capacity_stream_event_name || "error",
+              payloadShape: parsed.test_capacity_stream_payload_shape || "default",
+            },
           );
           return;
         }
@@ -432,6 +476,22 @@ async function run() {
       "capacity error 返回体未保留 upstream status",
     );
 
+    const capacityStatus200Response = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ test_capacity_error: true, test_capacity_status: 200 }),
+    });
+    const capacityStatus200Body = await capacityStatus200Response.json();
+    assert(capacityStatus200Response.status === 502, `200+capacity error 未返回 502: ${capacityStatus200Response.status}`);
+    assert(
+      capacityStatus200Body?.error?.code === "upstream_error_retry_triggered",
+      "200+capacity error 返回体未标记 retry trigger",
+    );
+    assert(
+      capacityStatus200Body?.error?.upstream_status_code === 200,
+      "200+capacity error 返回体未保留 upstream status",
+    );
+
     const capacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
       method: "POST",
       headers: { "content-type": "application/json" },
@@ -451,6 +511,25 @@ async function run() {
     );
     assert(capacityRecoveredEntry, "capacity 抖动恢复后的请求记录未保留重试次数");
 
+    const capacityStatus200RecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ test_capacity_before_success_times: 2, test_capacity_status: 200, test_reasoning_tokens: 128 }),
+    });
+    const capacityStatus200RecoveredBody = await capacityStatus200RecoveredResponse.json();
+    assert(capacityStatus200RecoveredResponse.status === 200, `200+capacity 抖动后未自动恢复: ${capacityStatus200RecoveredResponse.status}`);
+    assert(
+      capacityStatus200RecoveredBody?.usage?.output_tokens_details?.reasoning_tokens === 128,
+      "200+capacity 抖动恢复后的返回体异常",
+    );
+
+    const requestsAfterCapacityStatus200RecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=30`);
+    const requestsAfterCapacityStatus200Recovery = await requestsAfterCapacityStatus200RecoveryResponse.json();
+    const capacityStatus200RecoveredEntry = requestsAfterCapacityStatus200Recovery?.entries?.find(
+      (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.upstream_attempt_count >= 3,
+    );
+    assert(capacityStatus200RecoveredEntry, "200+capacity 抖动恢复后的请求记录未保留重试次数");
+
     const streamCapacityResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
       method: "POST",
       headers: { "content-type": "application/json" },
@@ -463,6 +542,23 @@ async function run() {
       "stream+capacity error 返回体未标记 retry trigger",
     );
 
+    const streamCapacityResponseFailed = 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_stream_event_name: "response.failed",
+        test_capacity_stream_payload_shape: "response_failed",
+      }),
+    });
+    const streamCapacityResponseFailedBody = await streamCapacityResponseFailed.json();
+    assert(streamCapacityResponseFailed.status === 502, `stream+response.failed capacity error 未返回 502: ${streamCapacityResponseFailed.status}`);
+    assert(
+      streamCapacityResponseFailedBody?.error?.code === "upstream_error_retry_triggered",
+      "stream+response.failed capacity error 返回体未标记 retry trigger",
+    );
+
     const streamCapacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
       method: "POST",
       headers: { "content-type": "application/json" },
@@ -479,6 +575,28 @@ async function run() {
     );
     assert(streamCapacityRecoveredEntry, "stream capacity 抖动恢复后的请求记录未保留重试次数");
 
+    const streamCapacityResponseFailedRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        stream: true,
+        test_capacity_before_success_times: 2,
+        test_reasoning_tokens: 128,
+        test_capacity_stream_event_name: "response.failed",
+        test_capacity_stream_payload_shape: "response_failed",
+      }),
+    });
+    const streamCapacityResponseFailedRecoveredText = await streamCapacityResponseFailedRecoveredResponse.text();
+    assert(streamCapacityResponseFailedRecoveredResponse.status === 200, `stream response.failed capacity 抖动后未自动恢复: ${streamCapacityResponseFailedRecoveredResponse.status}`);
+    assert(streamCapacityResponseFailedRecoveredText.includes("hello"), "stream response.failed capacity 恢复后未拿到正常 SSE 内容");
+
+    const requestsAfterStreamCapacityResponseFailedRecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=30`);
+    const requestsAfterStreamCapacityResponseFailedRecovery = await requestsAfterStreamCapacityResponseFailedRecoveryResponse.json();
+    const streamCapacityResponseFailedRecoveredEntry = requestsAfterStreamCapacityResponseFailedRecovery?.entries?.find(
+      (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.response_stream && entry.upstream_attempt_count >= 3,
+    );
+    assert(streamCapacityResponseFailedRecoveredEntry, "stream response.failed capacity 恢复后的请求记录未保留重试次数");
+
     for (const streamPath of [
       "/responses",
       "/v1/responses",