Add strict 502 mode and upstream retry handling for streaming responses
This commit is contained in:
nonononull
+620
-620
File diff suppressed because it is too large
Load Diff
@@ -1,109 +1,109 @@
|
||||
param(
|
||||
[string]$CodexConfigPath = "$HOME\.codex\config.toml",
|
||||
[string]$StateRoot = "$HOME\.codex-retry-gateway",
|
||||
[string]$ListenHost = "127.0.0.1",
|
||||
[int]$ListenPort = 4610
|
||||
)
|
||||
|
||||
$ErrorActionPreference = "Stop"
|
||||
Set-StrictMode -Version Latest
|
||||
|
||||
. (Join-Path $PSScriptRoot "common.ps1")
|
||||
|
||||
$paths = Get-GatewayStatePaths -StateRoot $StateRoot
|
||||
Ensure-Directory -Path $paths.StateRoot
|
||||
Ensure-Directory -Path $paths.ConfigDir
|
||||
Ensure-Directory -Path $paths.LogDir
|
||||
Ensure-Directory -Path $paths.BackupDir
|
||||
|
||||
if (-not (Test-Path -LiteralPath $CodexConfigPath)) {
|
||||
throw "Codex config file was not found: $CodexConfigPath"
|
||||
}
|
||||
|
||||
$providerContext = Get-CodexProviderContext -CodexConfigPath $CodexConfigPath
|
||||
$localGatewayBaseUrl = "http://{0}:{1}" -f $ListenHost, $ListenPort
|
||||
$existingState = Read-JsonFile -Path $paths.StatePath
|
||||
|
||||
$originalBaseUrl = $providerContext.CurrentBaseUrl
|
||||
if ($providerContext.CurrentBaseUrl -eq $localGatewayBaseUrl) {
|
||||
if ($null -eq $existingState -or [string]::IsNullOrWhiteSpace([string]$existingState.original_base_url)) {
|
||||
throw "Provider already points to the local gateway, but original_base_url is missing from state."
|
||||
}
|
||||
$originalBaseUrl = [string]$existingState.original_base_url
|
||||
}
|
||||
|
||||
if ($originalBaseUrl -eq $localGatewayBaseUrl) {
|
||||
throw "A real upstream_base_url could not be determined."
|
||||
}
|
||||
|
||||
$backupPath = Join-Path $paths.BackupDir ("config-" + (Get-Date -Format "yyyyMMdd-HHmmss") + ".toml")
|
||||
Copy-Item -LiteralPath $CodexConfigPath -Destination $backupPath -Force
|
||||
|
||||
$existingGatewayConfig = Read-JsonFile -Path $paths.ConfigPath
|
||||
$defaultEndpoints = @("/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions")
|
||||
$mergedEndpoints = @()
|
||||
foreach ($endpoint in @(
|
||||
$(if ($existingGatewayConfig) { Normalize-StringArray -Values $existingGatewayConfig.endpoints -Default @() } else { @() }) +
|
||||
$defaultEndpoints
|
||||
)) {
|
||||
if ([string]::IsNullOrWhiteSpace([string]$endpoint)) {
|
||||
continue
|
||||
}
|
||||
if ($mergedEndpoints -notcontains [string]$endpoint) {
|
||||
$mergedEndpoints += [string]$endpoint
|
||||
}
|
||||
}
|
||||
|
||||
$gatewayConfig = [ordered]@{
|
||||
listen_host = $ListenHost
|
||||
listen_port = $ListenPort
|
||||
upstream_base_url = $originalBaseUrl
|
||||
request_body_limit_bytes = if ($existingGatewayConfig -and $null -ne $existingGatewayConfig.request_body_limit_bytes) { [int]$existingGatewayConfig.request_body_limit_bytes } else { 10485760 }
|
||||
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" }
|
||||
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" }
|
||||
}
|
||||
|
||||
$previousConfigContent = Get-Content -LiteralPath $CodexConfigPath -Raw
|
||||
|
||||
try {
|
||||
Write-JsonFile -Path $paths.ConfigPath -Value $gatewayConfig
|
||||
Set-CodexProviderBaseUrl `
|
||||
-CodexConfigPath $CodexConfigPath `
|
||||
-ProviderName $providerContext.ProviderName `
|
||||
-NewBaseUrl $localGatewayBaseUrl
|
||||
|
||||
& (Join-Path $PSScriptRoot "start-gateway.ps1") `
|
||||
-StateRoot $StateRoot `
|
||||
-ConfigPath $paths.ConfigPath `
|
||||
-LogPath $paths.LogPath `
|
||||
-RestartIfRunning
|
||||
|
||||
$state = [ordered]@{
|
||||
installed_at = (Get-Date).ToString("o")
|
||||
codex_config_path = $CodexConfigPath
|
||||
provider_name = $providerContext.ProviderName
|
||||
original_base_url = $originalBaseUrl
|
||||
gateway_base_url = $localGatewayBaseUrl
|
||||
gateway_config_path = $paths.ConfigPath
|
||||
gateway_log_path = $paths.LogPath
|
||||
gateway_pid_path = $paths.PidPath
|
||||
latest_backup_path = $backupPath
|
||||
state_root = $paths.StateRoot
|
||||
}
|
||||
Write-JsonFile -Path $paths.StatePath -Value $state
|
||||
|
||||
Write-Output "Installed Codex Retry Gateway"
|
||||
Write-Output "provider=$($providerContext.ProviderName)"
|
||||
Write-Output "upstream=$originalBaseUrl"
|
||||
Write-Output "gateway=$localGatewayBaseUrl"
|
||||
Write-Output "config=$($paths.ConfigPath)"
|
||||
Write-Output "backup=$backupPath"
|
||||
} catch {
|
||||
Write-Utf8NoBomFile -Path $CodexConfigPath -Content $previousConfigContent
|
||||
& (Join-Path $PSScriptRoot "stop-gateway.ps1") -StateRoot $StateRoot -Quiet
|
||||
throw
|
||||
}
|
||||
param(
|
||||
[string]$CodexConfigPath = "$HOME\.codex\config.toml",
|
||||
[string]$StateRoot = "$HOME\.codex-retry-gateway",
|
||||
[string]$ListenHost = "127.0.0.1",
|
||||
[int]$ListenPort = 4610
|
||||
)
|
||||
|
||||
$ErrorActionPreference = "Stop"
|
||||
Set-StrictMode -Version Latest
|
||||
|
||||
. (Join-Path $PSScriptRoot "common.ps1")
|
||||
|
||||
$paths = Get-GatewayStatePaths -StateRoot $StateRoot
|
||||
Ensure-Directory -Path $paths.StateRoot
|
||||
Ensure-Directory -Path $paths.ConfigDir
|
||||
Ensure-Directory -Path $paths.LogDir
|
||||
Ensure-Directory -Path $paths.BackupDir
|
||||
|
||||
if (-not (Test-Path -LiteralPath $CodexConfigPath)) {
|
||||
throw "Codex config file was not found: $CodexConfigPath"
|
||||
}
|
||||
|
||||
$providerContext = Get-CodexProviderContext -CodexConfigPath $CodexConfigPath
|
||||
$localGatewayBaseUrl = "http://{0}:{1}" -f $ListenHost, $ListenPort
|
||||
$existingState = Read-JsonFile -Path $paths.StatePath
|
||||
|
||||
$originalBaseUrl = $providerContext.CurrentBaseUrl
|
||||
if ($providerContext.CurrentBaseUrl -eq $localGatewayBaseUrl) {
|
||||
if ($null -eq $existingState -or [string]::IsNullOrWhiteSpace([string]$existingState.original_base_url)) {
|
||||
throw "Provider already points to the local gateway, but original_base_url is missing from state."
|
||||
}
|
||||
$originalBaseUrl = [string]$existingState.original_base_url
|
||||
}
|
||||
|
||||
if ($originalBaseUrl -eq $localGatewayBaseUrl) {
|
||||
throw "A real upstream_base_url could not be determined."
|
||||
}
|
||||
|
||||
$backupPath = Join-Path $paths.BackupDir ("config-" + (Get-Date -Format "yyyyMMdd-HHmmss") + ".toml")
|
||||
Copy-Item -LiteralPath $CodexConfigPath -Destination $backupPath -Force
|
||||
|
||||
$existingGatewayConfig = Read-JsonFile -Path $paths.ConfigPath
|
||||
$defaultEndpoints = @("/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions")
|
||||
$mergedEndpoints = @()
|
||||
foreach ($endpoint in @(
|
||||
$(if ($existingGatewayConfig) { Normalize-StringArray -Values $existingGatewayConfig.endpoints -Default @() } else { @() }) +
|
||||
$defaultEndpoints
|
||||
)) {
|
||||
if ([string]::IsNullOrWhiteSpace([string]$endpoint)) {
|
||||
continue
|
||||
}
|
||||
if ($mergedEndpoints -notcontains [string]$endpoint) {
|
||||
$mergedEndpoints += [string]$endpoint
|
||||
}
|
||||
}
|
||||
|
||||
$gatewayConfig = [ordered]@{
|
||||
listen_host = $ListenHost
|
||||
listen_port = $ListenPort
|
||||
upstream_base_url = $originalBaseUrl
|
||||
request_body_limit_bytes = if ($existingGatewayConfig -and $null -ne $existingGatewayConfig.request_body_limit_bytes) { [int]$existingGatewayConfig.request_body_limit_bytes } else { 10485760 }
|
||||
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 { "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" }
|
||||
}
|
||||
|
||||
$previousConfigContent = Get-Content -LiteralPath $CodexConfigPath -Raw
|
||||
|
||||
try {
|
||||
Write-JsonFile -Path $paths.ConfigPath -Value $gatewayConfig
|
||||
Set-CodexProviderBaseUrl `
|
||||
-CodexConfigPath $CodexConfigPath `
|
||||
-ProviderName $providerContext.ProviderName `
|
||||
-NewBaseUrl $localGatewayBaseUrl
|
||||
|
||||
& (Join-Path $PSScriptRoot "start-gateway.ps1") `
|
||||
-StateRoot $StateRoot `
|
||||
-ConfigPath $paths.ConfigPath `
|
||||
-LogPath $paths.LogPath `
|
||||
-RestartIfRunning
|
||||
|
||||
$state = [ordered]@{
|
||||
installed_at = (Get-Date).ToString("o")
|
||||
codex_config_path = $CodexConfigPath
|
||||
provider_name = $providerContext.ProviderName
|
||||
original_base_url = $originalBaseUrl
|
||||
gateway_base_url = $localGatewayBaseUrl
|
||||
gateway_config_path = $paths.ConfigPath
|
||||
gateway_log_path = $paths.LogPath
|
||||
gateway_pid_path = $paths.PidPath
|
||||
latest_backup_path = $backupPath
|
||||
state_root = $paths.StateRoot
|
||||
}
|
||||
Write-JsonFile -Path $paths.StatePath -Value $state
|
||||
|
||||
Write-Output "Installed Codex Retry Gateway"
|
||||
Write-Output "provider=$($providerContext.ProviderName)"
|
||||
Write-Output "upstream=$originalBaseUrl"
|
||||
Write-Output "gateway=$localGatewayBaseUrl"
|
||||
Write-Output "config=$($paths.ConfigPath)"
|
||||
Write-Output "backup=$backupPath"
|
||||
} catch {
|
||||
Write-Utf8NoBomFile -Path $CodexConfigPath -Content $previousConfigContent
|
||||
& (Join-Path $PSScriptRoot "stop-gateway.ps1") -StateRoot $StateRoot -Quiet
|
||||
throw
|
||||
}
|
||||
|
||||
+353
-281
@@ -1,316 +1,388 @@
|
||||
#!/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, 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));
|
||||
}
|
||||
|
||||
#!/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) {
|
||||
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;
|
||||
}, 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;
|
||||
}, 20);
|
||||
|
||||
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 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"]);
|
||||
|
||||
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;
|
||||
});
|
||||
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.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,
|
||||
},
|
||||
},
|
||||
},
|
||||
{ "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");
|
||||
|
||||
},
|
||||
{ "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],
|
||||
non_stream_status_code: 502,
|
||||
stream_action: "disconnect",
|
||||
log_match: true,
|
||||
health_path: "/__codex_retry_gateway/health",
|
||||
};
|
||||
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],
|
||||
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);
|
||||
const gateway = startGateway(configPath, logPath);
|
||||
|
||||
try {
|
||||
await waitForHealth(`http://127.0.0.1:${gatewayPort}${config.health_path}`);
|
||||
|
||||
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 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 后未命中第二次上游请求");
|
||||
|
||||
await writeFile(configPath, JSON.stringify(config, null, 2), "utf8");
|
||||
|
||||
const upstream = await startFakeUpstream(upstreamPort);
|
||||
const gateway = startGateway(configPath, logPath);
|
||||
|
||||
try {
|
||||
await waitForHealth(`http://127.0.0.1:${gatewayPort}${config.health_path}`);
|
||||
|
||||
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 返回体异常`,
|
||||
);
|
||||
}
|
||||
|
||||
for (const streamPath of ["/chat/completions", "/v1/chat/completions"]) {
|
||||
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 未表现为中途断开`,
|
||||
);
|
||||
|
||||
const okStream = await readSseUntilClose(
|
||||
`http://127.0.0.1:${gatewayPort}${streamPath}`,
|
||||
{ stream: true, test_reasoning_tokens: 128 },
|
||||
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 不应异常断开`);
|
||||
}
|
||||
|
||||
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 });
|
||||
}
|
||||
}
|
||||
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}`);
|
||||
|
||||
run().catch((error) => {
|
||||
process.stderr.write(`${error?.stack || error}\n`);
|
||||
process.exit(1);
|
||||
});
|
||||
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);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user