export const remotePythonValidationScript = `def validate_gateway(api_key): resp = curl_api("GET", "/v1/models", bearer=api_key) parsed = resp.get("json") model_count = None if isinstance(parsed, dict) and isinstance(parsed.get("data"), list): model_count = len(parsed["data"]) return { "ok": resp.get("ok"), "httpStatus": resp.get("httpStatus"), "transportExitCode": resp.get("transportExitCode"), "modelCount": model_count, "bodyPreview": text(resp.get("body", ""), 500) if not resp.get("ok") else "", "stderr": resp.get("stderr", ""), "method": "GET /v1/models", "serviceDns": SERVICE_DNS, "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 = runtime_logs("5m", 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 recent_compact_gateway_evidence(): proc = runtime_logs("6h", 2500) stdout = proc.stdout.decode("utf-8", errors="replace") failures = [] successes = [] failovers = [] final_errors = [] context_canceled = [] for line in stdout.splitlines(): if "/responses/compact" not in line and "remote_compact" not in line: continue json_start = line.find("{") if json_start < 0: continue try: item = json.loads(line[json_start:]) except Exception: continue path = item.get("path") entry = { "requestId": item.get("request_id"), "clientRequestId": item.get("client_request_id"), "accountId": item.get("account_id"), "statusCode": item.get("status_code"), "upstreamStatus": item.get("upstream_status"), "latencyMs": item.get("latency_ms"), "path": path, } if "codex.remote_compact.failed" in line: failures.append(entry) elif "codex.remote_compact.succeeded" in line: successes.append(entry) elif "upstream_failover_switching" in line and path == "/responses/compact": failovers.append({ **entry, "switchCount": item.get("switch_count"), "maxSwitches": item.get("max_switches"), }) elif "http request completed" in line and path == "/responses/compact" and isinstance(item.get("status_code"), int) and item.get("status_code") >= 400: final_errors.append(entry) if "context canceled" in line and path == "/responses/compact": context_canceled.append(entry) return { "ok": True, "degraded": len(failures) > 0 or len(final_errors) > 0 or len(context_canceled) > 0, "window": "6h", "tailLines": 2500, "failureCount": len(failures), "successCount": len(successes), "failoverCount": len(failovers), "finalErrorCount": len(final_errors), "contextCanceledCount": len(context_canceled), "recentFailures": failures[-5:], "recentSuccesses": successes[-5:], "recentFailovers": failovers[-8:], "recentFinalErrors": final_errors[-5:], "recentContextCanceled": context_canceled[-5:], "logsExitCode": proc.returncode, "logsStderr": text(proc.stderr, 1000), "valuesPrinted": False, } def is_ignored_probe_noise(entry): for value in (entry.get("requestId"), entry.get("clientRequestId")): if isinstance(value, str) and value.startswith("unidesk-400-model-probe-"): return True return False def filter_ignored_probe_noise(items): return [item for item in items if not is_ignored_probe_noise(item)] def ignored_probe_noise(items, section): result = [] for item in items: if is_ignored_probe_noise(item): probe = dict(item) probe["section"] = section result.append(probe) return result def group_by_request_id(items): grouped = {} for item in items: request_id = item.get("requestId") if not isinstance(request_id, str) or not request_id: continue grouped.setdefault(request_id, []).append(item) return grouped def failover_budget_exhausted_evidence(failovers, final_errors): final_by_request = {} for item in final_errors: request_id = item.get("requestId") if isinstance(request_id, str) and request_id: final_by_request[request_id] = item exhausted = [] for request_id, request_failovers in group_by_request_id(failovers).items(): final = final_by_request.get(request_id) if not final: continue last = request_failovers[-1] switch_count = last.get("switchCount") max_switches = last.get("maxSwitches") final_status = final.get("statusCode") if ( isinstance(switch_count, int) and isinstance(max_switches, int) and max_switches > 0 and switch_count >= max_switches and isinstance(final_status, int) and final_status >= 500 ): exhausted.append({ "requestId": request_id, "clientRequestId": final.get("clientRequestId") or last.get("clientRequestId"), "path": final.get("path") or last.get("path"), "finalAccountId": final.get("accountId"), "finalStatusCode": final_status, "switchCount": switch_count, "maxSwitches": max_switches, "lastFailoverAccountId": last.get("accountId"), "lastUpstreamStatus": last.get("upstreamStatus"), }) return exhausted def recent_responses_gateway_evidence(): proc = runtime_logs("6h", 2500) stdout = proc.stdout.decode("utf-8", errors="replace") failovers = [] forward_failures = [] final_errors = [] context_canceled = [] slow_final_errors = [] for line in stdout.splitlines(): if '"/responses"' not in line and '"/v1/responses"' not in line: continue json_start = line.find("{") if json_start < 0: continue try: item = json.loads(line[json_start:]) except Exception: continue path = item.get("path") if path not in ("/responses", "/v1/responses"): continue entry = { "requestId": item.get("request_id"), "clientRequestId": item.get("client_request_id"), "accountId": item.get("account_id"), "statusCode": item.get("status_code"), "upstreamStatus": item.get("upstream_status"), "latencyMs": item.get("latency_ms"), "path": path, } if "upstream_failover_switching" in line: failovers.append({ **entry, "switchCount": item.get("switch_count"), "maxSwitches": item.get("max_switches"), }) elif "openai.forward_failed" in line: forward_failures.append({ **entry, "errorPreview": text(str(item.get("error") or ""), 500), "fallbackErrorResponseWritten": item.get("fallback_error_response_written"), "upstreamErrorResponseAlreadyWritten": item.get("upstream_error_response_already_written"), }) elif "http request completed" in line and isinstance(item.get("status_code"), int) and item.get("status_code") >= 400: final_errors.append(entry) latency_ms = item.get("latency_ms") if isinstance(latency_ms, int) and latency_ms >= 30000: slow_final_errors.append(entry) if "context canceled" in line: context_canceled.append(entry) visible_failovers = filter_ignored_probe_noise(failovers) visible_forward_failures = filter_ignored_probe_noise(forward_failures) visible_final_errors = filter_ignored_probe_noise(final_errors) visible_slow_final_errors = filter_ignored_probe_noise(slow_final_errors) visible_context_canceled = filter_ignored_probe_noise(context_canceled) failover_budget_exhausted = failover_budget_exhausted_evidence(visible_failovers, visible_final_errors) probe_noise = ( ignored_probe_noise(failovers, "failovers") + ignored_probe_noise(forward_failures, "forwardFailures") + ignored_probe_noise(final_errors, "finalErrors") + ignored_probe_noise(slow_final_errors, "slowFinalErrors") + ignored_probe_noise(context_canceled, "contextCanceled") ) return { "ok": True, "degraded": len(visible_forward_failures) > 0 or len(visible_final_errors) > 0 or len(visible_context_canceled) > 0, "window": "6h", "tailLines": 2500, "failoverCount": len(visible_failovers), "forwardFailureCount": len(visible_forward_failures), "finalErrorCount": len(visible_final_errors), "slowFinalErrorCount": len(visible_slow_final_errors), "contextCanceledCount": len(visible_context_canceled), "ignoredProbeNoiseCount": len(probe_noise), "failoverBudgetExhausted": failover_budget_exhausted[-8:], "rawCounts": { "failoverCount": len(failovers), "forwardFailureCount": len(forward_failures), "finalErrorCount": len(final_errors), "slowFinalErrorCount": len(slow_final_errors), "contextCanceledCount": len(context_canceled), }, "recentFailovers": visible_failovers[-8:], "recentForwardFailures": visible_forward_failures[-8:], "recentFinalErrors": visible_final_errors[-8:], "recentSlowFinalErrors": visible_slow_final_errors[-5:], "recentContextCanceled": visible_context_canceled[-5:], "recentProbeNoise": probe_noise[-5:], "logsExitCode": proc.returncode, "logsStderr": text(proc.stderr, 1000), "valuesPrinted": False, } 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" url="$3" 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" \ "$url" ''' started = time.time() if RUNTIME_MODE == "host-docker": if not isinstance(HOST_DOCKER_APP_PORT, int): raise RuntimeError("host-docker app port missing") proc = run(["sh", "-c", script, "sh", api_key, request_id, f"http://127.0.0.1:{HOST_DOCKER_APP_PORT}/v1/responses"], body) else: proc = run([ "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD, "--", "sh", "-c", script, "sh", api_key, request_id, "http://127.0.0.1:8080/v1/responses", ], 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 if isinstance(value, str): if value.lower() == "true": return True if value.lower() == "false": return False return False def normalize_temp_unschedulable_credentials(credentials): if not isinstance(credentials, dict): credentials = {} enabled = bool_value(credentials.get("temp_unschedulable_enabled")) raw_rules = credentials.get("temp_unschedulable_rules") if isinstance(raw_rules, str): try: raw_rules = json.loads(raw_rules) except json.JSONDecodeError: raw_rules = [] rules = [] if isinstance(raw_rules, list): for rule in raw_rules: if not isinstance(rule, dict): continue error_code = rule.get("error_code", rule.get("statusCode")) duration_minutes = rule.get("duration_minutes", rule.get("durationMinutes")) keywords = rule.get("keywords") if not isinstance(error_code, int) or not isinstance(duration_minutes, int) or not isinstance(keywords, list): continue clean_keywords = [item for item in keywords if isinstance(item, str) and item.strip()] if not clean_keywords: continue description = rule.get("description") if isinstance(rule.get("description"), str) else "" rules.append({ "error_code": error_code, "keywords": clean_keywords, "duration_minutes": duration_minutes, "description": description, }) return { "enabled": enabled, "rules": rules, } def summarize_temp_unschedulable_rules(rules): return [{ "errorCode": rule.get("error_code"), "durationMinutes": rule.get("duration_minutes"), "keywordCount": len(rule.get("keywords") or []), "keywords": rule.get("keywords") or [], "hasDescription": bool(rule.get("description")), } for rule in rules] def success_body_reclassification_requirement(): for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE): expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name]) if expected["enabled"] is not True: continue for rule in expected["rules"]: error_code = rule.get("error_code") keywords = rule.get("keywords") or [] if isinstance(error_code, int) and 200 <= error_code < 300 and keywords: return { "required": True, "sourceAccountName": name, "statusCode": error_code, "keywords": keywords, "representativeKeyword": keywords[0], "durationMinutes": rule.get("duration_minutes"), } return { "required": False, "sourceAccountName": None, "statusCode": None, "keywords": [], "representativeKeyword": None, "durationMinutes": None, } def model_routing_400_failover_requirement(): preferred = ["invalid_encrypted_content", "encrypted content", "could not be verified", "could not be decrypted", "暂不支持", "可用模型", "unsupported model", "model not supported", "does not support", "not supported", "model_not_found", "no available channel for model"] for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE): expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name]) if expected["enabled"] is not True: continue for rule in expected["rules"]: error_code = rule.get("error_code") keywords = rule.get("keywords") or [] if error_code != 400 or not keywords: continue representative_keyword = next((item for item in preferred if item in keywords), keywords[0]) return { "required": True, "sourceAccountName": name, "statusCode": error_code, "keywords": keywords, "representativeKeyword": representative_keyword, "durationMinutes": rule.get("duration_minutes"), } return { "required": False, "sourceAccountName": None, "statusCode": None, "keywords": [], "representativeKeyword": None, "durationMinutes": None, } def delete_probe_resource(token, method, path, label): if not path: return {"label": label, "ok": True, "skipped": True} resp = curl_api(method, path, bearer=token) ok = resp.get("ok") is True or resp.get("httpStatus") in (404, 410) return { "label": label, "ok": ok, "method": method, "path": path, "httpStatus": resp.get("httpStatus"), "transportExitCode": resp.get("transportExitCode"), "bodyPreview": "" if ok else text(resp.get("body", ""), 500), "valuesPrinted": False, } def validate_runtime_capabilities(token): success_body = success_body_reclassification_requirement() model_routing_400 = model_routing_400_failover_requirement() return { "ok": success_body.get("required") is not True and model_routing_400.get("required") is True, "runtimeImage": app_pod_runtime_image(), "successBodyReclassification": { "ok": success_body.get("required") is not True, "required": success_body.get("required"), "outcome": "not-required-by-yaml" if success_body.get("required") is not True else "requires-source-review-and-real-traffic-evidence", "requirement": success_body, "valuesPrinted": False, }, "modelRouting400Failover": { "ok": model_routing_400.get("required") is True, "required": model_routing_400.get("required"), "outcome": "yaml-rules-present-runtime-observed-by-real-traffic", "requirement": model_routing_400, "evidence": "Use Sub2API source review plus validation.gatewayResponsesRecent and Artificer/real request ids for runtime proof; default validate does not create mock upstreams or temporary failover accounts.", "valuesPrinted": False, }, "valuesPrinted": False, } def app_pod_runtime_image(): if RUNTIME_MODE == "host-docker": proc = docker(["inspect", HOST_DOCKER_APP_CONTAINER]) if proc.returncode != 0: return { "container": HOST_DOCKER_APP_CONTAINER, "error": text(proc.stderr, 1000) or text(proc.stdout, 1000), } try: data = json.loads(proc.stdout.decode("utf-8")) item = data[0] if isinstance(data, list) and data else {} except Exception as exc: return {"container": HOST_DOCKER_APP_CONTAINER, "error": str(exc)} state = item.get("State") if isinstance(item, dict) and isinstance(item.get("State"), dict) else {} health = state.get("Health") if isinstance(state.get("Health"), dict) else {} config = item.get("Config") if isinstance(item, dict) and isinstance(item.get("Config"), dict) else {} return { "container": HOST_DOCKER_APP_CONTAINER, "id": (item.get("Id") or "")[:12] if isinstance(item.get("Id"), str) else None, "image": config.get("Image"), "imageID": item.get("Image"), "ready": state.get("Running") is True and (not health or health.get("Status") in (None, "healthy")), "restartCount": item.get("RestartCount"), "startedAt": state.get("StartedAt"), "health": health.get("Status"), } try: pod = kube_json(["-n", NAMESPACE, "get", "pod", APP_POD], f"pod/{APP_POD}") spec_containers = ((pod.get("spec") or {}).get("containers") or []) if isinstance(pod, dict) else [] status_containers = ((pod.get("status") or {}).get("containerStatuses") or []) if isinstance(pod, dict) else [] spec = next((item for item in spec_containers if item.get("name") == "sub2api"), spec_containers[0] if spec_containers else {}) status = next((item for item in status_containers if item.get("name") == "sub2api"), status_containers[0] if status_containers else {}) return { "pod": APP_POD, "image": spec.get("image"), "imageID": status.get("imageID"), "ready": status.get("ready"), "restartCount": status.get("restartCount"), } except Exception as exc: return {"pod": APP_POD, "error": str(exc)} def get_account_detail(token, account): account_id = account.get("id") if isinstance(account, dict) else None if account_id is None: return account data = ensure_success(curl_api("GET", f"/api/v1/admin/accounts/{account_id}", bearer=token), f"get account {account.get('name')}") return data if isinstance(data, dict) else account def account_temp_unschedulable_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 = [] enabled_names = [] for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE): expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name]) account = by_name.get(name) if account is None: missing.append(name) items.append({ "accountName": name, "accountId": None, "expectedEnabled": expected["enabled"], "runtimeEnabled": None, "expectedRuleCount": len(expected["rules"]), "runtimeRuleCount": None, "ok": False, }) continue detail = get_account_detail(token, account) credentials = detail.get("credentials") if isinstance(detail.get("credentials"), dict) else {} runtime = normalize_temp_unschedulable_credentials(credentials) temp_until = detail.get("temp_unschedulable_until") or detail.get("tempUnschedulableUntil") temp_reason = detail.get("temp_unschedulable_reason") or detail.get("tempUnschedulableReason") or "" ok = runtime == expected if expected["enabled"]: enabled_names.append(name) if not ok: mismatched.append(name) items.append({ "accountName": name, "accountId": account.get("id"), "expectedEnabled": expected["enabled"], "runtimeEnabled": runtime["enabled"], "expectedRuleCount": len(expected["rules"]), "runtimeRuleCount": len(runtime["rules"]), "expectedRules": summarize_temp_unschedulable_rules(expected["rules"]), "runtimeRules": summarize_temp_unschedulable_rules(runtime["rules"]), "status": account.get("status"), "schedulable": account.get("schedulable"), "tempUnschedulableUntil": temp_until, "tempUnschedulableReasonPreview": text(str(temp_reason), 500) if temp_reason else "", "tempUnschedulableSet": temp_until is not None or bool(temp_reason), "ok": ok, }) return { "ok": len(missing) == 0 and len(mismatched) == 0, "desired": len(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE), "enabled": enabled_names, "missing": missing, "mismatched": mismatched, "items": items, "valuesPrinted": False, } def account_capacity_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 = [] expected_capacity_total = 0 runtime_concurrency_total = 0 schedulable_runtime_concurrency_total = 0 available_runtime_concurrency_total = 0 temp_unschedulable_runtime_concurrency_total = 0 for name in sorted(EXPECTED_ACCOUNT_CAPACITIES): expected = int(EXPECTED_ACCOUNT_CAPACITIES[name]) expected_capacity_total += expected account = by_name.get(name) if account is None: missing.append(name) items.append({ "accountName": name, "accountId": None, "expectedCapacity": expected, "runtimeConcurrency": None, "ok": False, }) continue runtime = account.get("concurrency") runtime_int = runtime if isinstance(runtime, int) else 0 runtime_concurrency_total += runtime_int if account.get("schedulable") is True: runtime_status = account.get("status") if runtime_status is None or runtime_status == "active": runtime_status_ok = True else: runtime_status_ok = False if runtime_status_ok: schedulable_runtime_concurrency_total += runtime_int temp_until = account.get("temp_unschedulable_until") or account.get("tempUnschedulableUntil") temp_reason = account.get("temp_unschedulable_reason") or account.get("tempUnschedulableReason") if temp_until is None and not bool(temp_reason): available_runtime_concurrency_total += runtime_int else: temp_unschedulable_runtime_concurrency_total += runtime_int ok = runtime == expected if not ok: mismatched.append(name) items.append({ "accountName": name, "accountId": account.get("id"), "expectedCapacity": expected, "runtimeConcurrency": runtime, "priority": account.get("priority"), "status": account.get("status"), "schedulable": account.get("schedulable"), "ok": ok, }) return { "ok": len(missing) == 0 and len(mismatched) == 0, "defaultAccountCapacity": POOL_DEFAULT_ACCOUNT_CAPACITY, "desired": len(EXPECTED_ACCOUNT_CAPACITIES), "totals": { "expectedCapacityTotal": expected_capacity_total, "runtimeConcurrencyTotal": runtime_concurrency_total, "schedulableRuntimeConcurrencyTotal": schedulable_runtime_concurrency_total, "availableRuntimeConcurrencyTotal": available_runtime_concurrency_total, "tempUnschedulableRuntimeConcurrencyTotal": temp_unschedulable_runtime_concurrency_total, "minOwnerConcurrency": MIN_OWNER_CONCURRENCY, "minOwnerConcurrencySource": MIN_OWNER_CONCURRENCY_SOURCE, }, "missing": missing, "mismatched": mismatched, "items": items, "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)} items = [] missing = [] mismatched = [] enabled_names = [] unschedulable_enabled = [] schedulable_enabled = [] for name in sorted(EXPECTED_ACCOUNT_WS_MODES): expected_mode = EXPECTED_ACCOUNT_WS_MODES[name] expected_enabled = expected_mode not in (None, "", "off") account = by_name.get(name) if account is None: missing.append(name) items.append({ "accountName": name, "accountId": None, "expectedMode": expected_mode, "expectedEnabled": expected_enabled, "runtimeMode": None, "runtimeEnabled": None, "ok": False, }) continue extra = account.get("extra") if isinstance(account.get("extra"), dict) else {} runtime_mode = extra.get("openai_apikey_responses_websockets_v2_mode") runtime_enabled = extra.get("openai_apikey_responses_websockets_v2_enabled") schedulable = account.get("schedulable") if expected_mode == "off": ok = runtime_mode == "off" and runtime_enabled is False elif expected_mode is None: ok = runtime_mode in (None, "", "off") and runtime_enabled in (None, False) else: ok = runtime_mode == expected_mode and runtime_enabled is True if not ok: mismatched.append(name) if expected_enabled: enabled_names.append(name) if schedulable is True: schedulable_enabled.append(name) else: unschedulable_enabled.append(name) items.append({ "accountName": name, "accountId": account.get("id"), "expectedMode": expected_mode, "expectedEnabled": expected_enabled, "runtimeMode": runtime_mode, "runtimeEnabled": runtime_enabled, "status": account.get("status"), "schedulable": schedulable, "ok": ok and ((not expected_enabled) or schedulable is True), }) availability_ok = len(enabled_names) == 0 or len(schedulable_enabled) > 0 return { "ok": len(missing) == 0 and len(mismatched) == 0 and availability_ok, "desired": len(EXPECTED_ACCOUNT_WS_MODES), "enabled": enabled_names, "schedulableEnabled": schedulable_enabled, "unschedulableEnabled": unschedulable_enabled, "missing": missing, "mismatched": mismatched, "items": items, "valuesPrinted": False, } def api_key_preview(api_key): if len(api_key) <= 14: return "***" return api_key[:10] + "..." + api_key[-4:] def secret_fingerprint(value): return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] `;