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

fix: add sub2api codex load factor controls

Browse Source
This commit is contained in:
Codex
2026-06-09 10:31:56 +00:00
parent 1120ed286e
commit fb4cc4baf6
5 changed files with 110 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/platform-infra-sub2api-codex.ts
+95 -3
View File
@@ -48,6 +48,7 @@ interface CodexProfile {
upstreamUserAgent: string | null;
priority: number;
capacity: number;
loadFactor: number;
tempUnschedulable: CodexTempUnschedulablePolicy;
authOpenAIKeyShape: string;
ok: boolean;
@@ -76,6 +77,7 @@ interface CodexPoolConfig {
minOwnerBalanceUsd: number;
minOwnerConcurrency: number;
defaultAccountCapacity: number;
defaultAccountLoadFactor: number;
defaultTempUnschedulable: CodexTempUnschedulablePolicy;
profiles: CodexPoolProfileConfig[];
publicExposure: CodexPoolPublicExposureConfig;
@@ -93,6 +95,7 @@ interface CodexPoolProfileConfig {
upstreamUserAgent: string | null;
priority: number;
capacity: number | null;
loadFactor: number | null;
tempUnschedulable: CodexTempUnschedulablePolicy;
}
@@ -266,6 +269,7 @@ async function codexPoolSync(config: UniDeskConfig, options: SyncOptions): Promi
minOwnerBalanceUsd: pool.minOwnerBalanceUsd,
minOwnerConcurrency: pool.minOwnerConcurrency,
defaultAccountCapacity: pool.defaultAccountCapacity,
defaultAccountLoadFactor: pool.defaultAccountLoadFactor,
},
profiles: profiles.map((profile) => ({
profile: profile.profile,
@@ -283,6 +287,7 @@ async function codexPoolSync(config: UniDeskConfig, options: SyncOptions): Promi
upstreamUserAgent: profile.upstreamUserAgent,
priority: profile.priority,
capacity: profile.capacity,
loadFactor: profile.loadFactor,
tempUnschedulable: profile.tempUnschedulable,
tempUnschedulableCredentials: renderSub2ApiTempUnschedulableCredentials(profile.tempUnschedulable),
})),
@@ -475,6 +480,7 @@ function collectCodexProfiles(): CodexProfile[] {
upstreamUserAgent: entry.upstreamUserAgent,
priority: entry.priority,
capacity: entry.capacity ?? pool.defaultAccountCapacity,
loadFactor: entry.loadFactor ?? pool.defaultAccountLoadFactor,
tempUnschedulable: entry.tempUnschedulable,
authOpenAIKeyShape: existsSync(authPath) ? "unknown" : "missing",
ok: false,
@@ -542,6 +548,7 @@ function discoverCodexProfileConfigs(codexDir: string, defaultTempUnschedulable
upstreamUserAgent: null,
priority: 1,
capacity: null,
loadFactor: null,
tempUnschedulable: defaultTempUnschedulable,
};
});
@@ -573,6 +580,7 @@ function readCodexPoolConfig(): CodexPoolConfig {
minOwnerBalanceUsd: numberValue(pool.minOwnerBalanceUsd) ?? defaults.minOwnerBalanceUsd,
minOwnerConcurrency: readAccountCapacity(pool.minOwnerConcurrency, "pool.minOwnerConcurrency"),
defaultAccountCapacity: readAccountCapacity(pool.defaultAccountCapacity, "pool.defaultAccountCapacity"),
defaultAccountLoadFactor: readAccountLoadFactor(pool.defaultAccountLoadFactor, "pool.defaultAccountLoadFactor"),
defaultTempUnschedulable,
profiles: readProfileConfig(parsed.profiles, defaults.profiles, defaultTempUnschedulable),
publicExposure: readPublicExposureConfig(parsed.publicExposure, defaults.publicExposure),
@@ -601,6 +609,7 @@ function defaultCodexPoolConfig(): CodexPoolConfig {
// Only used for the no-YAML fallback path; real pool configs must declare pool.minOwnerConcurrency.
minOwnerConcurrency: 1,
defaultAccountCapacity: 5,
defaultAccountLoadFactor: 1,
defaultTempUnschedulable: defaultCodexTempUnschedulablePolicy(),
profiles: [],
publicExposure: {
@@ -713,6 +722,7 @@ function readProfileConfig(value: unknown, defaults: CodexPoolProfileConfig[], d
const upstreamUserAgent = readUpstreamUserAgent(entry.upstreamUserAgent, `profiles.entries[${index}].upstreamUserAgent`);
const priority = readAccountPriority(entry.priority, `profiles.entries[${index}].priority`);
const capacity = entry.capacity === undefined || entry.capacity === null ? null : readAccountCapacity(entry.capacity, `profiles.entries[${index}].capacity`);
const loadFactor = entry.loadFactor === undefined || entry.loadFactor === null ? null : readAccountLoadFactor(entry.loadFactor, `profiles.entries[${index}].loadFactor`);
const tempUnschedulable = readTempUnschedulablePolicy(entry.tempUnschedulable, `profiles.entries[${index}].tempUnschedulable`, defaultTempUnschedulable);
return {
profile,
@@ -725,6 +735,7 @@ function readProfileConfig(value: unknown, defaults: CodexPoolProfileConfig[], d
upstreamUserAgent,
priority,
capacity,
loadFactor,
tempUnschedulable,
};
});
@@ -764,6 +775,14 @@ function readAccountCapacity(value: unknown, key: string): number {
return capacity;
}
function readAccountLoadFactor(value: unknown, key: string): number {
const loadFactor = numberValue(value);
if (loadFactor === null || !Number.isInteger(loadFactor) || loadFactor < 1 || loadFactor > 1000) {
throw new Error(`${codexPoolConfigPath}.${key} must be an integer from 1 to 1000`);
}
return loadFactor;
}
function readTempUnschedulablePolicy(value: unknown, key: string, fallback: CodexTempUnschedulablePolicy): CodexTempUnschedulablePolicy {
if (value === undefined || value === null) return cloneTempUnschedulablePolicy(fallback);
if (!isRecord(value)) throw new Error(`${codexPoolConfigPath}.${key} must be a YAML object`);
@@ -961,6 +980,7 @@ function redactProfile(profile: CodexProfile): Record<string, unknown> {
upstreamUserAgent: profile.upstreamUserAgent,
priority: profile.priority,
capacity: profile.capacity,
loadFactor: profile.loadFactor,
tempUnschedulable: tempUnschedulableSummary(profile.tempUnschedulable),
apiKeyPresent: profile.apiKey !== null && profile.apiKey.length > 0,
apiKeyBytes: profile.apiKey === null ? 0 : Buffer.byteLength(profile.apiKey, "utf8"),
@@ -1006,6 +1026,7 @@ function poolTarget(pool = readCodexPoolConfig()): Record<string, unknown> {
apiKeySecret: `${namespace}/${pool.apiKeySecretName}.${pool.apiKeySecretKey}`,
minOwnerConcurrency: pool.minOwnerConcurrency,
defaultAccountCapacity: pool.defaultAccountCapacity,
defaultAccountLoadFactor: pool.defaultAccountLoadFactor,
valuesPrinted: false,
};
}
@@ -1668,6 +1689,19 @@ function desiredAccountCapacityMap(pool: CodexPoolConfig): Record<string, number
return capacities;
}
function desiredAccountLoadFactorMap(pool: CodexPoolConfig): Record<string, number> {
const codexDir = join(homedir(), ".codex");
const seenAccountNames = new Set<string>();
const configs = pool.profiles.length > 0 ? pool.profiles : discoverCodexProfileConfigs(codexDir, pool.defaultTempUnschedulable);
const loadFactors: Record<string, number> = {};
for (const entry of configs) {
const accountName = entry.accountName ?? uniqueAccountName(entry.profile, seenAccountNames);
seenAccountNames.add(accountName);
loadFactors[accountName] = entry.loadFactor ?? pool.defaultAccountLoadFactor;
}
return loadFactors;
}
function desiredAccountWebSocketsV2ModeMap(pool: CodexPoolConfig): Record<string, OpenAIResponsesWebSocketsV2Mode | null> {
const codexDir = join(homedir(), ".codex");
const seenAccountNames = new Set<string>();
@@ -1719,8 +1753,10 @@ 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)}
POOL_DEFAULT_ACCOUNT_LOAD_FACTOR = ${JSON.stringify(pool.defaultAccountLoadFactor)}
RESPONSES_SMOKE_MODEL = ${JSON.stringify(pool.localCodex.responsesSmokeModel)}
EXPECTED_ACCOUNT_CAPACITIES = ${JSON.stringify(desiredAccountCapacityMap(pool))}
EXPECTED_ACCOUNT_LOAD_FACTORS = ${JSON.stringify(desiredAccountLoadFactorMap(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)))})
MODE = "${mode}"
@@ -1954,7 +1990,7 @@ def account_payload(profile, group_id):
"concurrency": int(profile.get("capacity", 5) or 5),
"priority": int(profile.get("priority", 1) or 1),
"rate_multiplier": 1,
"load_factor": 1,
"load_factor": int(profile.get("loadFactor", POOL_DEFAULT_ACCOUNT_LOAD_FACTOR) or POOL_DEFAULT_ACCOUNT_LOAD_FACTOR),
"group_ids": [group_id],
"confirm_mixed_channel_risk": True,
}
@@ -1988,7 +2024,9 @@ def ensure_accounts(token, profiles, group_id):
"openaiResponsesWebSocketsV2Mode": profile.get("openaiResponsesWebSocketsV2Mode"),
"priority": int(profile.get("priority", 1) or 1),
"capacity": int(profile.get("capacity", 5) or 5),
"loadFactor": int(profile.get("loadFactor", POOL_DEFAULT_ACCOUNT_LOAD_FACTOR) or POOL_DEFAULT_ACCOUNT_LOAD_FACTOR),
"runtimeConcurrency": data.get("concurrency") if isinstance(data, dict) else None,
"runtimeLoadFactor": data.get("load_factor") if isinstance(data, dict) else None,
"tempUnschedulableConfigured": bool(payload["credentials"].get("temp_unschedulable_enabled")),
"tempUnschedulableRuleCount": len(payload["credentials"].get("temp_unschedulable_rules") or []),
"upstreamUserAgentConfigured": bool(profile.get("upstreamUserAgent")),
@@ -2443,6 +2481,56 @@ def account_capacity_status(token):
"valuesPrinted": False,
}
def account_load_factor_status(token):
accounts = list_accounts(token)
by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)}
items = []
missing = []
mismatched = []
for name in sorted(EXPECTED_ACCOUNT_LOAD_FACTORS):
expected = int(EXPECTED_ACCOUNT_LOAD_FACTORS[name])
account = by_name.get(name)
if account is None:
missing.append(name)
items.append({
"accountName": name,
"accountId": None,
"expectedLoadFactor": expected,
"runtimeLoadFactor": None,
"ok": False,
})
continue
runtime_raw = account.get("load_factor")
if runtime_raw is None:
detail = get_account_detail(token, account)
runtime_raw = detail.get("load_factor") if isinstance(detail, dict) else None
try:
runtime = int(runtime_raw)
except Exception:
runtime = runtime_raw
ok = runtime == expected
if not ok:
mismatched.append(name)
items.append({
"accountName": name,
"accountId": account.get("id"),
"expectedLoadFactor": expected,
"runtimeLoadFactor": runtime,
"priority": account.get("priority"),
"status": account.get("status"),
"schedulable": account.get("schedulable"),
"ok": ok,
})
return {
"ok": len(missing) == 0 and len(mismatched) == 0,
"defaultAccountLoadFactor": POOL_DEFAULT_ACCOUNT_LOAD_FACTOR,
"desired": len(EXPECTED_ACCOUNT_LOAD_FACTORS),
"missing": missing,
"mismatched": mismatched,
"items": items,
"valuesPrinted": False,
}
def account_ws_v2_status(token):
accounts = list_accounts(token)
by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)}
@@ -2527,6 +2615,7 @@ def run_sync():
raise RuntimeError("pool group id missing after ensure")
account_results, pruned_account_results = ensure_accounts(token, profiles, group_id)
capacity_status = account_capacity_status(token)
load_factor_status = account_load_factor_status(token)
ws_v2_status = account_ws_v2_status(token)
temp_unschedulable_status = account_temp_unschedulable_status(token)
api_key, secret_action, secret_apply_stdout = ensure_api_key_secret(group_id)
@@ -2536,7 +2625,7 @@ def run_sync():
gateway = validate_gateway(api_key)
responses_smoke = validate_gateway_responses(api_key)
return {
"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,
"ok": gateway["ok"] is True and responses_smoke["ok"] is True and owner_concurrency["ok"] is True and capacity_status["ok"] is True and load_factor_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,
@@ -2554,6 +2643,7 @@ def run_sync():
"valuesPrinted": False,
},
"capacity": capacity_status,
"loadFactor": load_factor_status,
"webSocketsV2": ws_v2_status,
"tempUnschedulable": temp_unschedulable_status,
"apiKey": {
@@ -2585,12 +2675,13 @@ def run_validate():
owner_balance = ensure_pool_owner_balance(token, key_item["user_id"])
owner_concurrency = ensure_pool_owner_concurrency(token, key_item["user_id"])
capacity_status = account_capacity_status(token)
load_factor_status = account_load_factor_status(token)
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 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,
"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 load_factor_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,
@@ -2607,6 +2698,7 @@ def run_validate():
"ownerBalance": owner_balance,
"ownerConcurrency": owner_concurrency,
"capacity": capacity_status,
"loadFactor": load_factor_status,
"webSocketsV2": ws_v2_status,
"tempUnschedulable": temp_unschedulable_status,
"validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke},