mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: reduce Sub2API Codex upstream 502 retries

Browse Source
This commit is contained in:
Codex
2026-06-09 10:04:56 +00:00
parent 979e855caf
commit 195d33dbb5
5 changed files with 144 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/platform-infra-sub2api-codex-routing-contract-test.ts
+7
View File
@@ -16,6 +16,7 @@ const parsed = Bun.YAML.parse(readFileSync(configPath, "utf8")) as {
};
};
profiles?: { entries?: Array<{ profile?: string; accountName?: string; capacity?: number; openaiResponsesWebSocketsV2Mode?: string | null }> };
localCodex?: { responsesSmokeModel?: string };
};
const entries = parsed.profiles?.entries ?? [];
@@ -40,7 +41,11 @@ if (parsed.pool?.defaultTempUnschedulable?.enabled === true) {
assertCondition(rules.every((rule) => Number.isInteger(rule.statusCode) && (rule.statusCode ?? 0) >= 100 && (rule.statusCode ?? 0) <= 599), "temporary unschedulable rules must declare valid HTTP status codes", rules);
assertCondition(rules.every((rule) => Array.isArray(rule.keywords) && rule.keywords.length > 0), "temporary unschedulable rules must declare non-empty keywords", rules);
assertCondition(rules.every((rule) => Number.isInteger(rule.durationMinutes) && (rule.durationMinutes ?? 0) > 0), "temporary unschedulable rules must declare positive cooldown durations", rules);
const gateway502Rule = rules.find((rule) => rule.statusCode === 502);
const gateway502Keywords = new Set((gateway502Rule?.keywords ?? []).map((keyword) => keyword.toLowerCase()));
assertCondition(gateway502Keywords.has("bad gateway") && gateway502Keywords.has("upstream request failed"), "502 temporary-unschedulable rule must catch generic gateway failures", gateway502Rule);
}
assertCondition(typeof parsed.localCodex?.responsesSmokeModel === "string" && parsed.localCodex.responsesSmokeModel.length > 0, "localCodex.responsesSmokeModel must be declared for Responses smoke validation", parsed.localCodex);
console.log(JSON.stringify({
ok: true,
@@ -49,5 +54,7 @@ console.log(JSON.stringify({
"pool owner concurrency covers the YAML account capacity set",
"optional WebSocket mode overrides use supported values",
"temporary unschedulable rules are structurally valid when enabled",
"generic 502 gateway failures cool down the selected account",
"Responses smoke model is YAML-declared",
],
}));
scripts/src/platform-infra-sub2api-codex.ts
+130 -5
View File
@@ -128,6 +128,7 @@ interface CodexPoolLocalCodexConfig {
wireApi: string;
supportsWebSockets: boolean;
responsesWebSocketsV2: boolean;
responsesSmokeModel: string;
}
interface CodexLocalConsumerTomlOptions {
@@ -408,6 +409,7 @@ async function codexPoolConfigureLocal(config: UniDeskConfig, options: ConfirmOp
wireApi: pool.localCodex.wireApi,
supportsWebSockets: pool.localCodex.supportsWebSockets,
responsesWebSocketsV2: pool.localCodex.responsesWebSocketsV2,
responsesSmokeModel: pool.localCodex.responsesSmokeModel,
valuesPrinted: false,
},
next: {
@@ -632,6 +634,7 @@ function defaultCodexPoolConfig(): CodexPoolConfig {
wireApi: "responses",
supportsWebSockets: true,
responsesWebSocketsV2: true,
responsesSmokeModel: "gpt-5.5",
},
};
}
@@ -666,7 +669,7 @@ export function defaultCodexTempUnschedulablePolicy(): CodexTempUnschedulablePol
},
{
statusCode: 502,
keywords: ["capacity", "overloaded", "temporarily unavailable", "temporary", "upstream"],
keywords: ["capacity", "overloaded", "temporarily unavailable", "temporary", "upstream", "bad gateway", "upstream request failed", "websocket dial", "handshake response"],
durationMinutes: 5,
description: "Gateway upstream failures should prefer another account for a short period.",
},
@@ -884,10 +887,12 @@ function readLocalCodexConfig(value: unknown, defaults: CodexPoolLocalCodexConfi
wireApi: stringValue(value.wireApi) ?? defaults.wireApi,
supportsWebSockets: readBooleanConfig(value.supportsWebSockets, "localCodex.supportsWebSockets", defaults.supportsWebSockets),
responsesWebSocketsV2: readBooleanConfig(value.responsesWebSocketsV2, "localCodex.responsesWebSocketsV2", defaults.responsesWebSocketsV2),
responsesSmokeModel: stringValue(value.responsesSmokeModel) ?? defaults.responsesSmokeModel,
};
if (!/^[A-Za-z0-9._-]+$/u.test(config.backupSuffix)) throw new Error(`${codexPoolConfigPath}.localCodex.backupSuffix has an unsupported format`);
validateProxyName(config.providerName, "localCodex.providerName");
validateProxyName(config.wireApi, "localCodex.wireApi");
validateModelName(config.responsesSmokeModel, "localCodex.responsesSmokeModel");
return config;
}
@@ -901,6 +906,10 @@ function validateProxyName(value: string, key: string): void {
if (!/^[A-Za-z0-9._-]+$/u.test(value)) throw new Error(`${codexPoolConfigPath}.${key} has an unsupported format`);
}
function validateModelName(value: string, key: string): void {
if (!/^[A-Za-z0-9._:-]+$/u.test(value)) throw new Error(`${codexPoolConfigPath}.${key} has an unsupported model name`);
}
function validatePublicHostname(value: string, key: string): void {
if (value.length > 253 || !/^[A-Za-z0-9.-]+$/u.test(value) || value.startsWith(".") || value.endsWith(".")) {
throw new Error(`${codexPoolConfigPath}.${key} has an unsupported hostname format`);
@@ -1710,6 +1719,7 @@ POOL_API_KEY_SECRET_KEY = "${pool.apiKeySecretKey}"
MIN_OWNER_BALANCE_USD = ${JSON.stringify(pool.minOwnerBalanceUsd)}
MIN_OWNER_CONCURRENCY = ${JSON.stringify(pool.minOwnerConcurrency)}
POOL_DEFAULT_ACCOUNT_CAPACITY = ${JSON.stringify(pool.defaultAccountCapacity)}
RESPONSES_SMOKE_MODEL = ${JSON.stringify(pool.localCodex.responsesSmokeModel)}
EXPECTED_ACCOUNT_CAPACITIES = ${JSON.stringify(desiredAccountCapacityMap(pool))}
EXPECTED_ACCOUNT_WS_MODES = json.loads(${JSON.stringify(JSON.stringify(desiredAccountWebSocketsV2ModeMap(pool)))})
EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE = json.loads(${JSON.stringify(JSON.stringify(desiredAccountTempUnschedulableMap(pool)))})
@@ -2165,6 +2175,117 @@ def validate_gateway(api_key):
"valuesPrinted": False,
}
def response_output_preview(parsed):
if not isinstance(parsed, dict):
return ""
if isinstance(parsed.get("output_text"), str):
return parsed["output_text"][:240]
output = parsed.get("output")
if not isinstance(output, list):
return ""
parts = []
for item in output:
if not isinstance(item, dict):
continue
content = item.get("content")
if not isinstance(content, list):
continue
for block in content:
if not isinstance(block, dict):
continue
text_value = block.get("text")
if isinstance(text_value, str) and text_value:
parts.append(text_value)
return "\\n".join(parts)[:240]
def request_log_evidence(request_id):
proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", "--since=5m", "--tail=800"])
stdout = proc.stdout.decode("utf-8", errors="replace")
lines = [line for line in stdout.splitlines() if request_id in line]
failovers = []
final = None
for line in lines:
json_start = line.find("{")
if json_start < 0:
continue
try:
item = json.loads(line[json_start:])
except Exception:
continue
if "upstream_failover_switching" in line:
failovers.append({
"accountId": item.get("account_id"),
"upstreamStatus": item.get("upstream_status"),
"switchCount": item.get("switch_count"),
"maxSwitches": item.get("max_switches"),
})
if "http request completed" in line:
final = {
"accountId": item.get("account_id"),
"statusCode": item.get("status_code"),
"latencyMs": item.get("latency_ms"),
"path": item.get("path"),
}
return {
"requestId": request_id,
"matchedLogLineCount": len(lines),
"failovers": failovers,
"final": final,
"logsExitCode": proc.returncode,
"logsStderr": text(proc.stderr, 1000),
}
def validate_gateway_responses(api_key):
request_id = "unidesk-codex-pool-validate-" + str(int(time.time() * 1000))
payload = {
"model": RESPONSES_SMOKE_MODEL,
"input": "Reply exactly: unidesk-sub2api-validate-ok",
"stream": False,
"store": False,
"max_output_tokens": 32,
}
body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
script = r'''
set -eu
token="$1"
request_id="$2"
tmp="$(mktemp)"
trap 'rm -f "$tmp"' EXIT
cat > "$tmp"
curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X POST \
-H "Authorization: Bearer $token" \
-H 'Content-Type: application/json' \
-H "X-Request-ID: $request_id" \
-H "OpenAI-Client-Request-ID: $request_id" \
--data-binary @"$tmp" \
http://127.0.0.1:8080/v1/responses
'''
started = time.time()
proc = run([
"kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
"--", "sh", "-c", script, "sh", api_key, request_id,
], body)
resp = parse_curl_output(proc)
evidence = request_log_evidence(request_id)
parsed = resp.get("json")
failover_count = len(evidence.get("failovers") or [])
return {
"ok": resp.get("ok"),
"degraded": failover_count > 0,
"outcome": "succeeded-with-failover" if resp.get("ok") and failover_count > 0 else ("succeeded" if resp.get("ok") else "failed"),
"httpStatus": resp.get("httpStatus"),
"transportExitCode": resp.get("transportExitCode"),
"method": "POST /v1/responses",
"model": RESPONSES_SMOKE_MODEL,
"requestId": request_id,
"durationMs": int((time.time() - started) * 1000),
"outputTextPreview": response_output_preview(parsed),
"bodyPreview": "" if resp.get("ok") else text(resp.get("body", ""), 800),
"stderr": resp.get("stderr", ""),
"evidence": evidence,
"valuesPrinted": False,
}
def bool_value(value):
if isinstance(value, bool):
return value
@@ -2413,8 +2534,10 @@ def run_sync():
owner_balance = ensure_pool_owner_balance(token, api_key_result["userId"])
owner_concurrency = ensure_pool_owner_concurrency(token, api_key_result["userId"])
gateway = validate_gateway(api_key)
responses_smoke = validate_gateway_responses(api_key)
return {
"ok": gateway["ok"] is True and owner_concurrency["ok"] is True and capacity_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True,
"ok": gateway["ok"] is True and responses_smoke["ok"] is True and owner_concurrency["ok"] is True and capacity_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True,
"degraded": bool(responses_smoke.get("degraded")),
"mode": "sync",
"namespace": NAMESPACE,
"serviceDns": SERVICE_DNS,
@@ -2447,7 +2570,7 @@ def run_sync():
},
"ownerBalance": owner_balance,
"ownerConcurrency": owner_concurrency,
"validation": {"gatewayModels": gateway},
"validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke},
}
def run_validate():
@@ -2465,8 +2588,10 @@ def run_validate():
ws_v2_status = account_ws_v2_status(token)
temp_unschedulable_status = account_temp_unschedulable_status(token)
gateway = validate_gateway(api_key)
responses_smoke = validate_gateway_responses(api_key)
return {
"ok": gateway["ok"] is True and (owner_concurrency is None or owner_concurrency["ok"] is True) and capacity_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True,
"ok": gateway["ok"] is True and responses_smoke["ok"] is True and (owner_concurrency is None or owner_concurrency["ok"] is True) and capacity_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True,
"degraded": bool(responses_smoke.get("degraded")),
"mode": "validate",
"namespace": NAMESPACE,
"serviceDns": SERVICE_DNS,
@@ -2484,7 +2609,7 @@ def run_validate():
"capacity": capacity_status,
"webSocketsV2": ws_v2_status,
"tempUnschedulable": temp_unschedulable_status,
"validation": {"gatewayModels": gateway},
"validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke},
}
try: