Merge pull request #1 from nonononull/codex/strict-502-streaming-516-20260626

Add strict 502 mode and upstream retry handling for streaming responses

README.md

@@ -7,7 +7,7 @@
 - 保持 Codex 继续使用现有 `auth.json`
 - 只把 `config.toml` 的当前 provider `base_url` 改成本地网关
 - 非流式命中 `reasoning_tokens = 516` 时返回 `502`
-- 流式命中时直接断开连接，让 Codex 自行重试
+- 流式命中时默认先缓存并判断；一旦命中 `516`，统一返回 `502`
 - 默认同时拦截 root 路径和 `/v1` 路径：
   - `/responses`
   - `/chat/completions`
@@ -191,7 +191,9 @@ macOS / Linux: ~/.codex-retry-gateway/config/config.json
 - `non_stream_status_code`
   - 默认 `502`
 - `stream_action`
-  - 默认 `disconnect`
+  - 默认 `strict_502`
+  - `strict_502`：先缓存整个流，命中 `516` 时统一返回 `502`
+  - `disconnect`：兼容旧行为；若命中发生在已透传 chunk 之后，则直接断开连接
 - `log_match`
   - 是否记录命中日志
 

config.example.json

@@ -6,7 +6,7 @@
   "endpoints": ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"],
   "reasoning_equals": [516],
   "non_stream_status_code": 502,
-  "stream_action": "disconnect",
+  "stream_action": "strict_502",
   "log_match": true,
   "health_path": "/__codex_retry_gateway/health"
 }

err.md

@@ -7,9 +7,9 @@
 - 只解决 Codex 已可访问上游时的 `reasoning_tokens = 516` 重试问题
 - 不替代 `cc-switch` 的协议路由转换
 - 流式场景默认策略是：
-  - 先实时透传
+  - 先缓存上游流
   - 一旦检测到命中 `516`
-  - 直接断开连接
+  - 统一返回 `502`
 
 ### 当前已知限制
 
@@ -111,6 +111,32 @@
      - `.sh` 优先选择 `node.exe`
      - 在 WSL / Bash 场景下把路径参数转换回 Windows 路径后再交给 `node.exe`
 
+14. 上游流式连接中途终止时被误记为网关错误，首次瞬断也缺少最小重试
+   - 现象：
+     - 日志出现：
+       - `TypeError: terminated`
+       - `TypeError: fetch failed`
+     - 其中一部分来自上游 SSE 中途断流，另一部分来自上游首次连接瞬时失败
+   - 根因：
+     - `handleStreaming()` 直接把 `reader.read()` 抛出的 `AbortError` / `TypeError: terminated` 冒到统一错误处理
+     - `proxyRequest()` 对上游 `fetch()` 没有做一次轻量重试，首个瞬断会直接返回 `502`
+   - 处理：
+     - 新增预期流终止识别：
+       - `AbortError`
+       - `TypeError: terminated`
+     - 这两类在流式处理中按“连接已结束”收口，不再记 `[error]`
+     - 新增上游 `fetch failed` 的一次自动重试
+     - 新增严格 `502` 流式模式：
+       - 默认不再抢先透传 `200` 头和首个 chunk
+       - 先缓存流，再根据 `reasoning_tokens` 决定透传或返回 `502`
+   - 验证：
+     - `scripts/test-gateway-e2e.mjs`
+       - 新增 `/responses` 流式覆盖
+       - 新增“上游半路断流不刷 error 日志”断言
+       - 新增“首次 fetch failed 后第二次成功恢复”断言
+       - 新增“流式 `516` 统一返回 `502`，不再先透传半截 chunk”断言
+     - `scripts/test-install-restore.mjs` 继续通过
+
 ### 2026-06-26 实测证据
 
 - 假上游 E2E

gateway.mjs

@@ -24,7 +24,7 @@ const DEFAULT_CONFIG = {
   endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"],
   reasoning_equals: [516],
   non_stream_status_code: 502,
-  stream_action: "disconnect",
+  stream_action: "strict_502",
   log_match: true,
   health_path: "/__codex_retry_gateway/health",
 };
@@ -63,7 +63,7 @@ function printHelp() {
       "说明:",
       "  独立 Codex 本地重试网关。",
       "  非流式命中 reasoning_tokens=516 时返回 502。",
-      "  流式命中时默认直接断开连接，交给 Codex 自身重试。",
+      "  流式命中时默认缓存并返回 502，避免半截流返回。",
       "",
     ].join("\n"),
   );
@@ -153,6 +153,16 @@ function buildBlockedBody(pathname, reasoning, statusCode) {
   });
 }
 
+function buildGatewayErrorBody(message) {
+  return JSON.stringify({
+    error: {
+      message,
+      type: "codex_retry_gateway_error",
+      code: "gateway_error",
+    },
+  });
+}
+
 function createMonitor() {
   return {
     started_at: new Date().toISOString(),
@@ -1198,6 +1208,42 @@ function reasoningMatched(config, reasoning) {
   return reasoning !== null && config.reasoning_equals.includes(reasoning);
 }
 
+function isExpectedStreamTermination(error) {
+  if (!error) {
+    return false;
+  }
+  if (error.name === "AbortError") {
+    return true;
+  }
+  return error instanceof TypeError && error.message === "terminated";
+}
+
+function isRetryableUpstreamFetchError(error) {
+  if (!error) {
+    return false;
+  }
+  return error instanceof TypeError && error.message === "fetch failed";
+}
+
+async function fetchUpstreamWithRetry(upstreamUrl, init, logger) {
+  const maxAttempts = 2;
+  let lastError = null;
+
+  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
+    try {
+      return await fetch(upstreamUrl, init);
+    } catch (error) {
+      lastError = error;
+      if (!isRetryableUpstreamFetchError(error) || attempt === maxAttempts) {
+        break;
+      }
+      logger?.(`[retry] upstream fetch failed attempt=${attempt} url=${upstreamUrl}`);
+    }
+  }
+
+  throw lastError;
+}
+
 function inspectSseChunk(state, chunk) {
   const decoded = state.decoder.decode(chunk, { stream: true });
   state.buffer += decoded;
@@ -1280,9 +1326,7 @@ async function handleStreaming({
   res,
   abortController,
 }) {
-  copyHeadersToClient(upstreamResponse.headers, res);
-  res.writeHead(upstreamResponse.status);
-
+  const strict502Mode = config.stream_action !== "disconnect";
   const reader = upstreamResponse.body.getReader();
   const sseState = {
     decoder: new TextDecoder("utf8"),
@@ -1291,14 +1335,46 @@ async function handleStreaming({
 
   let wroteAnyChunk = false;
   let observedReasoning = null;
+  const bufferedChunks = [];
+
+  if (!strict502Mode) {
+    copyHeadersToClient(upstreamResponse.headers, res);
+    res.writeHead(upstreamResponse.status);
+  }
+
   while (true) {
-    const { done, value } = await reader.read();
+    let readResult;
+    try {
+      readResult = await reader.read();
+    } catch (error) {
+      if (isExpectedStreamTermination(error)) {
+        recordInspectedResponse(monitor, observedReasoning, false);
+        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"));
+        } else {
+          res.end();
+        }
+        return;
+      }
+      throw error;
+    }
+
+    const { done, value } = readResult;
     if (done) {
       recordInspectedResponse(monitor, observedReasoning, false);
+      if (strict502Mode) {
+        copyHeadersToClient(upstreamResponse.headers, res);
+        res.writeHead(upstreamResponse.status);
+        res.end(Buffer.concat(bufferedChunks));
+      } else {
         res.end();
+      }
       return;
     }
 
+    const chunkBuffer = Buffer.from(value);
     const reasoning = inspectSseChunk(sseState, value);
     if (Number.isInteger(reasoning)) {
       observedReasoning = reasoning;
@@ -1311,14 +1387,14 @@ async function handleStreaming({
         );
       }
 
-      if (!wroteAnyChunk) {
+      if (strict502Mode || !wroteAnyChunk) {
+        abortController.abort();
+        reader.cancel().catch(() => {});
         const blockedBody = buildBlockedBody(pathname, reasoning, config.non_stream_status_code);
-        if (!res.headersSent) {
         res.writeHead(config.non_stream_status_code, {
           "content-type": "application/json; charset=utf-8",
           "x-codex-retry-gateway-reason": "reasoning-guard-triggered",
         });
-        }
         res.end(blockedBody);
       } else {
         abortController.abort();
@@ -1328,8 +1404,12 @@ async function handleStreaming({
       return;
     }
 
+    if (strict502Mode) {
+      bufferedChunks.push(chunkBuffer);
+    } else {
       wroteAnyChunk = true;
-    res.write(Buffer.from(value));
+      res.write(chunkBuffer);
+    }
   }
 }
 
@@ -1367,12 +1447,12 @@ async function proxyRequest(runtime, req, res) {
   const upstreamUrl = buildUpstreamUrl(config.upstream_base_url, incomingUrl);
   const abortController = new AbortController();
 
-  const upstreamResponse = await fetch(upstreamUrl, {
+  const upstreamResponse = await fetchUpstreamWithRetry(upstreamUrl, {
     method: req.method,
     headers: cloneHeadersForUpstream(req.headers),
     body: requestBody.length > 0 ? requestBody : undefined,
     signal: abortController.signal,
-  });
+  }, logger);
 
   const shouldInspect = matchPath(config, pathname);
   const responseIsStream =

scripts/admin-lib.mjs

@@ -418,7 +418,7 @@ export async function installForCurrentProvider({
       existingGatewayConfig?.non_stream_status_code === undefined || existingGatewayConfig?.non_stream_status_code === null
         ? 502
         : Number.parseInt(`${existingGatewayConfig.non_stream_status_code}`, 10),
-    stream_action: existingGatewayConfig?.stream_action || "disconnect",
+    stream_action: existingGatewayConfig?.stream_action || "strict_502",
     log_match: existingGatewayConfig?.log_match === undefined ? true : Boolean(existingGatewayConfig.log_match),
     health_path: existingGatewayConfig?.health_path || DEFAULT_HEALTH_PATH,
   };

scripts/install-for-current-provider.ps1

@@ -62,7 +62,7 @@ $gatewayConfig = [ordered]@{
   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 }
-  stream_action = if ($existingGatewayConfig -and -not [string]::IsNullOrWhiteSpace([string]$existingGatewayConfig.stream_action)) { [string]$existingGatewayConfig.stream_action } else { "disconnect" }
+  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" }
 }

scripts/test-gateway-e2e.mjs

@@ -4,7 +4,7 @@ import http from "node:http";
 import net from "node:net";
 import { once } from "node:events";
 import { spawn } from "node:child_process";
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 
@@ -63,7 +63,25 @@ function createSseResponse(res, chunks) {
   });
 }
 
+function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) {
+  res.writeHead(200, {
+    "content-type": "text/event-stream; charset=utf-8",
+    "cache-control": "no-cache",
+    connection: "keep-alive",
+    "x-upstream-test": "sse-terminated",
+  });
+
+  for (const chunk of chunks) {
+    res.write(chunk);
+  }
+
+  setTimeout(() => {
+    res.socket?.destroy();
+  }, destroyDelayMs);
+}
+
 function startFakeUpstream(port) {
+  const failBeforeResponseCounts = new Map();
   const server = http.createServer((req, res) => {
     const responsePaths = new Set(["/responses", "/v1/responses"]);
     const chatCompletionPaths = new Set(["/chat/completions", "/v1/chat/completions"]);
@@ -90,11 +108,37 @@ function startFakeUpstream(port) {
       req.on("end", () => {
         const parsed = JSON.parse(body || "{}");
         const reasoning = parsed.test_reasoning_tokens ?? 128;
+        if (parsed.test_fail_before_response_once) {
+          const failKey = `${req.url}:fail-before-response-once`;
+          const failCount = (failBeforeResponseCounts.get(failKey) || 0) + 1;
+          failBeforeResponseCounts.set(failKey, failCount);
+          if (failCount === 1) {
+            res.socket?.destroy();
+            return;
+          }
+        }
+        if (parsed.test_force_terminate) {
+          createTerminatedSseResponse(res, [
+            'data: {"type":"response.output_text.delta","delta":"hello"}\n\n',
+          ]);
+          return;
+        }
+        if (parsed.stream) {
+          createSseResponse(res, [
+            'data: {"type":"response.output_text.delta","delta":"hello"}\n\n',
+            `data: {"response":{"usage":{"output_tokens_details":{"reasoning_tokens":${reasoning}}}}}\n\n`,
+            "data: [DONE]\n\n",
+          ]);
+          return;
+        }
         createJsonResponse(
           res,
           200,
           {
             id: "resp_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,
@@ -232,7 +276,7 @@ async function run() {
     endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"],
     reasoning_equals: [516],
     non_stream_status_code: 502,
-    stream_action: "disconnect",
+    stream_action: "strict_502",
     log_match: true,
     health_path: "/__codex_retry_gateway/health",
   };
@@ -279,17 +323,32 @@ async function run() {
       );
     }
 
-    for (const streamPath of ["/chat/completions", "/v1/chat/completions"]) {
+    const recoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ test_fail_before_response_once: true }),
+    });
+    const recoveredBody = await recoveredResponse.json();
+    assert(recoveredResponse.status === 200, `首次 fetch failed 后未自动恢复: ${recoveredResponse.status}`);
+    assert(recoveredBody?.retry_attempt === 2, "首次 fetch failed 后未命中第二次上游请求");
+
+    for (const streamPath of [
+      "/responses",
+      "/v1/responses",
+      "/chat/completions",
+      "/v1/chat/completions",
+    ]) {
       const blockedStream = await readSseUntilClose(
         `http://127.0.0.1:${gatewayPort}${streamPath}`,
         { stream: true, test_reasoning_tokens: 516 },
       );
-      assert(blockedStream.status === 200, `${streamPath} 516 首状态异常: ${blockedStream.status}`);
-      assert(blockedStream.text.includes('"content":"hello"'), `${streamPath} 流式 516 未先透传正常 chunk`);
-      assert(!blockedStream.text.includes("[DONE]"), `${streamPath} 流式 516 不应完整结束`);
+      assert(blockedStream.status === 502, `${streamPath} 516 未返回 502: ${blockedStream.status}`);
+      assert(!blockedStream.text.includes("hello"), `${streamPath} 严格 502 模式不应先透传正常 chunk`);
+      assert(!blockedStream.text.includes("[DONE]"), `${streamPath} 严格 502 模式不应回放 DONE`);
+      const blockedStreamBody = JSON.parse(blockedStream.text);
       assert(
-        blockedStream.closedByError || blockedStream.text.includes("[[reader-error:"),
-        `${streamPath} 流式 516 未表现为中途断开`,
+        blockedStreamBody?.error?.code === "reasoning_guard_triggered",
+        `${streamPath} 流式 516 返回体不正确`,
       );
 
       const okStream = await readSseUntilClose(
@@ -301,6 +360,19 @@ async function run() {
       assert(!okStream.closedByError, `${streamPath} 流式 128 不应异常断开`);
     }
 
+    const terminatedStream = await readSseUntilClose(
+      `http://127.0.0.1:${gatewayPort}/responses`,
+      { stream: true, test_force_terminate: true },
+    );
+    assert(terminatedStream.status === 502, `/responses 上游半路断流未返回 502: ${terminatedStream.status}`);
+
+    await new Promise((resolve) => setTimeout(resolve, 120));
+    const logText = await readFile(logPath, "utf8");
+    assert(
+      !logText.includes("[error] TypeError: terminated"),
+      "上游半路断流后不应记录 terminated error 日志",
+    );
+
     process.stdout.write("PASS codex-retry-gateway e2e\n");
   } finally {
     gateway.child.kill();