shujakuin/codex-retry-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add retry wave visibility controls

Browse Source
This commit is contained in:
yunyaozhou
2026-07-07 20:28:24 +08:00
parent 446a0e99a8
commit a62f2cf1b9
11 changed files with 1760 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/test-gateway-e2e.mjs
+289 -39
View File
@@ -43,13 +43,14 @@ function createJsonResponse(res, statusCode, body, extraHeaders = {}) {
res.end(JSON.stringify(body));
}
function createSseResponse(res, chunks, intervalMs = 20) {
function createSseResponse(res, chunks, intervalMs = 20, options = {}) {
res.writeHead(200, {
"content-type": "text/event-stream; charset=utf-8",
"cache-control": "no-cache",
connection: "keep-alive",
"x-upstream-test": "sse",
});
"x-upstream-test": "sse",
...(options.headers || {}),
});
let index = 0;
const timer = setInterval(() => {
@@ -67,12 +68,13 @@ function createSseResponse(res, chunks, intervalMs = 20) {
});
}
function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) {
function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20, options = {}) {
res.writeHead(200, {
"content-type": "text/event-stream; charset=utf-8",
"cache-control": "no-cache",
connection: "keep-alive",
"x-upstream-test": "sse-terminated",
...(options.headers || {}),
});
for (const chunk of chunks) {
@@ -120,12 +122,57 @@ function createCapacityErrorSseResponse(
`data: ${JSON.stringify(payload)}\n\n`,
],
intervalMs,
{ headers: options.headers || {} },
);
}
function reasoningRetryKeyForRequest(url, parsed) {
return [
url,
parsed.stream ? "stream" : "non-stream",
parsed.test_reasoning_retry_key || parsed.thread_id || "missing-thread",
].join(":");
}
function beginReasoningRetryTrackedRequest(statsMap, key, res) {
const stats = statsMap.get(key) || {
totalRequests: 0,
activeRequests: 0,
maxConcurrent: 0,
cancelledRequests: 0,
};
stats.totalRequests += 1;
stats.activeRequests += 1;
stats.maxConcurrent = Math.max(stats.maxConcurrent, stats.activeRequests);
statsMap.set(key, stats);
let finished = false;
const finish = (cancelled = false) => {
if (finished) {
return;
}
finished = true;
stats.activeRequests = Math.max(0, stats.activeRequests - 1);
if (cancelled) {
stats.cancelledRequests += 1;
}
};
res.on("close", () => {
finish(!res.writableEnded);
});
return {
stats,
finish,
};
}
function startFakeUpstream(port) {
const failBeforeResponseCounts = new Map();
const capacityBeforeSuccessCounts = new Map();
const reasoningBeforeSuccessCounts = new Map();
const reasoningRetryStats = new Map();
const server = http.createServer((req, res) => {
const responsePaths = new Set(["/responses", "/v1/responses"]);
const chatCompletionPaths = new Set(["/chat/completions", "/v1/chat/completions"]);
@@ -148,10 +195,25 @@ function startFakeUpstream(port) {
req.setEncoding("utf8");
req.on("data", (chunk) => {
body += chunk;
});
});
req.on("end", () => {
const parsed = JSON.parse(body || "{}");
const reasoning = parsed.test_reasoning_tokens ?? 128;
let reasoning = parsed.test_reasoning_tokens ?? 128;
let reasoningAttempt = null;
const reasoningRetryKey = Number.isInteger(parsed.test_reasoning_before_success_times)
? reasoningRetryKeyForRequest(req.url, parsed)
: null;
const reasoningRetryTracker = reasoningRetryKey
? beginReasoningRetryTrackedRequest(reasoningRetryStats, reasoningRetryKey, res)
: null;
if (Number.isInteger(parsed.test_reasoning_before_success_times) && reasoningRetryKey) {
const currentCount = (reasoningBeforeSuccessCounts.get(reasoningRetryKey) || 0) + 1;
reasoningBeforeSuccessCounts.set(reasoningRetryKey, currentCount);
reasoningAttempt = currentCount;
reasoning = currentCount <= parsed.test_reasoning_before_success_times
? 516
: (parsed.test_reasoning_success_tokens ?? 128);
}
if (parsed.test_fail_before_response_once) {
const failKey = `${req.url}:fail-before-response-once`;
const failCount = (failBeforeResponseCounts.get(failKey) || 0) + 1;
@@ -236,32 +298,50 @@ function startFakeUpstream(port) {
}
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: ${JSON.stringify({ type: "response.output_text.delta", delta: "hello", response_id: "resp_stream", thread_id: parsed.thread_id || "thread_stream", retry_attempt: reasoningAttempt })}\n\n`,
`data: {"response":{"usage":{"output_tokens_details":{"reasoning_tokens":${reasoning}}}}}\n\n`,
"data: [DONE]\n\n",
], parsed.test_stream_chunk_delay_ms ?? 20);
], parsed.test_reasoning_response_delay_ms ?? parsed.test_stream_chunk_delay_ms ?? 20, {
headers: reasoningAttempt
? { "x-upstream-reasoning-attempt": `${reasoningAttempt}` }
: {},
});
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,
const sendJsonResponse = () => {
if (res.writableEnded || res.destroyed) {
reasoningRetryTracker?.finish(true);
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
: reasoningAttempt || 0,
usage: {
output_tokens_details: {
reasoning_tokens: reasoning,
},
},
},
},
{ "x-upstream-test": `responses-${reasoning}` },
);
});
return;
}
{
"x-upstream-test": `responses-${reasoning}`,
...(reasoningAttempt ? { "x-upstream-reasoning-attempt": `${reasoningAttempt}` } : {}),
},
);
};
if (parsed.test_reasoning_response_delay_ms) {
setTimeout(sendJsonResponse, parsed.test_reasoning_response_delay_ms);
return;
}
sendJsonResponse();
});
return;
}
if (req.method === "POST" && chatCompletionPaths.has(req.url)) {
let body = "";
@@ -293,11 +373,21 @@ function startFakeUpstream(port) {
createJsonResponse(res, 404, { error: "not found" });
});
return new Promise((resolve, reject) => {
server.once("error", reject);
server.listen(port, "127.0.0.1", () => resolve(server));
});
}
return new Promise((resolve, reject) => {
server.once("error", reject);
server.listen(port, "127.0.0.1", () => {
server.getReasoningRetryStat = (key) => {
return reasoningRetryStats.get(key) || {
totalRequests: 0,
activeRequests: 0,
maxConcurrent: 0,
cancelledRequests: 0,
};
};
resolve(server);
});
});
}
async function waitForHealth(url, timeoutMs = 5000) {
const startedAt = Date.now();
@@ -386,7 +476,8 @@ async function run() {
upstream_base_url: `http://127.0.0.1:${upstreamPort}`,
request_body_limit_bytes: 1024 * 1024 * 1024,
endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"],
reasoning_equals: [516],
reasoning_match_mode: "formula_518n_minus_2",
reasoning_equals: [516, 1034, 1552],
retryable_status_codes: [429, 503],
retryable_error_messages: [
"Selected model is at capacity. Please try a different model.",
@@ -450,9 +541,9 @@ async function run() {
"/v1/models 未保留上游头",
);
for (const responsePath of ["/responses", "/v1/responses"]) {
const blockedResponse = await fetch(`http://127.0.0.1:${gatewayPort}${responsePath}`, {
method: "POST",
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 }),
});
@@ -473,10 +564,145 @@ async function run() {
assert(okResponse.headers.get("x-upstream-test") === "responses-128", `${responsePath} 128 未保留头`);
assert(
okBody?.usage?.output_tokens_details?.reasoning_tokens === 128,
`${responsePath} 128 返回体异常`,
);
}
`${responsePath} 128 返回体异常`,
);
}
const blockedFormulaResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ test_reasoning_tokens: 2070 }),
});
const blockedFormulaBody = await blockedFormulaResponse.json();
assert(blockedFormulaResponse.status === 502, `/responses 2070 未按 518n-2 返回 502: ${blockedFormulaResponse.status}`);
assert(
blockedFormulaBody?.error?.code === "reasoning_guard_triggered",
"/responses 2070 返回体不正确",
);
const missingThreadRetryKey = reasoningRetryKeyForRequest("/responses", {
stream: false,
test_reasoning_retry_key: "missing-thread-fallback",
});
const missingThreadRetryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
test_reasoning_before_success_times: 1,
test_reasoning_retry_key: "missing-thread-fallback",
}),
});
const missingThreadRetryBody = await missingThreadRetryResponse.json();
assert(missingThreadRetryResponse.status === 502, `无 thread_id 的 responses 请求不应自动重打: ${missingThreadRetryResponse.status}`);
assert(
missingThreadRetryBody?.error?.code === "reasoning_guard_triggered",
"无 thread_id 的 responses 请求返回体异常",
);
const missingThreadRetryStats = upstream.getReasoningRetryStat(missingThreadRetryKey);
assert(missingThreadRetryStats.totalRequests === 1, "无 thread_id 的 responses 请求不应启动多轮重打");
const retryRound2ThreadId = "thread_retry_round2";
const retryRound2Response = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
thread_id: retryRound2ThreadId,
test_reasoning_before_success_times: 1,
test_reasoning_retry_key: "round2",
}),
});
const retryRound2Body = await retryRound2Response.json();
assert(retryRound2Response.status === 200, `1,1 重打未恢复: ${retryRound2Response.status}`);
assert(retryRound2Body?.usage?.output_tokens_details?.reasoning_tokens === 128, "1,1 重打恢复后的 reasoning_tokens 异常");
assert(retryRound2Response.headers.get("x-upstream-reasoning-attempt") === "2", "1,1 重打未命中第二次上游请求");
const retryRound2EntryResponse = await fetch(
`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(retryRound2ThreadId)}`,
{ headers: adminHeaders },
);
const retryRound2EntryPayload = await retryRound2EntryResponse.json();
const retryRound2Entry = (retryRound2EntryPayload?.entries || []).find((entry) => entry.thread_id === retryRound2ThreadId);
assert(retryRound2Entry?.reasoning_retry_query_count === 2, "1,1 重打未记录两次 query");
assert(retryRound2Entry?.reasoning_retry_round_count === 2, "1,1 重打未记录两轮");
assert(retryRound2Entry?.reasoning_retry_current_round === 2, "1,1 重打未记录当前轮次");
assert(retryRound2Entry?.reasoning_retry_current_width === 1, "1,1 重打未记录当前轮并行数");
assert(
Array.isArray(retryRound2Entry?.reasoning_retry_current_firsts) &&
retryRound2Entry.reasoning_retry_current_firsts.length === 1 &&
Number.isInteger(retryRound2Entry.reasoning_retry_current_firsts[0]?.first_response_delay_ms) &&
retryRound2Entry.reasoning_retry_current_firsts.every((first) => first?.outcome !== "pending"),
"1,1 重打未记录当前轮 first 列表",
);
assert(retryRound2Entry?.reasoning_retry_winner_round === 2, "1,1 重打赢家轮次异常");
assert(retryRound2Entry?.reasoning_retry_winner_slot === 1, "1,1 重打赢家槽位异常");
assert(retryRound2Entry?.reasoning_retry_stop_reason === "success", "1,1 重打 stop reason 异常");
const retryWave2ThreadId = "thread_retry_wave2";
const retryWave2Key = reasoningRetryKeyForRequest("/responses", {
stream: false,
thread_id: retryWave2ThreadId,
test_reasoning_retry_key: "wave2",
});
const retryWave2Response = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
thread_id: retryWave2ThreadId,
test_reasoning_before_success_times: 3,
test_reasoning_retry_key: "wave2",
test_reasoning_response_delay_ms: 80,
}),
});
const retryWave2Body = await retryWave2Response.json();
assert(retryWave2Response.status === 200, `1,1,2 重打未恢复: ${retryWave2Response.status}`);
assert(retryWave2Body?.usage?.output_tokens_details?.reasoning_tokens === 128, "1,1,2 重打恢复后的 reasoning_tokens 异常");
assert(retryWave2Response.headers.get("x-upstream-reasoning-attempt") === "4", "1,1,2 重打未命中第四次上游请求");
const retryWave2Stats = upstream.getReasoningRetryStat(retryWave2Key);
assert(retryWave2Stats.totalRequests === 4, "1,1,2 重打总请求数异常");
assert(retryWave2Stats.maxConcurrent >= 2, "1,1,2 重打未出现第二轮并行");
const retryWave2EntryResponse = await fetch(
`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(retryWave2ThreadId)}`,
{ headers: adminHeaders },
);
const retryWave2EntryPayload = await retryWave2EntryResponse.json();
const retryWave2Entry = (retryWave2EntryPayload?.entries || []).find((entry) => entry.thread_id === retryWave2ThreadId);
assert(retryWave2Entry?.reasoning_retry_query_count === 4, "1,1,2 重打未记录四次 query");
assert(retryWave2Entry?.reasoning_retry_round_count === 3, "1,1,2 重打未记录三轮");
assert(retryWave2Entry?.reasoning_retry_current_round === 3, "1,1,2 重打未记录当前轮次");
assert(retryWave2Entry?.reasoning_retry_current_width === 2, "1,1,2 重打未记录当前轮并行数");
assert(
Array.isArray(retryWave2Entry?.reasoning_retry_current_firsts) &&
retryWave2Entry.reasoning_retry_current_firsts.length === 2 &&
retryWave2Entry.reasoning_retry_current_firsts.some((first) => Number.isInteger(first?.first_response_delay_ms)) &&
retryWave2Entry.reasoning_retry_current_firsts.every((first) => first?.outcome !== "pending"),
"1,1,2 重打未记录当前两请求 first 列表",
);
assert(retryWave2Entry?.reasoning_retry_winner_round === 3, "1,1,2 重打赢家轮次异常");
assert([1, 2].includes(retryWave2Entry?.reasoning_retry_winner_slot), "1,1,2 重打赢家槽位异常");
assert(retryWave2Entry?.reasoning_retry_stop_reason === "success", "1,1,2 重打 stop reason 异常");
const streamRetryThreadId = "thread_stream_retry_v1";
const streamRetryResponse = await readSseUntilClose(
`http://127.0.0.1:${gatewayPort}/v1/responses`,
{
stream: true,
thread_id: streamRetryThreadId,
test_reasoning_before_success_times: 1,
test_reasoning_retry_key: "stream-v1-round2",
},
);
assert(streamRetryResponse.status === 200, `/v1/responses 流式 1,1 重打未恢复: ${streamRetryResponse.status}`);
assert(streamRetryResponse.text.includes("hello"), "/v1/responses 流式 1,1 重打未拿到正常 SSE 内容");
assert(streamRetryResponse.text.includes("[DONE]"), "/v1/responses 流式 1,1 重打未完整结束");
assert(streamRetryResponse.headers.get("x-upstream-reasoning-attempt") === "2", "/v1/responses 流式 1,1 重打未命中第二次上游请求");
const streamRetryEntryResponse = await fetch(
`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(streamRetryThreadId)}`,
{ headers: adminHeaders },
);
const streamRetryEntryPayload = await streamRetryEntryResponse.json();
const streamRetryEntry = (streamRetryEntryPayload?.entries || []).find((entry) => entry.thread_id === streamRetryThreadId);
assert(streamRetryEntry?.reasoning_retry_query_count === 2, "/v1/responses 流式 1,1 重打未记录两次 query");
assert(streamRetryEntry?.response_stream === true, "/v1/responses 流式 1,1 重打未保留流式标记");
const recoveredPayload = JSON.stringify({ test_fail_before_response_once: true });
const recoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
method: "POST",
@@ -513,6 +739,30 @@ async function run() {
assert(threadEntry?.response_id === "resp_test", "non-stream 请求记录未保留 response_id");
assert(threadEntry?.thread_id === "thread_nonstream", "non-stream 请求记录未保留 thread_id");
const effortTrackedResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
test_reasoning_tokens: 128,
reasoning: {
effort: "xhigh",
summary: "auto",
},
}),
});
assert(effortTrackedResponse.status === 200, `reasoning.effort 请求失败: ${effortTrackedResponse.status}`);
await effortTrackedResponse.json();
const effortRequestsResponse = await fetch(
`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent("xhigh")}`,
{ headers: adminHeaders },
);
const effortRequestsPayload = await effortRequestsResponse.json();
const effortEntry = (effortRequestsPayload?.entries || []).find((entry) => entry.reasoning_effort === "xhigh");
assert(effortRequestsResponse.status === 200, `reasoning.effort 搜索失败: ${effortRequestsResponse.status}`);
assert(effortEntry?.reasoning_effort === "xhigh", "请求记录未保留 reasoning.effort");
assert(effortEntry?.reasoning_summary === "auto", "请求记录未保留 reasoning.summary");
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",