retry stream capacity errors in gateway

gateway.mjs

@@ -6,6 +6,7 @@ import { chmod, copyFile, mkdir, readFile, readdir, rm, writeFile } from "node:f
 import fs from "node:fs";
 import path from "node:path";
 import { DatabaseSync } from "node:sqlite";
+import zlib from "node:zlib";
 import { TextDecoder } from "node:util";
 import { fileURLToPath } from "node:url";
 
@@ -23,6 +24,7 @@ const PROFILE_PROBE_API_PATH = `${ADMIN_BASE_PATH}/api/profiles/probe`;
 const PROFILE_SWITCH_API_PATH = `${ADMIN_BASE_PATH}/api/profiles/switch`;
 const PROFILE_ITEM_API_PREFIX = `${ADMIN_BASE_PATH}/api/profiles/`;
 const RESTORE_API_PATH = `${ADMIN_BASE_PATH}/api/restore`;
+const STATUS_REASONING_COUNT_LIMIT = 24;
 
 const DEFAULT_CONFIG = {
   profile_name: "default",
@@ -1091,6 +1093,14 @@ async function writeProfile(runtime, payload) {
 function buildMetricsSnapshot(monitor) {
   const reasoning516Count = monitor.observed_reasoning_counts["516"] || 0;
   const inspectedResponseCount = monitor.inspected_response_count;
+  const reasoningEntries = Object.entries(monitor.observed_reasoning_counts).sort((left, right) => {
+    const countDelta = Number(right[1] || 0) - Number(left[1] || 0);
+    if (countDelta !== 0) {
+      return countDelta;
+    }
+    return Number(left[0] || 0) - Number(right[0] || 0);
+  });
+  const visibleReasoningEntries = reasoningEntries.slice(0, STATUS_REASONING_COUNT_LIMIT);
   return {
     started_at: monitor.started_at,
     persistent_since: monitor.persistent_since,
@@ -1101,7 +1111,12 @@ function buildMetricsSnapshot(monitor) {
     reasoning_516_ratio:
       inspectedResponseCount === 0 ? 0 : reasoning516Count / inspectedResponseCount,
     token_totals: { ...monitor.token_totals },
-    observed_reasoning_counts: { ...monitor.observed_reasoning_counts },
+    observed_reasoning_counts: Object.fromEntries(visibleReasoningEntries),
+    observed_reasoning_counts_total_keys: reasoningEntries.length,
+    observed_reasoning_counts_omitted: Math.max(
+      0,
+      reasoningEntries.length - visibleReasoningEntries.length,
+    ),
   };
 }
 
@@ -1758,12 +1773,65 @@ async function restoreRuntimeState(runtime, state) {
   ]);
 }
 
-function jsonResponse(res, statusCode, payload, headers = {}) {
+function pickContentEncoding(acceptEncoding = "") {
+  const value = `${acceptEncoding}`.toLowerCase();
+  if (value.includes("br")) {
+    return "br";
+  }
+  if (value.includes("gzip")) {
+    return "gzip";
+  }
+  return null;
+}
+
+function shouldCompressContent(headers = {}) {
+  const contentType = `${headers["content-type"] || headers["Content-Type"] || ""}`.toLowerCase();
+  return (
+    contentType.startsWith("text/") ||
+    contentType.includes("javascript") ||
+    contentType.includes("json") ||
+    contentType.includes("xml") ||
+    contentType.includes("svg")
+  );
+}
+
+function maybeCompressBody(body, headers = {}, acceptEncoding = "") {
+  if (!body || body.length < 1024 || !shouldCompressContent(headers)) {
+    return { body, encoding: null };
+  }
+  const encoding = pickContentEncoding(acceptEncoding);
+  if (encoding === "br") {
+    return { body: zlib.brotliCompressSync(body), encoding };
+  }
+  if (encoding === "gzip") {
+    return { body: zlib.gzipSync(body), encoding };
+  }
+  return { body, encoding: null };
+}
+
+function respondBuffer(res, statusCode, body, headers = {}, acceptEncoding = "") {
+  const { body: responseBody, encoding } = maybeCompressBody(body, headers, acceptEncoding);
   res.writeHead(statusCode, {
-    "content-type": "application/json; charset=utf-8",
+    "content-length": responseBody.length,
+    vary: "accept-encoding",
+    ...(encoding ? { "content-encoding": encoding } : {}),
     ...headers,
   });
-  res.end(JSON.stringify(payload));
+  res.end(responseBody);
+}
+
+function jsonResponse(req, res, statusCode, payload, headers = {}) {
+  const body = Buffer.from(JSON.stringify(payload));
+  respondBuffer(
+    res,
+    statusCode,
+    body,
+    {
+      "content-type": "application/json; charset=utf-8",
+      ...headers,
+    },
+    req?.headers?.["accept-encoding"] || "",
+  );
 }
 
 
@@ -1794,26 +1862,31 @@ function safeJoinStatic(root, requestPath) {
   return fullPath;
 }
 
-async function serveStaticFile(res, filePath) {
+async function serveStaticFile(req, res, filePath) {
   try {
     const body = await readFile(filePath);
-    res.writeHead(200, {
+    respondBuffer(
-      "content-type": contentTypeForFile(filePath),
+      res,
-      "cache-control": filePath.includes(`${path.sep}assets${path.sep}`)
+      200,
-        ? "public, max-age=31536000, immutable"
+      body,
-        : "no-cache",
+      {
-    });
+        "content-type": contentTypeForFile(filePath),
-    res.end(body);
+        "cache-control": filePath.includes(`${path.sep}assets${path.sep}`)
+          ? "public, max-age=31536000, immutable"
+          : "no-cache",
+      },
+      req?.headers?.["accept-encoding"] || "",
+    );
     return true;
   } catch {
     return false;
   }
 }
 
-async function serveManagementUi(res, requestPathname) {
+async function serveManagementUi(req, res, requestPathname) {
   const uiPrefix = `${UI_PATH}/`;
   if (requestPathname === UI_PATH || requestPathname === `${UI_PATH}/`) {
-    return serveStaticFile(res, path.join(UI_STATIC_ROOT, "index.html"));
+    return serveStaticFile(req, res, path.join(UI_STATIC_ROOT, "index.html"));
   }
 
   if (!requestPathname.startsWith(uiPrefix)) {
@@ -1822,7 +1895,7 @@ async function serveManagementUi(res, requestPathname) {
 
   const staticPath = safeJoinStatic(UI_STATIC_ROOT, requestPathname.slice(uiPrefix.length));
   if (!staticPath) {
-    jsonResponse(res, 403, {
+    jsonResponse(req, res, 403, {
       error: {
         message: "invalid static path",
         code: "invalid_static_path",
@@ -1831,11 +1904,11 @@ async function serveManagementUi(res, requestPathname) {
     return true;
   }
 
-  if (await serveStaticFile(res, staticPath)) {
+  if (await serveStaticFile(req, res, staticPath)) {
     return true;
   }
 
-  return serveStaticFile(res, path.join(UI_STATIC_ROOT, "index.html"));
+  return serveStaticFile(req, res, path.join(UI_STATIC_ROOT, "index.html"));
 }
 
 function buildEditableConfig(currentConfig, payload) {
@@ -1913,8 +1986,8 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
   const pathname = normalizePath(requestUrl.pathname);
 
   if (pathname === UI_PATH || pathname.startsWith(`${UI_PATH}/`)) {
-    if (!(await serveManagementUi(res, requestUrl.pathname))) {
+    if (!(await serveManagementUi(req, res, requestUrl.pathname))) {
-      jsonResponse(res, 503, {
+      jsonResponse(req, res, 503, {
         error: {
           message: "UI assets were not built. Run: npm run build:ui",
           code: "ui_not_built",
@@ -1926,7 +1999,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
 
   if (pathname === STATUS_API_PATH && req.method === "GET") {
     const state = await readRuntimeState(runtime);
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       listen: `${runtime.config.listen_host}:${runtime.config.listen_port}`,
       config: sanitizeConfigForStatus(runtime.config),
@@ -1949,7 +2022,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     const limitRaw = requestUrl.searchParams.get("limit");
     const sinceSeq = sinceSeqRaw === null ? null : Number.parseInt(sinceSeqRaw, 10);
     const limit = limitRaw === null ? 500 : Number.parseInt(limitRaw, 10);
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       ...await buildPersistentLogsSnapshot(
         runtime,
@@ -1967,7 +2040,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     const filter = requestUrl.searchParams.get("filter") || "all";
     const limit = limitRaw === null ? 50 : Number.parseInt(limitRaw, 10);
     const offset = offsetRaw === null ? 0 : Number.parseInt(offsetRaw, 10);
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       ...await buildPersistentRequestsSnapshot(runtime, {
         limit: Number.isInteger(limit) ? limit : 50,
@@ -1980,7 +2053,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
   }
 
   if (pathname === PROFILES_API_PATH && req.method === "GET") {
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       profiles_dir: runtime.paths.profilesDir,
       active_profile: runtime.config.profile_name || "default",
@@ -1993,7 +2066,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     const body = await readRequestBody(req, runtime.config.request_body_limit_bytes);
     const payload = parseJsonSafely(body);
     if (!payload) {
-      jsonResponse(res, 400, {
+      jsonResponse(req, res, 400, {
         error: {
           message: "profile 保存请求必须是有效 JSON",
           code: "invalid_json",
@@ -2008,7 +2081,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
       applied = await applyProfileConfig(runtime, result.name);
     }
     runtime.logger(`[profile] saved name=${result.name} path=${result.file_path}`);
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       message: applied ? "profile 已保存并已热应用" : "profile 已保存",
       saved_profile: result,
@@ -2024,7 +2097,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     const body = await readRequestBody(req, runtime.config.request_body_limit_bytes);
     const payload = parseJsonSafely(body);
     if (!payload) {
-      jsonResponse(res, 400, {
+      jsonResponse(req, res, 400, {
         error: {
           message: "profile probe 请求必须是有效 JSON",
           code: "invalid_json",
@@ -2037,7 +2110,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     runtime.logger(
       `[profile-probe] profile=${result.profile} auth=${result.auth_mode}/${result.auth_source} upstream=${result.upstream_base_url}`,
     );
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       ...result,
     });
@@ -2049,7 +2122,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     const payload = parseJsonSafely(body);
     const profileName = `${payload?.profile || ""}`.trim();
     if (!profileName) {
-      jsonResponse(res, 400, {
+      jsonResponse(req, res, 400, {
         error: {
           message: "缺少 profile",
           code: "profile_required",
@@ -2059,7 +2132,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     }
 
     const result = await applyProfileConfig(runtime, profileName);
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       message: "profile 已热切换，无需重启 gateway",
       ...result,
@@ -2070,7 +2143,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
   if (pathname.startsWith(PROFILE_ITEM_API_PREFIX) && req.method === "DELETE") {
     const rawName = pathname.slice(PROFILE_ITEM_API_PREFIX.length);
     if (!rawName || rawName.includes("/")) {
-      jsonResponse(res, 400, {
+      jsonResponse(req, res, 400, {
         error: {
           message: "无效的 profile 名称",
           code: "invalid_profile_name",
@@ -2082,7 +2155,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     const profileName = decodeURIComponent(rawName);
     const result = await deleteProfile(runtime, profileName);
     runtime.logger(`[profile] deleted name=${result.name} path=${result.file_path}`);
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       message: "profile 已删除",
       deleted_profile: result,
@@ -2097,7 +2170,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
     const body = await readRequestBody(req, runtime.config.request_body_limit_bytes);
     const payload = parseJsonSafely(body);
     if (!payload) {
-      jsonResponse(res, 400, {
+      jsonResponse(req, res, 400, {
         error: {
           message: "配置保存请求必须是有效 JSON",
           code: "invalid_json",
@@ -2113,7 +2186,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
       `[config] updated reasoning_equals=${nextConfig.reasoning_equals.join(",")} retryable_status_codes=${nextConfig.retryable_status_codes.join(",")} endpoints=${nextConfig.endpoints.join(",")}`,
     );
     const state = await readRuntimeState(runtime);
-    jsonResponse(res, 200, {
+    jsonResponse(req, res, 200, {
       ok: true,
       message: "配置已保存并立即生效",
       config: sanitizeConfigForStatus(runtime.config),
@@ -2132,7 +2205,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
   if (pathname === RESTORE_API_PATH && req.method === "POST") {
     const state = await readRuntimeState(runtime);
     if (!state) {
-      jsonResponse(res, 409, {
+      jsonResponse(req, res, 409, {
         error: {
           message: "当前未检测到安装状态，无法恢复 Codex 原设置",
           code: "state_not_found",
@@ -2143,7 +2216,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
 
     await restoreRuntimeState(runtime, state);
     runtime.logger(`[restore] restored via UI state_root=${runtime.paths.stateRoot}`);
-    jsonResponse(res, 202, {
+    jsonResponse(req, res, 202, {
       ok: true,
       message: "原设置已恢复，gateway 即将关闭",
     });
@@ -2366,6 +2439,18 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody,
     return null;
   }
 
+  return matchRetryableMessage(config, parsedBody, bodyText);
+}
+
+function matchRetryableMessage(config, parsedBody, bodyText) {
+  const retryableMessages = normalizePhraseList(
+    config.retryable_error_messages,
+    DEFAULT_CONFIG.retryable_error_messages,
+  );
+  if (retryableMessages.length === 0) {
+    return null;
+  }
+
   const normalizedPatterns = retryableMessages.map((message) => ({
     original: message,
     normalized: message.toLowerCase(),
@@ -2395,6 +2480,32 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody,
   return null;
 }
 
+function isRetryableStreamErrorShape(parsedBody, eventName = "") {
+  if (`${eventName || ""}`.trim().toLowerCase() === "error") {
+    return true;
+  }
+  if (!parsedBody || typeof parsedBody !== "object" || Array.isArray(parsedBody)) {
+    return false;
+  }
+  if (Object.hasOwn(parsedBody, "error")) {
+    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)) {
+    return null;
+  }
+  return matchRetryableMessage(config, parsedBody, bodyText);
+}
+
 function isExpectedStreamTermination(error) {
   if (!error) {
     return false;
@@ -2535,13 +2646,14 @@ async function fetchUpstreamWithRetry(upstreamUrl, init, config, logger, request
   };
 }
 
-function inspectSseChunk(state, chunk) {
+function inspectSseChunk(state, chunk, config) {
   const decoded = state.decoder.decode(chunk, { stream: true });
   state.buffer += decoded;
 
   const result = {
     reasoning: null,
     usage: null,
+    retryable_upstream_error: null,
   };
 
   const blocks = state.buffer.split(/\r?\n\r?\n/);
@@ -2552,6 +2664,10 @@ function inspectSseChunk(state, chunk) {
       .split(/\r?\n/)
       .map((line) => line.trimEnd())
       .filter(Boolean);
+    const eventName = lines
+      .filter((line) => line.startsWith("event:"))
+      .map((line) => line.replace(/^event:\s?/, "").trim())
+      .find(Boolean) || "";
     const dataLines = lines
       .filter((line) => line.startsWith("data:"))
       .map((line) => line.replace(/^data:\s?/, ""));
@@ -2563,8 +2679,9 @@ function inspectSseChunk(state, chunk) {
     if (payloadText === "[DONE]") {
       continue;
     }
+    let parsed = null;
     try {
-      const parsed = JSON.parse(payloadText);
+      parsed = JSON.parse(payloadText);
       const reasoning = extractReasoningTokens(parsed);
       if (reasoning !== null) {
         result.reasoning = reasoning;
@@ -2573,6 +2690,15 @@ function inspectSseChunk(state, chunk) {
     } catch {
       // ignore malformed SSE payloads
     }
+    const retryableUpstreamError = findRetryableStreamErrorMatch(
+      config,
+      parsed,
+      payloadText,
+      eventName,
+    );
+    if (retryableUpstreamError) {
+      result.retryable_upstream_error = retryableUpstreamError;
+    }
   }
   return result;
 }
@@ -2764,8 +2890,27 @@ async function handleStreaming({
 
     const chunkBuffer = Buffer.from(value);
     const now = new Date();
+    const inspection = inspectSseChunk(sseState, value, config);
+    const retryableUpstreamError = inspection.retryable_upstream_error;
+    if (retryableUpstreamError) {
+      abortController.abort();
+      reader.cancel().catch(() => {});
+      return {
+        inspected: true,
+        matched: true,
+        retry_requested: strict502Mode || !wroteAnyChunk,
+        retryable_upstream_error: retryableUpstreamError,
+        upstream_status_code: upstreamResponse.status,
+        reasoning_tokens: observedReasoning,
+        usage: observedUsage,
+        error: `retryable upstream error: ${retryableUpstreamError.matched_pattern}`,
+        match_reason: "retryable_upstream_error",
+        response_bytes_received: requestEntry.response_bytes_received,
+        stream_chunk_count: requestEntry.stream_chunk_count,
+      };
+    }
+
     markAndPersistFirstResponse(runtime, requestEntry, now);
-    const inspection = inspectSseChunk(sseState, value);
     const reasoning = inspection.reasoning;
     const usageUpdated = Boolean(inspection.usage);
     observedUsage = mergeUsageSnapshots(observedUsage, inspection.usage);
@@ -2830,6 +2975,10 @@ async function handleStreaming({
 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"}`);
@@ -2880,6 +3029,7 @@ 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.model = requestJson?.model || null;
     requestEntry.requested_model = parsedRequestJson?.model || null;
@@ -2889,7 +3039,6 @@ async function proxyRequest(runtime, req, res) {
     upsertRequestEntry(runtime, requestEntry);
 
     const upstreamUrl = buildUpstreamUrl(config.upstream_base_url, incomingUrl);
-    const abortController = new AbortController();
     const upstreamAuth = await resolveUpstreamAuth(config);
     requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth });
     upsertRequestEntry(runtime, requestEntry);
@@ -2899,57 +3048,129 @@ async function proxyRequest(runtime, req, res) {
       );
     }
 
-    const {
+    while (totalUpstreamAttempts < maxUpstreamAttempts) {
-      response: upstreamResponse,
+      const abortController = new AbortController();
-      attempt_count: upstreamAttemptCount,
+      const {
-      retryable_upstream_error: terminalRetryableUpstreamError,
+        response: upstreamResponse,
-    } = await fetchUpstreamWithRetry(upstreamUrl, {
+        attempt_count: upstreamAttemptCount,
-      method: req.method,
+        retryable_upstream_error: terminalRetryableUpstreamError,
-      headers: cloneHeadersForUpstream(req.headers, upstreamAuth),
+      } = await fetchUpstreamWithRetry(upstreamUrl, {
-      body: requestBody.length > 0 ? requestBody : undefined,
+        method: req.method,
-      signal: abortController.signal,
+        headers: cloneHeadersForUpstream(req.headers, upstreamAuth),
-    }, config, logger, { method: req.method, pathname });
+        body: requestBody.length > 0 ? requestBody : undefined,
+        signal: abortController.signal,
+      }, {
+        ...config,
+        upstream_fetch_retry_attempts: Math.max(1, maxUpstreamAttempts - totalUpstreamAttempts),
+      }, logger, { method: req.method, pathname });
 
-    const shouldInspect = matchPath(config, pathname);
+      totalUpstreamAttempts += upstreamAttemptCount;
-    const responseContentType = upstreamResponse.headers.get("content-type");
+
-    const responseIsStream = isSseContentType(responseContentType) || (
+      const shouldInspect = matchPath(config, pathname);
-      requestIsStream &&
+      const responseContentType = upstreamResponse.headers.get("content-type");
-      !isJsonContentType(responseContentType) &&
+      const responseIsStream = isSseContentType(responseContentType) || (
-      !isUpstreamErrorStatus(upstreamResponse.status)
+        requestIsStream &&
-    );
+        !isJsonContentType(responseContentType) &&
-    requestEntry.response_stream = responseIsStream;
+        !isUpstreamErrorStatus(upstreamResponse.status)
-    requestEntry.inspected = shouldInspect;
-    requestEntry.upstream_status_code = upstreamResponse.status;
-    requestEntry.upstream_attempt_count = upstreamAttemptCount;
-    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 || "-"}`,
       );
-    }
+      requestEntry.response_stream = responseIsStream;
-
+      requestEntry.inspected = shouldInspect;
-    if (!shouldInspect) {
+      requestEntry.upstream_status_code = upstreamResponse.status;
-      markRequestFirstResponse(requestEntry);
+      requestEntry.upstream_attempt_count = totalUpstreamAttempts;
+      requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth, upstreamResponse });
       upsertRequestEntry(runtime, requestEntry);
-      copyHeadersToClient(upstreamResponse.headers, res);
+      if (isUpstreamErrorStatus(upstreamResponse.status)) {
-      res.writeHead(upstreamResponse.status);
+        logger?.(
-      const body = Buffer.from(await upstreamResponse.arrayBuffer());
+          `[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 || "-"}`,
-      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) {
+      if (!shouldInspect) {
-      const result = await handleStreaming({
+        markRequestFirstResponse(requestEntry);
+        upsertRequestEntry(runtime, requestEntry);
+        copyHeadersToClient(upstreamResponse.headers, res);
+        res.writeHead(upstreamResponse.status);
+        const body = Buffer.from(await upstreamResponse.arrayBuffer());
+        res.end(body);
+        recordRequestEntry(
+          runtime,
+          finalizeRequestEntry(requestEntry, {
+            status_code: upstreamResponse.status,
+            upstream_status_code: upstreamResponse.status,
+            inspected: false,
+          }),
+          config.request_history_limit,
+        );
+        return;
+      }
+
+      if (responseIsStream) {
+        const result = await handleStreaming({
+          runtime,
+          config,
+          logger,
+          monitor: runtime.monitor,
+          pathname,
+          upstreamResponse,
+          res,
+          abortController,
+          requestEntry,
+        });
+        if (result.retry_requested) {
+          if (totalUpstreamAttempts < maxUpstreamAttempts) {
+            const backoffMs = computeRetryBackoffMs(config, totalUpstreamAttempts);
+            logger?.(
+              `[retry] upstream retryable stream error attempt=${totalUpstreamAttempts} next_attempt=${totalUpstreamAttempts + 1} status=${upstreamResponse.status} path=${pathname || "-"} reason=${JSON.stringify(result.retryable_upstream_error?.matched_pattern)} backoff_ms=${backoffMs}`,
+            );
+            if (backoffMs > 0) {
+              await sleep(backoffMs);
+            }
+            continue;
+          }
+
+          recordInspectedResponse(runtime.monitor, result.reasoning_tokens, true);
+          if (config.log_match) {
+            logger?.(
+              `[match] stream path=${pathname} upstream_status=${upstreamResponse.status} retryable_error=${JSON.stringify(result.retryable_upstream_error?.matched_pattern)} action=status_${config.non_stream_status_code}`,
+            );
+          }
+          const blockedBody = buildRetryableUpstreamErrorBody(
+            pathname,
+            upstreamResponse.status,
+            result.retryable_upstream_error?.matched_message || result.retryable_upstream_error?.matched_pattern,
+            config.non_stream_status_code,
+          );
+          res.writeHead(config.non_stream_status_code, {
+            "content-type": "application/json; charset=utf-8",
+            "x-codex-retry-gateway-reason": "upstream-error-retry-triggered",
+          });
+          res.end(blockedBody);
+          recordRequestEntry(
+            runtime,
+            finalizeRequestEntry(requestEntry, {
+              response_stream: true,
+              ...result,
+              matched: true,
+              status_code: config.non_stream_status_code,
+              upstream_status_code: upstreamResponse.status,
+            }),
+            config.request_history_limit,
+          );
+          return;
+        }
+
+        recordRequestEntry(
+          runtime,
+          finalizeRequestEntry(requestEntry, {
+            response_stream: true,
+            ...result,
+          }),
+          config.request_history_limit,
+        );
+        return;
+      }
+
+      const result = await handleNonStreaming({
         runtime,
         config,
         logger,
@@ -2957,43 +3178,22 @@ async function proxyRequest(runtime, req, res) {
         pathname,
         upstreamResponse,
         res,
-        abortController,
         requestEntry,
+        terminalRetryableUpstreamError,
       });
       recordRequestEntry(
         runtime,
         finalizeRequestEntry(requestEntry, {
-          response_stream: true,
+          response_stream: false,
           ...result,
         }),
         config.request_history_limit,
       );
       return;
     }
-
-    const result = await handleNonStreaming({
-      runtime,
-      config,
-      logger,
-      monitor: runtime.monitor,
-      pathname,
-      upstreamResponse,
-      res,
-      requestEntry,
-      terminalRetryableUpstreamError,
-    });
-    recordRequestEntry(
-      runtime,
-      finalizeRequestEntry(requestEntry, {
-        response_stream: false,
-        ...result,
-      }),
-      config.request_history_limit,
-    );
-    return;
   } catch (error) {
     if (Number.isInteger(error?.gatewayAttemptCount)) {
-      requestEntry.upstream_attempt_count = error.gatewayAttemptCount;
+      requestEntry.upstream_attempt_count = (requestEntry.upstream_attempt_count || 0) + error.gatewayAttemptCount;
     }
     recordRequestEntry(
       runtime,

scripts/test-gateway-e2e.mjs

@@ -80,6 +80,21 @@ function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) {
   }, destroyDelayMs);
 }
 
+function createCapacityErrorSseResponse(
+  res,
+  message = "Selected model is at capacity. Please try a different model.",
+  intervalMs = 20,
+) {
+  createSseResponse(
+    res,
+    [
+      'event: error\n',
+      `data: ${JSON.stringify({ error: { message, type: "server_error" } })}\n\n`,
+    ],
+    intervalMs,
+  );
+}
+
 function startFakeUpstream(port) {
   const failBeforeResponseCounts = new Map();
   const capacityBeforeSuccessCounts = new Map();
@@ -143,6 +158,13 @@ function startFakeUpstream(port) {
           const capacityCount = (capacityBeforeSuccessCounts.get(capacityKey) || 0) + 1;
           capacityBeforeSuccessCounts.set(capacityKey, capacityCount);
           if (capacityCount <= parsed.test_capacity_before_success_times) {
+            if (parsed.stream) {
+              createCapacityErrorSseResponse(
+                res,
+                parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.",
+              );
+              return;
+            }
             createJsonResponse(
               res,
               parsed.test_capacity_status ?? 503,
@@ -157,6 +179,13 @@ function startFakeUpstream(port) {
             return;
           }
         }
+        if (parsed.stream && parsed.test_capacity_error) {
+          createCapacityErrorSseResponse(
+            res,
+            parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.",
+          );
+          return;
+        }
         if (parsed.stream) {
           createSseResponse(res, [
             'data: {"type":"response.output_text.delta","delta":"hello"}\n\n',
@@ -434,6 +463,22 @@ async function run() {
       "stream+capacity error 返回体未标记 retry trigger",
     );
 
+    const streamCapacityRecoveredResponse = 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 }),
+    });
+    const streamCapacityRecoveredText = await streamCapacityRecoveredResponse.text();
+    assert(streamCapacityRecoveredResponse.status === 200, `stream capacity 抖动后未自动恢复: ${streamCapacityRecoveredResponse.status}`);
+    assert(streamCapacityRecoveredText.includes("hello"), "stream capacity 恢复后未拿到正常 SSE 内容");
+
+    const requestsAfterStreamCapacityRecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=20`);
+    const requestsAfterStreamCapacityRecovery = await requestsAfterStreamCapacityRecoveryResponse.json();
+    const streamCapacityRecoveredEntry = requestsAfterStreamCapacityRecovery?.entries?.find(
+      (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.response_stream && entry.upstream_attempt_count >= 3,
+    );
+    assert(streamCapacityRecoveredEntry, "stream capacity 抖动恢复后的请求记录未保留重试次数");
+
     for (const streamPath of [
       "/responses",
       "/v1/responses",

ui-src/src/App.tsx

@@ -39,6 +39,8 @@ type Metrics = {
     cached_tokens?: number;
   };
   observed_reasoning_counts?: Record<string, number>;
+  observed_reasoning_counts_total_keys?: number;
+  observed_reasoning_counts_omitted?: number;
 };
 
 type StatusPayload = {
@@ -222,7 +224,7 @@ const api = {
   restore: "/__codex_retry_gateway/api/restore",
 };
 
-const REQUEST_PAGE_SIZE = 40;
+const REQUEST_PAGE_SIZE = 20;
 const LOG_PAGE_SIZE = 200;
 
 const zhNumberFormatter = new Intl.NumberFormat("zh-CN");
@@ -666,7 +668,7 @@ export default function App() {
   }, [page, requestQuery, requestFilter, latestLogSeq, requestLimit]);
 
   useEffect(() => {
-    if (page === "overview" || page === "rules") {
+    if (page === "overview" || page === "rules" || page === "requests") {
       return;
     }
     loadPageData(page, { incrementalLogs: false }).catch((error) => {
@@ -997,11 +999,18 @@ export default function App() {
                       {reasoningChips.length === 0 ? (
                         <span className="chip">还没有 reasoning 观测</span>
                       ) : (
-                        reasoningChips.map(([reasoning, count]) => (
+                        <>
-                          <span className="chip" key={reasoning}>
+                          {reasoningChips.map(([reasoning, count]) => (
-                            reasoning {reasoning}: {count}
+                            <span className="chip" key={reasoning}>
-                          </span>
+                              reasoning {reasoning}: {count}
-                        ))
+                            </span>
+                          ))}
+                          {(metrics.observed_reasoning_counts_omitted || 0) > 0 ? (
+                            <span className="chip">
+                              其余 {metrics.observed_reasoning_counts_omitted} 项未展开
+                            </span>
+                          ) : null}
+                        </>
                       )}
                     </div>
                   </Card>