#!/usr/bin/env node

import http from "node:http";
import net from "node:net";
import { once } from "node:events";
import { spawn } from "node:child_process";
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import os from "node:os";
import path from "node:path";

const gatewayRoot = path.resolve(import.meta.dirname, "..");
const gatewayEntry = path.join(gatewayRoot, "gateway.mjs");

function assert(condition, message) {
  if (!condition) {
    throw new Error(message);
  }
}

async function getFreePort() {
  const server = net.createServer();
  server.listen(0, "127.0.0.1");
  await once(server, "listening");
  const address = server.address();
  const port = address && typeof address === "object" ? address.port : null;
  server.close();
  await once(server, "close");
  if (!port) {
    throw new Error("无法分配空闲端口");
  }
  return port;
}

function createJsonResponse(res, statusCode, body, extraHeaders = {}) {
  res.writeHead(statusCode, {
    "content-type": "application/json; charset=utf-8",
    ...extraHeaders,
  });
  res.end(JSON.stringify(body));
}

function createSseResponse(res, chunks, intervalMs = 20) {
  res.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-cache",
    connection: "keep-alive",
    "x-upstream-test": "sse",
  });

  let index = 0;
  const timer = setInterval(() => {
    if (index >= chunks.length) {
      clearInterval(timer);
      res.end();
      return;
    }
    res.write(chunks[index]);
    index += 1;
  }, intervalMs);

  res.on("close", () => {
    clearInterval(timer);
  });
}

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 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: ${eventName}\n`,
      `data: ${JSON.stringify(payload)}\n\n`,
    ],
    intervalMs,
  );
}

function startFakeUpstream(port) {
  const failBeforeResponseCounts = new Map();
  const capacityBeforeSuccessCounts = new Map();
  const server = http.createServer((req, res) => {
    const responsePaths = new Set(["/responses", "/v1/responses"]);
    const chatCompletionPaths = new Set(["/chat/completions", "/v1/chat/completions"]);

    if (req.method === "GET" && req.url === "/v1/models") {
      createJsonResponse(
        res,
        200,
        {
          object: "list",
          data: [{ id: "fake-model" }],
        },
        { "x-upstream-test": "models-ok" },
      );
      return;
    }

    if (req.method === "POST" && responsePaths.has(req.url)) {
      let body = "";
      req.setEncoding("utf8");
      req.on("data", (chunk) => {
        body += chunk;
      });
      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.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: capacityMessage,
                type: "server_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,
            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,
                capacityMessage,
                20,
                {
                  eventName: parsed.test_capacity_stream_event_name || "error",
                  payloadShape: parsed.test_capacity_stream_payload_shape || "default",
                },
              );
              return;
            }
            createJsonResponse(
              res,
              parsed.test_capacity_status ?? 503,
              {
                error: {
                  message: capacityMessage,
                  type: "server_error",
                },
              },
              { "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` },
            );
            return;
          }
        }
        if (parsed.stream && parsed.test_capacity_error) {
          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;
        }
        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: {"response":{"usage":{"output_tokens_details":{"reasoning_tokens":${reasoning}}}}}\n\n`,
            "data: [DONE]\n\n",
          ], parsed.test_stream_chunk_delay_ms ?? 20);
          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,
              },
            },
          },
          { "x-upstream-test": `responses-${reasoning}` },
        );
      });
      return;
    }

    if (req.method === "POST" && chatCompletionPaths.has(req.url)) {
      let body = "";
      req.setEncoding("utf8");
      req.on("data", (chunk) => {
        body += chunk;
      });
      req.on("end", () => {
        const parsed = JSON.parse(body || "{}");
        const reasoning = parsed.test_reasoning_tokens ?? 128;
        if (reasoning === 516) {
          createSseResponse(res, [
            'data: {"id":"chunk-1","choices":[{"delta":{"content":"hello"}}]}\n\n',
            'data: {"usage":{"completion_tokens_details":{"reasoning_tokens":516}}}\n\n',
            "data: [DONE]\n\n",
          ]);
          return;
        }

        createSseResponse(res, [
          'data: {"id":"chunk-1","choices":[{"delta":{"content":"hello"}}]}\n\n',
          'data: {"usage":{"completion_tokens_details":{"reasoning_tokens":128}}}\n\n',
          "data: [DONE]\n\n",
        ]);
      });
      return;
    }

    createJsonResponse(res, 404, { error: "not found" });
  });

  return new Promise((resolve, reject) => {
    server.once("error", reject);
    server.listen(port, "127.0.0.1", () => resolve(server));
  });
}

async function waitForHealth(url, timeoutMs = 5000) {
  const startedAt = Date.now();
  while (Date.now() - startedAt < timeoutMs) {
    try {
      const response = await fetch(url);
      if (response.ok) {
        return;
      }
    } catch {
      // ignore startup race
    }
    await new Promise((resolve) => setTimeout(resolve, 100));
  }
  throw new Error(`等待网关健康检查超时: ${url}`);
}

function startGateway(configPath, logPath) {
  const child = spawn(process.execPath, [gatewayEntry, "--config", configPath, "--log", logPath], {
    cwd: gatewayRoot,
    stdio: ["ignore", "pipe", "pipe"],
  });

  let stdout = "";
  let stderr = "";
  child.stdout.on("data", (chunk) => {
    stdout += chunk.toString();
  });
  child.stderr.on("data", (chunk) => {
    stderr += chunk.toString();
  });

  return {
    child,
    getOutput() {
      return { stdout, stderr };
    },
  };
}

async function readSseUntilClose(url, requestBody) {
  const response = await fetch(url, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify(requestBody),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder("utf8");
  let text = "";
  let closedByError = false;

  while (true) {
    try {
      const { done, value } = await reader.read();
      if (done) {
        break;
      }
      text += decoder.decode(value, { stream: true });
    } catch (error) {
      closedByError = true;
      text += `\n[[reader-error:${error?.name || "unknown"}]]`;
      break;
    }
  }

  text += decoder.decode();
  return {
    status: response.status,
    headers: response.headers,
    text,
    closedByError,
  };
}

async function run() {
  const tempRoot = await mkdtemp(path.join(os.tmpdir(), "codex-retry-gateway-"));
  const upstreamPort = await getFreePort();
  const gatewayPort = await getFreePort();
  const configPath = path.join(tempRoot, "config.json");
  const logPath = path.join(tempRoot, "gateway.log");

  const config = {
    listen_host: "127.0.0.1",
    listen_port: gatewayPort,
    upstream_base_url: `http://127.0.0.1:${upstreamPort}`,
    request_body_limit_bytes: 10 * 1024 * 1024,
    endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"],
    reasoning_equals: [516],
    retryable_status_codes: [429, 503],
    retryable_error_messages: [
      "Selected model is at capacity. Please try a different model.",
      "stream disconnected before completion: Concurrency limit exceeded for account, please retry later",
    ],
    upstream_fetch_retry_attempts: 5,
    upstream_fetch_retry_backoff_ms: 25,
    non_stream_status_code: 502,
    stream_action: "strict_502",
    log_match: true,
    health_path: "/__codex_retry_gateway/health",
  };

  await writeFile(configPath, JSON.stringify(config, null, 2), "utf8");

  const upstream = await startFakeUpstream(upstreamPort);
  let gateway = startGateway(configPath, logPath);

  try {
    try {
      await waitForHealth(`http://127.0.0.1:${gatewayPort}${config.health_path}`);
    } catch (error) {
      const output = gateway.getOutput();
      throw new Error(
        `${error?.message || error}\nstdout:\n${output.stdout || "(empty)"}\nstderr:\n${output.stderr || "(empty)"}`,
      );
    }

    const modelsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/v1/models`);
    assert(modelsResponse.status === 200, `/v1/models 透传状态异常: ${modelsResponse.status}`);
    assert(
      modelsResponse.headers.get("x-upstream-test") === "models-ok",
      "/v1/models 未保留上游头",
    );

    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 }),
      });
      const blockedBody = await blockedResponse.json();
      assert(blockedResponse.status === 502, `${responsePath} 516 未返回 502: ${blockedResponse.status}`);
      assert(
        blockedBody?.error?.code === "reasoning_guard_triggered",
        `${responsePath} 516 返回体不正确`,
      );

      const okResponse = await fetch(`http://127.0.0.1:${gatewayPort}${responsePath}`, {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify({ test_reasoning_tokens: 128 }),
      });
      const okBody = await okResponse.json();
      assert(okResponse.status === 200, `${responsePath} 128 透传状态异常: ${okResponse.status}`);
      assert(okResponse.headers.get("x-upstream-test") === "responses-128", `${responsePath} 128 未保留头`);
      assert(
        okBody?.usage?.output_tokens_details?.reasoning_tokens === 128,
        `${responsePath} 128 返回体异常`,
      );
    }

    const recoveredPayload = JSON.stringify({ test_fail_before_response_once: true });
    const recoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: recoveredPayload,
    });
    const recoveredBody = await recoveredResponse.json();
    assert(recoveredResponse.status === 200, `首次 fetch failed 后未自动恢复: ${recoveredResponse.status}`);
    assert(recoveredBody?.retry_attempt === 2, "首次 fetch failed 后未命中第二次上游请求");

    const requestsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=20`);
    const requestsPayload = await requestsResponse.json();
    const recoveredEntry = requestsPayload?.entries?.find((entry) => entry.path === "/responses" && entry.status_code === 200);
    assert(requestsResponse.status === 200, `请求历史 API 状态异常: ${requestsResponse.status}`);
    assert(
      recoveredEntry?.request_body_bytes === Buffer.byteLength(recoveredPayload),
      `请求体大小记录异常: ${recoveredEntry?.request_body_bytes}`,
    );
    assert(recoveredEntry?.request_id, "请求记录未生成 request_id");

    const threadTrackedResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ test_reasoning_tokens: 128, thread_id: "thread_nonstream" }),
    });
    const threadTrackedBody = await threadTrackedResponse.json();
    assert(threadTrackedResponse.status === 200, `thread non-stream 请求失败: ${threadTrackedResponse.status}`);
    assert(threadTrackedBody?.id === "resp_test", "thread non-stream 返回体缺少 response id");
    assert(threadTrackedBody?.thread_id === "thread_nonstream", "thread non-stream 返回体缺少 thread_id");

    const threadRequestsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent("thread_nonstream")}`);
    const threadRequestsPayload = await threadRequestsResponse.json();
    const threadEntry = (threadRequestsPayload?.entries || []).find((entry) => entry.thread_id === "thread_nonstream");
    assert(threadEntry?.response_id === "resp_test", "non-stream 请求记录未保留 response_id");
    assert(threadEntry?.thread_id === "thread_nonstream", "non-stream 请求记录未保留 thread_id");

    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",
      headers: { "content-type": "application/json" },
      body: sameRequestPayload,
    });
    assert(sameRequestFirstResponse.status === 200, `相同请求首次发送失败: ${sameRequestFirstResponse.status}`);
    await sameRequestFirstResponse.json();

    const sameRequestSecondResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: sameRequestPayload,
    });
    assert(sameRequestSecondResponse.status === 200, `相同请求二次发送失败: ${sameRequestSecondResponse.status}`);
    await sameRequestSecondResponse.json();

    const differentBodyPayload = JSON.stringify({ test_reasoning_tokens: 128, test_request_id_marker: "different" });
    const differentBodyResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: differentBodyPayload,
    });
    assert(differentBodyResponse.status === 200, `不同请求体发送失败: ${differentBodyResponse.status}`);
    await differentBodyResponse.json();

    const differentPathResponse = await fetch(`http://127.0.0.1:${gatewayPort}/v1/responses`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: sameRequestPayload,
    });
    assert(differentPathResponse.status === 200, `不同路径发送失败: ${differentPathResponse.status}`);
    await differentPathResponse.json();

    const requestIdRequestsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=40`);
    const requestIdRequestsPayload = await requestIdRequestsResponse.json();
    const sameRequestEntries = (requestIdRequestsPayload?.entries || []).filter(
      (entry) => entry.path === "/responses" && entry.request_body_bytes === Buffer.byteLength(sameRequestPayload),
    );
    assert(sameRequestEntries.length >= 2, "未找到两条相同请求记录");
    const [sameRequestEntryA, sameRequestEntryB] = sameRequestEntries;
    assert(sameRequestEntryA.request_id, "相同请求记录缺少 request_id");
    assert(sameRequestEntryA.request_id === sameRequestEntryB.request_id, "相同请求未复用 request_id");
    assert(sameRequestEntryA.seq !== sameRequestEntryB.seq, "相同请求不应复用 seq");

    const differentBodyEntry = (requestIdRequestsPayload?.entries || []).find(
      (entry) =>
        entry.path === "/responses" &&
        entry.request_body_bytes !== Buffer.byteLength(sameRequestPayload) &&
        entry.request_body_bytes === Buffer.byteLength(differentBodyPayload),
    );
    assert(differentBodyEntry?.request_id, "不同请求体记录缺少 request_id");
    assert(differentBodyEntry.request_id !== sameRequestEntryA.request_id, "不同请求体错误复用 request_id");

    const differentPathEntry = (requestIdRequestsPayload?.entries || []).find(
      (entry) => entry.path === "/v1/responses" && entry.request_body_bytes === Buffer.byteLength(sameRequestPayload),
    );
    assert(differentPathEntry?.request_id, "不同路径记录缺少 request_id");
    assert(differentPathEntry.request_id !== sameRequestEntryA.request_id, "不同路径错误复用 request_id");

    const requestIdQueryResponse = await fetch(
      `http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(sameRequestEntryA.request_id)}`,
    );
    const requestIdQueryPayload = await requestIdQueryResponse.json();
    assert(requestIdQueryResponse.status === 200, `request_id 搜索失败: ${requestIdQueryResponse.status}`);
    assert(
      (requestIdQueryPayload?.entries || []).some((entry) => entry.request_id === sameRequestEntryA.request_id),
      "request_id 搜索未命中对应记录",
    );

    const capacityResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ test_capacity_error: true }),
    });
    const capacityBody = await capacityResponse.json();
    assert(capacityResponse.status === 502, `capacity error 未返回 502: ${capacityResponse.status}`);
    assert(
      capacityBody?.error?.code === "upstream_error_retry_triggered",
      "capacity error 返回体未标记 retry trigger",
    );
    assert(
      capacityBody?.error?.upstream_status_code === 503,
      "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" },
      body: JSON.stringify({ test_capacity_before_success_times: 2, test_reasoning_tokens: 128 }),
    });
    const capacityRecoveredBody = await capacityRecoveredResponse.json();
    assert(capacityRecoveredResponse.status === 200, `capacity 抖动后未自动恢复: ${capacityRecoveredResponse.status}`);
    assert(
      capacityRecoveredBody?.usage?.output_tokens_details?.reasoning_tokens === 128,
      "capacity 抖动恢复后的返回体异常",
    );

    const requestsAfterCapacityRecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=20`);
    const requestsAfterCapacityRecovery = await requestsAfterCapacityRecoveryResponse.json();
    const capacityRecoveredEntry = requestsAfterCapacityRecovery?.entries?.find(
      (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.upstream_attempt_count >= 3,
    );
    assert(capacityRecoveredEntry, "capacity 抖动恢复后的请求记录未保留重试次数");
    assert(capacityRecoveredEntry.request_id, "capacity 恢复后的请求记录缺少 request_id");

    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" },
      body: JSON.stringify({ stream: true, test_capacity_error: true }),
    });
    const streamCapacityBody = await streamCapacityResponse.json();
    assert(streamCapacityResponse.status === 502, `stream+capacity error 未返回 502: ${streamCapacityResponse.status}`);
    assert(
      streamCapacityBody?.error?.code === "upstream_error_retry_triggered",
      "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" },
      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 抖动恢复后的请求记录未保留重试次数");

    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 恢复后的请求记录未保留重试次数");

    const streamThreadResponse = await readSseUntilClose(
      `http://127.0.0.1:${gatewayPort}/responses`,
      { stream: true, test_reasoning_tokens: 128, thread_id: "thread_stream_ok" },
    );
    assert(streamThreadResponse.status === 200, `stream thread 请求失败: ${streamThreadResponse.status}`);
    const streamThreadRequestsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent("thread_stream_ok")}`);
    const streamThreadRequestsPayload = await streamThreadRequestsResponse.json();
    const streamThreadEntry = (streamThreadRequestsPayload?.entries || []).find((entry) => entry.thread_id === "thread_stream_ok");
    assert(streamThreadEntry?.response_id === "resp_stream", "stream 请求记录未保留 response_id");
    assert(streamThreadEntry?.thread_id === "thread_stream_ok", "stream 请求记录未保留 thread_id");

    const metadataThreadResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
      method: "POST",
      headers: {
        "content-type": "application/json",
        "thread-id": "thread_header_fallback",
        "x-client-request-id": "thread_header_request_id",
      },
      body: JSON.stringify({
        test_reasoning_tokens: 128,
        client_metadata: {
          thread_id: "thread_client_metadata",
          "x-codex-thread-id": "thread_client_metadata_alias",
        },
      }),
    });
    assert(metadataThreadResponse.status === 200, `metadata thread 请求失败: ${metadataThreadResponse.status}`);
    await metadataThreadResponse.json();
    const metadataThreadRequestsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent("thread_client_metadata")}`);
    const metadataThreadRequestsPayload = await metadataThreadRequestsResponse.json();
    const metadataThreadEntry = (metadataThreadRequestsPayload?.entries || []).find((entry) => entry.thread_id === "thread_client_metadata");
    assert(metadataThreadEntry?.thread_id === "thread_client_metadata", "client_metadata.thread_id 未写入请求记录");

    const streamDisconnectedRetryResponse = 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_capacity_message: "stream disconnected before completion: Concurrency limit exceeded for account, please retry later",
        test_reasoning_tokens: 128,
      }),
    });
    const streamDisconnectedRetryText = await streamDisconnectedRetryResponse.text();
    assert(streamDisconnectedRetryResponse.status === 200, `stream disconnected capacity 抖动后未自动恢复: ${streamDisconnectedRetryResponse.status}`);
    assert(streamDisconnectedRetryText.includes("hello"), "stream disconnected capacity 恢复后未拿到正常 SSE 内容");

    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 === 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(
        blockedStreamBody?.error?.code === "reasoning_guard_triggered",
        `${streamPath} 流式 516 返回体不正确`,
      );

      const okStream = await readSseUntilClose(
        `http://127.0.0.1:${gatewayPort}${streamPath}`,
        { stream: true, test_reasoning_tokens: 128 },
      );
      assert(okStream.status === 200, `${streamPath} 128 首状态异常: ${okStream.status}`);
      assert(okStream.text.includes("[DONE]"), `${streamPath} 流式 128 未完整结束`);
      assert(!okStream.closedByError, `${streamPath} 流式 128 不应异常断开`);
    }

    const streamProgressPromise = fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ stream: true, test_reasoning_tokens: 128, test_stream_chunk_delay_ms: 180 }),
    });
    await new Promise((resolve) => setTimeout(resolve, 260));
    const midRequestsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=20`);
    const midRequestsPayload = await midRequestsResponse.json();
    const inFlightStreamEntry = midRequestsPayload?.entries?.find(
      (entry) =>
        entry.path === "/responses" &&
        entry.response_stream === true &&
        (entry.lifecycle_state === "streaming" || entry.lifecycle_state === "receive_first") &&
        (entry.stream_chunk_count || 0) >= 1,
    );
    assert(inFlightStreamEntry, "流式请求过程中未暴露进行中状态");
    assert(
      (inFlightStreamEntry?.response_bytes_received || 0) > 0,
      "流式请求过程中未累计接收字节数",
    );
    const streamProgressResponse = await streamProgressPromise;
    assert(streamProgressResponse.status === 200, `stream progress 响应状态异常: ${streamProgressResponse.status}`);
    const streamReader = streamProgressResponse.body.getReader();
    while (true) {
      const { done } = await streamReader.read();
      if (done) {
        break;
      }
    }

    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}`);

    const metricsBeforeRestartResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/status`);
    const metricsBeforeRestart = await metricsBeforeRestartResponse.json();
    assert(metricsBeforeRestartResponse.status === 200, `status API 状态异常: ${metricsBeforeRestartResponse.status}`);
    assert(metricsBeforeRestart?.metrics?.reasoning_516_count >= 1, "重启前 reasoning_516_count 未累计");
    assert(metricsBeforeRestart?.metrics?.observed_reasoning_counts?.["128"] >= 1, "重启前 reasoning 128 未累计");
    assert(metricsBeforeRestart?.metrics?.total_proxy_request_count >= 1, "重启前 total_proxy_request_count 未累计");

    gateway.child.kill();
    await once(gateway.child, "exit");
    gateway = startGateway(configPath, logPath);
    await waitForHealth(`http://127.0.0.1:${gatewayPort}${config.health_path}`);

    const metricsAfterRestartResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/status`);
    const metricsAfterRestart = await metricsAfterRestartResponse.json();
    assert(metricsAfterRestartResponse.status === 200, `重启后 status API 状态异常: ${metricsAfterRestartResponse.status}`);
    assert(metricsAfterRestart?.metrics?.reasoning_516_count >= metricsBeforeRestart?.metrics?.reasoning_516_count, "重启后 reasoning_516_count 未保留");
    assert(metricsAfterRestart?.metrics?.observed_reasoning_counts?.["128"] >= metricsBeforeRestart?.metrics?.observed_reasoning_counts?.["128"], "重启后 reasoning 128 计数未保留");
    assert(metricsAfterRestart?.metrics?.total_proxy_request_count >= metricsBeforeRestart?.metrics?.total_proxy_request_count, "重启后 total_proxy_request_count 未保留");
    assert(metricsAfterRestart?.metrics?.persistent_since, "重启后未返回 persistent_since");

    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();
    upstream.close();
    await once(upstream, "close");
    await rm(tempRoot, { recursive: true, force: true });
  }
}

run().catch((error) => {
  process.stderr.write(`${error?.stack || error}\n`);
  process.exit(1);
});