// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. remote-python-sync module for scripts/src/platform-infra-sub2api-codex.ts. // Moved mechanically from scripts/src/platform-infra-sub2api-codex.ts:4265-5400 for #903. import { chmodSync, copyFileSync, existsSync, mkdirSync, readFileSync, statSync, writeFileSync } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import type { UniDeskConfig } from "../config"; import { rootPath } from "../config"; import type { RenderedCliResult } from "../output"; import { applyPk01CaddyBlock, prepareFrpcSecret, renderFrpcManifest, type PublicServiceExposure, type PublicServiceTarget } from "../platform-infra-public-service"; import { shortSha256Fingerprint } from "../platform-infra-ops-library"; import { codexPoolSentinelSummary, codexPoolSentinelRuntimeImage, readCodexPoolSentinelConfig, renderCodexPoolSentinelManifest, type CodexPoolSentinelConfig, type CodexPoolSentinelProfileSecret, } from "../platform-infra-sub2api-codex-sentinel"; import { parseEnvFile, readTextFile, redactRepoPath, requiredEnvValue } from "../secrets"; import { runSshCommandCapture, type SshCaptureResult } from "../ssh"; import type { CodexPoolConfig, CodexPoolRuntimeTarget } from "./types"; import { desiredAccountCapacityMap, desiredAccountLoadFactorMap, desiredAccountTempUnschedulableMap, desiredAccountWebSocketsV2ModeMap } from "./accounts"; import { resolvedManualAccountProtections } from "./public-exposure"; import { fieldManager } from "./types"; export function remotePythonScript(mode: "sync" | "validate" | "trace" | "cleanup-probes" | "sentinel-probe", encodedPayload: string, pool: CodexPoolConfig, target: CodexPoolRuntimeTarget): string { return ` set -u python3 - <<'PY' import base64 import json import re import secrets import string import subprocess import sys import time from datetime import datetime, timezone, timedelta from urllib.parse import quote TARGET_ID = ${JSON.stringify(target.id)} NAMESPACE = ${JSON.stringify(target.namespace)} SERVICE_NAME = ${JSON.stringify(target.serviceName)} SERVICE_DNS = ${JSON.stringify(target.serviceDns)} FIELD_MANAGER = "${fieldManager}" APP_SECRET_NAME = ${JSON.stringify(target.appSecretName)} POOL_GROUP_NAME = "${pool.groupName}" POOL_GROUP_DESCRIPTION = ${JSON.stringify(pool.groupDescription)} POOL_API_KEY_NAME = "${pool.apiKeyName}" POOL_API_KEY_SECRET_NAME = "${pool.apiKeySecretName}" POOL_API_KEY_SECRET_KEY = "${pool.apiKeySecretKey}" POOL_ADMIN_EMAIL_DEFAULT = ${JSON.stringify(pool.adminEmailDefault)} MIN_OWNER_BALANCE_USD = ${JSON.stringify(pool.minOwnerBalanceUsd)} MIN_OWNER_CONCURRENCY = ${JSON.stringify(pool.minOwnerConcurrency)} MIN_OWNER_CONCURRENCY_SOURCE = ${JSON.stringify(pool.minOwnerConcurrencySource)} POOL_DEFAULT_ACCOUNT_PRIORITY = ${JSON.stringify(pool.defaultAccountPriority)} 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)))}) MANUAL_ACCOUNT_PROTECTIONS = json.loads(${JSON.stringify(JSON.stringify(resolvedManualAccountProtections(pool, target)))}) SENTINEL_CONFIG = json.loads(${JSON.stringify(JSON.stringify(pool.sentinel))}) TARGET_EGRESS_PROXY = json.loads(${JSON.stringify(JSON.stringify(target.egressProxy))}) MODE = "${mode}" PAYLOAD_B64 = "${encodedPayload}" def run(cmd, input_bytes=None): return subprocess.run(cmd, input=input_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE) def text(data, limit=4000): if isinstance(data, bytes): data = data.decode("utf-8", errors="replace") return data[-limit:] def kubectl(args, input_obj=None): if isinstance(input_obj, str): input_bytes = input_obj.encode("utf-8") else: input_bytes = input_obj return run(["kubectl", *args], input_bytes) def require_kubectl(args, input_obj=None, label="kubectl"): proc = kubectl(args, input_obj) if proc.returncode != 0: raise RuntimeError(f"{label} failed: {text(proc.stderr, 1000)}") return proc.stdout def kube_json(args, label): raw = require_kubectl([*args, "-o", "json"], label=label) return json.loads(raw.decode("utf-8")) def decode_secret_value(name, key): data = kube_json(["-n", NAMESPACE, "get", "secret", name], f"secret/{name}").get("data") or {} if key not in data: return None return base64.b64decode(data[key]).decode("utf-8") def get_config_value(name, key): data = kube_json(["-n", NAMESPACE, "get", "configmap", name], f"configmap/{name}").get("data") or {} value = data.get(key) return value if isinstance(value, str) and value else None def select_app_pod(): pods = kube_json(["-n", NAMESPACE, "get", "pods", "-l", "app.kubernetes.io/name=sub2api"], "sub2api pods").get("items") or [] for pod in pods: status = pod.get("status") or {} if status.get("phase") != "Running": continue statuses = status.get("containerStatuses") or [] if statuses and all(item.get("ready") is True for item in statuses): return pod["metadata"]["name"] if pods: return pods[0]["metadata"]["name"] raise RuntimeError("sub2api app pod not found") APP_POD = select_app_pod() def parse_curl_output(proc): stdout = proc.stdout.decode("utf-8", errors="replace") marker = "\\n__HTTP_CODE__:" pos = stdout.rfind(marker) if pos < 0: return { "ok": False, "httpStatus": 0, "json": None, "body": stdout, "stderr": text(proc.stderr, 1000), "transportExitCode": proc.returncode, } body = stdout[:pos] status_text = stdout[pos + len(marker):].strip() try: http_status = int(status_text[-3:]) except ValueError: http_status = 0 try: parsed = json.loads(body) if body.strip() else None except json.JSONDecodeError: parsed = None return { "ok": proc.returncode == 0 and 200 <= http_status < 300, "httpStatus": http_status, "json": parsed, "body": body, "stderr": text(proc.stderr, 1000), "transportExitCode": proc.returncode, } def utc_iso(offset_seconds=0): return (datetime.now(timezone.utc) + timedelta(seconds=offset_seconds)).strftime("%Y-%m-%dT%H:%M:%SZ") def normalize_runtime_base_url(value): if not isinstance(value, str): return None value = value.strip().rstrip("/") return value or None def empty_to_none(value): return value if isinstance(value, str) and value else None def sentinel_quality_gate_enabled(): return (SENTINEL_CONFIG.get("monitor") or {}).get("enabled") is True and (SENTINEL_CONFIG.get("actions") or {}).get("enabled") is True def account_notes_fingerprint(account): notes = account.get("notes") if isinstance(account, dict) else None if not isinstance(notes, str): return None match = re.search(r"fingerprint=([A-Za-z0-9_-]+)", notes) return match.group(1) if match else None def runtime_account_credentials(account): credentials = account.get("credentials") if isinstance(account, dict) and isinstance(account.get("credentials"), dict) else {} return credentials def runtime_account_extra(account): extra = account.get("extra") if isinstance(account, dict) and isinstance(account.get("extra"), dict) else {} return extra def sentinel_probe_change_reasons(current, profile): if not isinstance(current, dict) or current.get("id") is None: return ["created"] credentials = runtime_account_credentials(current) extra = runtime_account_extra(current) expected_base_url = normalize_runtime_base_url(profile.get("baseUrl")) runtime_base_url = normalize_runtime_base_url(credentials.get("base_url")) expected_user_agent = empty_to_none(profile.get("upstreamUserAgent")) runtime_user_agent = empty_to_none(credentials.get("user_agent")) expected_ws_mode = empty_to_none(profile.get("openaiResponsesWebSocketsV2Mode")) runtime_ws_mode = empty_to_none(extra.get("openai_apikey_responses_websockets_v2_mode")) expected_trust_upstream = profile.get("trustUpstream") is True runtime_trust_upstream = extra.get("unidesk_trust_upstream") is True expected_protect = profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False} runtime_protect = extra.get("unidesk_sentinel_protect") if isinstance(extra.get("unidesk_sentinel_protect"), dict) else {"enabled": False} reasons = [] if empty_to_none(extra.get("unidesk_codex_profile")) != profile.get("profile"): reasons.append("profile") if runtime_base_url != expected_base_url: reasons.append("base-url") if account_notes_fingerprint(current) != profile.get("apiKeyFingerprint"): reasons.append("api-key-fingerprint") if runtime_user_agent != expected_user_agent: reasons.append("upstream-user-agent") if runtime_ws_mode != expected_ws_mode: reasons.append("responses-websockets-v2-mode") if runtime_trust_upstream != expected_trust_upstream: reasons.append("trust-upstream") if runtime_protect != expected_protect: reasons.append("sentinel-protect") return reasons def curl_api(method, path, bearer=None, payload=None): body = b"" if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8") script = r''' set -eu method="$1" url="$2" token="\${3:-}" tmp="$(mktemp)" trap 'rm -f "$tmp"' EXIT cat > "$tmp" if [ -n "$token" ]; then if [ "$method" = "GET" ] && [ ! -s "$tmp" ]; then curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" -H "Authorization: Bearer $token" "$url" else curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' -H "Authorization: Bearer $token" --data-binary @"$tmp" "$url" fi else if [ "$method" = "GET" ] && [ ! -s "$tmp" ]; then curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" "$url" else curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' --data-binary @"$tmp" "$url" fi fi ''' proc = run([ "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD, "--", "sh", "-c", script, "sh", method, f"http://127.0.0.1:8080{path}", bearer or "", ], body) return parse_curl_output(proc) def envelope_data(parsed): if isinstance(parsed, dict) and "data" in parsed: return parsed.get("data") return parsed def ensure_success(resp, label): parsed = resp.get("json") code = parsed.get("code") if isinstance(parsed, dict) else None if not resp.get("ok") or (code is not None and code != 0): message = parsed.get("message") if isinstance(parsed, dict) else text(resp.get("body", ""), 500) raise RuntimeError(f"{label} failed: http={resp.get('httpStatus')} message={message}") return envelope_data(parsed) def ensure_admin_compliance(token): status_resp = curl_api("GET", "/api/v1/admin/compliance", bearer=token) if status_resp.get("httpStatus") == 404: return { "ok": True, "action": "not-supported-by-runtime", "valuesPrinted": False, } status = ensure_success(status_resp, "get admin compliance") if not isinstance(status, dict): return { "ok": True, "action": "status-unstructured", "valuesPrinted": False, } version = status.get("version") required = status.get("required") is True if not required: return { "ok": True, "action": "already-acknowledged", "version": version, "requiredBefore": False, "requiredAfter": False, "phrasePrinted": False, "valuesPrinted": False, } phrase = status.get("ack_phrase_zh") if isinstance(status.get("ack_phrase_zh"), str) and status.get("ack_phrase_zh") else status.get("ack_phrase_en") language = "zh" if isinstance(status.get("ack_phrase_zh"), str) and status.get("ack_phrase_zh") else "en" if not isinstance(phrase, str) or not phrase: raise RuntimeError("admin compliance acknowledgement phrase missing") accepted = ensure_success( curl_api("POST", "/api/v1/admin/compliance/accept", bearer=token, payload={"phrase": phrase, "language": language}), "accept admin compliance", ) required_after = accepted.get("required") if isinstance(accepted, dict) else None return { "ok": required_after is False, "action": "accepted", "version": version, "requiredBefore": True, "requiredAfter": required_after, "phrasePrinted": False, "valuesPrinted": False, } def extract_items(data): if isinstance(data, list): return data if isinstance(data, dict): if isinstance(data.get("items"), list): return data["items"] for key in ("groups", "accounts", "api_keys", "keys"): if isinstance(data.get(key), list): return data[key] return [] def find_access_token(data): if isinstance(data, dict): for key in ("access_token", "token"): if isinstance(data.get(key), str) and data[key]: return data[key] for value in data.values(): token = find_access_token(value) if token: return token return None def login(): admin_email = get_config_value("sub2api-config", "ADMIN_EMAIL") or POOL_ADMIN_EMAIL_DEFAULT admin_password = decode_secret_value(APP_SECRET_NAME, "ADMIN_PASSWORD") if not admin_password: raise RuntimeError("ADMIN_PASSWORD missing from sub2api-secrets") data = ensure_success(curl_api("POST", "/api/v1/auth/login", payload={"email": admin_email, "password": admin_password}), "admin login") token = find_access_token(data) if not token: raise RuntimeError("admin login response did not contain access_token") compliance = ensure_admin_compliance(token) if compliance.get("ok") is not True: raise RuntimeError("admin compliance acknowledgement failed") return admin_email, token, compliance def group_payload(): return { "name": POOL_GROUP_NAME, "description": POOL_GROUP_DESCRIPTION, "platform": "openai", "rate_multiplier": 1, "is_exclusive": False, "subscription_type": "standard", "allow_messages_dispatch": True, "require_oauth_only": False, "require_privacy_set": False, "rpm_limit": 0, } def list_groups(token): data = ensure_success(curl_api("GET", "/api/v1/admin/groups/all?platform=openai", bearer=token), "list groups") return extract_items(data) def ensure_group(token): existing = next((item for item in list_groups(token) if item.get("name") == POOL_GROUP_NAME), None) payload = group_payload() if existing is None: created = ensure_success(curl_api("POST", "/api/v1/admin/groups", bearer=token, payload=payload), "create group") return created, "created" group_id = existing.get("id") if group_id is None: raise RuntimeError("existing group has no id") payload["status"] = "active" updated = ensure_success(curl_api("PUT", f"/api/v1/admin/groups/{group_id}", bearer=token, payload=payload), "update group") return updated if isinstance(updated, dict) else existing, "updated" def list_accounts_for_group(token, group_id): path = f"/api/v1/admin/accounts?group_id={group_id}&page=1&page_size=500&platform=openai" data = ensure_success(curl_api("GET", path, bearer=token), f"list accounts for group {group_id}") return extract_items(data) def account_group_ids(token, account): if not isinstance(account, dict) or account.get("id") is None: return [] account_id = account.get("id") account_name = account.get("name") ids = [] for group in list_groups(token): group_id = group.get("id") if isinstance(group, dict) else None if group_id is None: continue members = list_accounts_for_group(token, group_id) if any(item.get("id") == account_id or item.get("name") == account_name for item in members if isinstance(item, dict)): ids.append(group_id) return sorted(set(ids)) def list_accounts(token): path = "/api/v1/admin/accounts?page=1&page_size=200&platform=openai&type=apikey&search=" + quote("unidesk-codex-") data = ensure_success(curl_api("GET", path, bearer=token), "list accounts") return extract_items(data) def find_account_by_name(token, name): path = "/api/v1/admin/accounts?page=1&page_size=50&platform=openai&search=" + quote(str(name)) data = ensure_success(curl_api("GET", path, bearer=token), "find account " + str(name)) for item in extract_items(data): if isinstance(item, dict) and item.get("name") == name: return item return None def list_proxy_candidates(token, search=""): path = "/api/v1/admin/proxies?page=1&page_size=200" if search: path += "&search=" + quote(str(search)) data = ensure_success(curl_api("GET", path, bearer=token), "list proxies") return extract_items(data) def find_proxy_by_name(token, name): for item in list_proxy_candidates(token, name): if isinstance(item, dict) and item.get("name") == name: return item return None def manual_binding_plan(binding, expected_kind): if not isinstance(binding, dict): return None plan = binding.get("sourcePlan") if not isinstance(plan, dict): raise RuntimeError("manual account binding is missing resolved sourcePlan") if plan.get("enabled") is not True: raise RuntimeError(f"manual account binding source {plan.get('id')} is disabled") if plan.get("kind") != expected_kind: raise RuntimeError(f"manual account binding source {plan.get('id')} kind must be {expected_kind}") return plan def desired_manual_proxy_payload(protection): binding = protection.get("proxyBinding") if isinstance(protection, dict) else None if not isinstance(binding, dict) or binding.get("enabled") is not True: return None plan = manual_binding_plan(binding, "proxy") if plan.get("provider") != "target-egress-proxy": raise RuntimeError(f"manual account proxyBinding source {plan.get('id')} uses unsupported provider {plan.get('provider')}") target_proxy = plan.get("targetEgressProxy") if not isinstance(target_proxy, dict): raise RuntimeError(f"manual account proxyBinding source {plan.get('id')} requires an enabled target egressProxy") service_name = target_proxy.get("serviceName") listen_port = target_proxy.get("listenPort") namespace = target_proxy.get("namespace") if isinstance(target_proxy.get("namespace"), str) and target_proxy.get("namespace") else NAMESPACE proxy_name = binding.get("proxyName") if not isinstance(service_name, str) or not service_name: raise RuntimeError("target egressProxy serviceName is missing") if not isinstance(listen_port, int) or listen_port <= 0: raise RuntimeError("target egressProxy listenPort is missing") if not isinstance(proxy_name, str) or not proxy_name: raise RuntimeError("manual account proxyBinding proxyName is missing") return { "name": proxy_name, "protocol": "http", "host": f"{service_name}.{namespace}.svc.cluster.local", "port": listen_port, "fallback_mode": "none", "expiry_warn_days": 0, } def proxy_needs_update(proxy, payload): if not isinstance(proxy, dict): return True for key in ("name", "protocol", "host", "port", "fallback_mode", "expiry_warn_days"): if proxy.get(key) != payload.get(key): return True return proxy.get("status") != "active" def ensure_manual_proxy(token, payload): current = find_proxy_by_name(token, payload["name"]) if current is None: created = ensure_success(curl_api("POST", "/api/v1/admin/proxies", bearer=token, payload=payload), f"create proxy {payload['name']}") return created, "created" if proxy_needs_update(current, payload): update_payload = dict(payload) update_payload["status"] = "active" updated = ensure_success(curl_api("PUT", f"/api/v1/admin/proxies/{current['id']}", bearer=token, payload=update_payload), f"update proxy {payload['name']}") return updated if isinstance(updated, dict) else current, "updated" return current, "unchanged" def manual_proxy_status(token, account, protection): payload = desired_manual_proxy_payload(protection) if payload is None: return { "enabled": False, "ok": True, "action": "not-configured", "valuesPrinted": False, } binding = protection.get("proxyBinding") if isinstance(protection, dict) else None plan = manual_binding_plan(binding, "proxy") proxy = find_proxy_by_name(token, payload["name"]) runtime_proxy = account.get("proxy") if isinstance(account, dict) and isinstance(account.get("proxy"), dict) else None binding_aligned = ( isinstance(account, dict) and isinstance(proxy, dict) and account.get("proxy_id") == proxy.get("id") ) return { "enabled": True, "ok": binding_aligned, "action": "validate", "source": plan.get("id"), "sourceProvider": plan.get("provider"), "expectedProxyName": payload["name"], "expectedProtocol": payload["protocol"], "expectedHost": payload["host"], "expectedPort": payload["port"], "proxyRecordExists": isinstance(proxy, dict), "proxyId": proxy.get("id") if isinstance(proxy, dict) else None, "runtimeProxyId": account.get("proxy_id") if isinstance(account, dict) else None, "runtimeProxyName": runtime_proxy.get("name") if isinstance(runtime_proxy, dict) else None, "bindingAligned": binding_aligned, "valuesPrinted": False, } def ensure_manual_account_proxy_bindings(token): items = [] for protection in MANUAL_ACCOUNT_PROTECTIONS: if not isinstance(protection, dict): continue name = protection.get("accountName") if not isinstance(name, str) or not name: continue account = find_account_by_name(token, name) payload = desired_manual_proxy_payload(protection) if payload is None: items.append({ "accountName": name, "enabled": False, "action": "not-configured", "ok": True, "valuesPrinted": False, }) continue if not isinstance(account, dict): items.append({ "accountName": name, "enabled": True, "action": "account-missing", "ok": False, "expectedProxyName": payload["name"], "valuesPrinted": False, }) continue extra = account.get("extra") if isinstance(account.get("extra"), dict) else {} if extra.get("unidesk_managed") is True: items.append({ "accountName": name, "accountId": account.get("id"), "enabled": True, "action": "refused-managed-runtime-account", "ok": False, "expectedProxyName": payload["name"], "valuesPrinted": False, }) continue proxy, proxy_action = ensure_manual_proxy(token, payload) binding = protection.get("proxyBinding") if isinstance(protection, dict) else None plan = manual_binding_plan(binding, "proxy") proxy_id = proxy.get("id") if isinstance(proxy, dict) else None if proxy_id is None: raise RuntimeError(f"proxy {payload['name']} has no id") action = "unchanged" if account.get("proxy_id") != proxy_id: updated = ensure_success(curl_api("PUT", f"/api/v1/admin/accounts/{account['id']}", bearer=token, payload={"proxy_id": proxy_id}), f"bind manual account proxy {name}") account = updated if isinstance(updated, dict) else account action = "bound" runtime_proxy = account.get("proxy") if isinstance(account.get("proxy"), dict) else None items.append({ "accountName": name, "accountId": account.get("id"), "enabled": True, "action": action, "proxyAction": proxy_action, "source": plan.get("id"), "sourceProvider": plan.get("provider"), "ok": account.get("proxy_id") == proxy_id, "expectedProxyName": payload["name"], "expectedProtocol": payload["protocol"], "expectedHost": payload["host"], "expectedPort": payload["port"], "proxyId": proxy_id, "runtimeProxyId": account.get("proxy_id"), "runtimeProxyName": runtime_proxy.get("name") if isinstance(runtime_proxy, dict) else None, "bindingAligned": account.get("proxy_id") == proxy_id, "controlPolicy": "manual-protected: only proxy_id binding is YAML-controlled; credentials/status/schedulable are untouched", "valuesPrinted": False, }) return { "ok": all(item.get("ok") is True for item in items), "itemCount": len(items), "items": items, "valuesPrinted": False, } def manual_group_binding_plan(protection): binding = protection.get("groupBinding") if isinstance(protection, dict) else None if not isinstance(binding, dict) or binding.get("enabled") is not True: return None plan = manual_binding_plan(binding, "group") if plan.get("provider") != "pool-group": raise RuntimeError(f"manual account groupBinding source {plan.get('id')} uses unsupported provider {plan.get('provider')}") return plan def manual_group_binding_enabled(protection): return manual_group_binding_plan(protection) is not None def manual_group_status(token, account, protection, group_id): plan = manual_group_binding_plan(protection) if plan is None: return { "enabled": False, "ok": True, "action": "not-configured", "valuesPrinted": False, } group_accounts = list_accounts_for_group(token, group_id) account_id = account.get("id") if isinstance(account, dict) else None account_name = account.get("name") if isinstance(account, dict) else None binding_aligned = any( item.get("id") == account_id or item.get("name") == account_name for item in group_accounts if isinstance(item, dict) ) return { "enabled": True, "ok": binding_aligned, "action": "validate", "source": plan.get("id"), "sourceProvider": plan.get("provider"), "poolGroupName": POOL_GROUP_NAME, "poolGroupId": group_id, "bindingAligned": binding_aligned, "valuesPrinted": False, } def ensure_manual_account_group_bindings(token, group_id): items = [] for protection in MANUAL_ACCOUNT_PROTECTIONS: if not isinstance(protection, dict): continue name = protection.get("accountName") if not isinstance(name, str) or not name: continue if not manual_group_binding_enabled(protection): items.append({ "accountName": name, "enabled": False, "action": "not-configured", "ok": True, "valuesPrinted": False, }) continue plan = manual_group_binding_plan(protection) account = find_account_by_name(token, name) if not isinstance(account, dict): items.append({ "accountName": name, "enabled": True, "action": "account-missing", "ok": False, "poolGroupName": POOL_GROUP_NAME, "poolGroupId": group_id, "valuesPrinted": False, }) continue extra = account.get("extra") if isinstance(account.get("extra"), dict) else {} if extra.get("unidesk_managed") is True: items.append({ "accountName": name, "accountId": account.get("id"), "enabled": True, "action": "refused-managed-runtime-account", "ok": False, "poolGroupName": POOL_GROUP_NAME, "poolGroupId": group_id, "valuesPrinted": False, }) continue existing_group_ids = account_group_ids(token, account) desired_group_ids = sorted(set(existing_group_ids + [group_id])) action = "unchanged" if group_id not in existing_group_ids: updated = ensure_success(curl_api("PUT", f"/api/v1/admin/accounts/{account['id']}", bearer=token, payload={"group_ids": desired_group_ids}), f"bind manual account group {name}") account = updated if isinstance(updated, dict) else account action = "bound" binding_aligned = any( item.get("id") == account.get("id") or item.get("name") == name for item in list_accounts_for_group(token, group_id) if isinstance(item, dict) ) items.append({ "accountName": name, "accountId": account.get("id"), "enabled": True, "ok": binding_aligned, "action": action, "source": plan.get("id") if isinstance(plan, dict) else None, "sourceProvider": plan.get("provider") if isinstance(plan, dict) else None, "poolGroupName": POOL_GROUP_NAME, "poolGroupId": group_id, "previousGroupIds": existing_group_ids, "desiredGroupIds": desired_group_ids, "bindingAligned": binding_aligned, "controlPolicy": "manual-protected: only pool group membership is YAML-controlled; credentials/status/schedulable are untouched and sentinel does not probe it", "valuesPrinted": False, }) return { "ok": all(item.get("ok") is True for item in items), "itemCount": len(items), "items": items, "valuesPrinted": False, } def manual_account_protection_status(token, group_id=None): items = [] desired_names = set(EXPECTED_ACCOUNT_CAPACITIES.keys()) for protection in MANUAL_ACCOUNT_PROTECTIONS: if not isinstance(protection, dict): continue name = protection.get("accountName") if not isinstance(name, str) or not name: continue account = find_account_by_name(token, name) extra = account.get("extra") if isinstance(account, dict) and isinstance(account.get("extra"), dict) else {} proxy_status = manual_proxy_status(token, account, protection) group_status = manual_group_status(token, account, protection, group_id) if group_id is not None else {"enabled": False, "ok": True, "action": "not-checked", "valuesPrinted": False} items.append({ "accountName": name, "reason": protection.get("reason") if isinstance(protection.get("reason"), str) else None, "exists": isinstance(account, dict), "accountId": account.get("id") if isinstance(account, dict) else None, "status": account.get("status") if isinstance(account, dict) else None, "schedulable": account.get("schedulable") if isinstance(account, dict) else None, "inYamlProfiles": name in desired_names, "runtimeMarkedUnideskManaged": extra.get("unidesk_managed") is True, "proxyBinding": proxy_status, "groupBinding": group_status, "ok": proxy_status.get("ok") is True and group_status.get("ok") is True, "controlPolicy": "manual-protected: no create/update/prune/probe/freeze; optional proxy_id and pool group membership binding only when configured", "valuesPrinted": False, }) return { "ok": all(item.get("ok") is True for item in items), "protectedCount": len(items), "items": items, "valuesPrinted": False, } def list_probe_accounts(token): path = "/api/v1/admin/accounts?page=1&page_size=200&platform=openai&type=apikey&search=" + quote("unidesk-probe-") data = ensure_success(curl_api("GET", path, bearer=token), "list probe accounts") return [item for item in extract_items(data) if isinstance(item.get("name"), str) and item.get("name").startswith("unidesk-probe-")] def list_probe_groups(token): data = ensure_success(curl_api("GET", "/api/v1/admin/groups/all?platform=openai", bearer=token), "list groups") return [item for item in extract_items(data) if isinstance(item.get("name"), str) and item.get("name").startswith("unidesk-probe-")] def cleanup_probe_resources(token): api_key_results = [] account_results = [] group_results = [] for item in list_user_keys(token): name = item.get("name") key_id = item.get("id") if not isinstance(name, str) or not name.startswith("unidesk-probe-") or key_id is None: continue api_key_results.append(delete_probe_resource(token, "DELETE", f"/api/v1/keys/{key_id}", "api-key")) for item in list_probe_accounts(token): account_id = item.get("id") if account_id is None: continue account_results.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/accounts/{account_id}", "account")) for item in list_probe_groups(token): group_id = item.get("id") if group_id is None: continue group_results.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/groups/{group_id}", "group")) return { "ok": all(item.get("ok") is True for item in api_key_results + account_results + group_results), "apiKeysDeleted": len(api_key_results), "accountsDeleted": len(account_results), "groupsDeleted": len(group_results), "items": { "apiKeys": api_key_results, "accounts": account_results, "groups": group_results, }, "valuesPrinted": False, } def temp_unschedulable_credentials(profile): credentials = profile.get("tempUnschedulableCredentials") if not isinstance(credentials, dict): credentials = {} return normalize_temp_unschedulable_credentials(credentials) def account_payload(profile, group_id): extra = { "openai_responses_mode": "force_responses", "unidesk_codex_profile": profile["profile"], "unidesk_managed": True, "unidesk_trust_upstream": profile.get("trustUpstream") is True, "unidesk_sentinel_protect": profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False}, } ws_mode = profile.get("openaiResponsesWebSocketsV2Mode") if ws_mode: extra["openai_apikey_responses_websockets_v2_mode"] = ws_mode extra["openai_apikey_responses_websockets_v2_enabled"] = ws_mode != "off" credentials = { "api_key": profile["apiKey"], "base_url": profile["baseUrl"], } upstream_user_agent = profile.get("upstreamUserAgent") if upstream_user_agent: credentials["user_agent"] = upstream_user_agent temp_unschedulable = temp_unschedulable_credentials(profile) if temp_unschedulable["enabled"]: credentials["temp_unschedulable_enabled"] = True credentials["temp_unschedulable_rules"] = temp_unschedulable["rules"] return { "name": profile["accountName"], "notes": f"UniDesk-managed Codex profile {profile['profile']} from {profile['configFile']} and {profile['authFile']}; secret source={profile['apiKeySource']}; fingerprint={profile['apiKeyFingerprint']}.", "platform": "openai", "type": "apikey", "credentials": credentials, "extra": extra, "concurrency": int(profile.get("capacity", 5) or 5), "priority": int(profile.get("priority", POOL_DEFAULT_ACCOUNT_PRIORITY) or POOL_DEFAULT_ACCOUNT_PRIORITY), "rate_multiplier": 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, } def planned_sentinel_account_results(profiles, existing_accounts): existing = {item.get("name"): item for item in existing_accounts if isinstance(item, dict)} results = [] for profile in profiles: change_reasons = sentinel_probe_change_reasons(existing.get(profile["accountName"]), profile) quality_gate_required = sentinel_quality_gate_enabled() and len(change_reasons) > 0 results.append({ "profile": profile["profile"], "accountName": profile["accountName"], "profileConfig": { "accountName": profile["accountName"], "trustUpstream": profile.get("trustUpstream") is True, "sentinelProtect": profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False}, }, "sentinelProbeConfigFingerprint": profile.get("sentinelProbeConfigFingerprint"), "sentinelProbeRequired": quality_gate_required, "sentinelChangeReasons": change_reasons if quality_gate_required else [], "sentinelProbePending": quality_gate_required, "sentinelDefaultFrozen": False, "valuesPrinted": False, }) return results def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_frozen_names=None, existing_accounts=None): if not isinstance(protected_frozen_names, set): protected_frozen_names = set() if not isinstance(existing_accounts, list): existing_accounts = list_accounts(token) existing = {item.get("name"): item for item in existing_accounts} desired_names = {profile["accountName"] for profile in profiles} results = [] for profile in profiles: payload = account_payload(profile, group_id) current = existing.get(profile["accountName"]) change_reasons = sentinel_probe_change_reasons(current, profile) quality_gate_required = sentinel_quality_gate_enabled() and len(change_reasons) > 0 keep_frozen = profile["accountName"] in protected_frozen_names if current and current.get("id") is not None: account_id = current["id"] update_payload = dict(payload) update_payload.pop("platform", None) update_payload["status"] = "active" data = ensure_success(curl_api("PUT", f"/api/v1/admin/accounts/{account_id}", bearer=token, payload=update_payload), f"update account {profile['accountName']}") action = "updated" else: data = ensure_success(curl_api("POST", "/api/v1/admin/accounts", bearer=token, payload=payload), f"create account {profile['accountName']}") action = "created" results.append({ "profile": profile["profile"], "accountName": profile["accountName"], "profileConfig": { "accountName": profile["accountName"], "trustUpstream": profile.get("trustUpstream") is True, "sentinelProtect": profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False}, }, "accountId": data.get("id") if isinstance(data, dict) else None, "action": action, "baseUrl": profile["baseUrl"], "apiKeySource": profile["apiKeySource"], "apiKeyFingerprint": profile["apiKeyFingerprint"], "sentinelProbeConfigFingerprint": profile.get("sentinelProbeConfigFingerprint"), "sentinelProbeRequired": quality_gate_required, "sentinelChangeReasons": change_reasons if quality_gate_required else [], "sentinelProbePending": quality_gate_required, "sentinelDefaultFrozen": False, "sentinelFreezeProtected": keep_frozen, "schedulableControl": "sentinel-marker-restore", "openaiResponsesWebSocketsV2Mode": profile.get("openaiResponsesWebSocketsV2Mode"), "trustUpstream": profile.get("trustUpstream") is True, "priority": int(profile.get("priority", POOL_DEFAULT_ACCOUNT_PRIORITY) or POOL_DEFAULT_ACCOUNT_PRIORITY), "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, "runtimeSchedulable": data.get("schedulable") 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")), "valuesPrinted": False, }) prune_results = prune_removed_accounts(token, existing_accounts, desired_names) if prune_removed else [] return results, prune_results def prune_removed_accounts(token, existing_accounts, desired_names): results = [] for account in existing_accounts: name = account.get("name") account_id = account.get("id") extra = account.get("extra") if isinstance(account.get("extra"), dict) else {} if not isinstance(name, str) or not name.startswith("unidesk-codex-"): continue if extra.get("unidesk_managed") is not True: continue if name in desired_names: continue if account_id is None: raise RuntimeError(f"removed account {name} has no id") ensure_success(curl_api("DELETE", f"/api/v1/admin/accounts/{account_id}", bearer=token), f"delete removed account {name}") results.append({ "accountName": name, "accountId": account_id, "profile": extra.get("unidesk_codex_profile"), "action": "deleted", "reason": "removed-from-yaml", "valuesPrinted": False, }) return results def generate_api_key(): alphabet = string.ascii_letters + string.digits return "sk-unidesk-codex-" + "".join(secrets.choice(alphabet) for _ in range(48)) def ensure_api_key_secret(group_id): existing = None try: existing = decode_secret_value(POOL_API_KEY_SECRET_NAME, POOL_API_KEY_SECRET_KEY) except Exception: existing = None api_key = existing if existing else generate_api_key() secret_action = "kept-existing" if existing else "created" manifest = { "apiVersion": "v1", "kind": "Secret", "metadata": { "name": POOL_API_KEY_SECRET_NAME, "namespace": NAMESPACE, "labels": { "app.kubernetes.io/name": "sub2api", "app.kubernetes.io/part-of": "platform-infra", "app.kubernetes.io/managed-by": "unidesk", "unidesk.ai/secret-purpose": "sub2api-codex-pool-api-key", }, }, "type": "Opaque", "stringData": { POOL_API_KEY_SECRET_KEY: api_key, "GROUP_ID": str(group_id), "GROUP_NAME": POOL_GROUP_NAME, "SERVICE_DNS": SERVICE_DNS, }, } proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], json.dumps(manifest)) if proc.returncode != 0: raise RuntimeError(f"apply API key secret failed: {text(proc.stderr, 1000)}") return api_key, secret_action, text(proc.stdout, 1000) def apply_sentinel_manifest(manifest): if not isinstance(manifest, str) or not manifest.strip(): return { "ok": False, "action": "missing-manifest", "valuesPrinted": False, } proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], manifest) if proc.returncode != 0: return { "ok": False, "action": "apply-failed", "stdoutTail": text(proc.stdout, 2000), "stderrTail": text(proc.stderr, 4000), "valuesPrinted": False, } status = sentinel_runtime_status() return { "ok": status.get("ok") is True, "action": "applied", "stdoutTail": text(proc.stdout, 2000), "runtime": status, "valuesPrinted": False, } def safe_kube_json(args, label): proc = kubectl([*args, "-o", "json"]) if proc.returncode != 0: return None, {"label": label, "exitCode": proc.returncode, "stderrTail": text(proc.stderr, 1000)} try: return json.loads(proc.stdout.decode("utf-8")), None except Exception as exc: return None, {"label": label, "exitCode": proc.returncode, "error": str(exc), "stdoutTail": text(proc.stdout, 1000)} def sentinel_runtime_status(): cfg = SENTINEL_CONFIG cronjob_name = cfg.get("cronJobName") secret_name = cfg.get("credentialsSecretName") configmap_name = cfg.get("configMapName") state_name = cfg.get("stateConfigMapName") cronjob, cronjob_error = safe_kube_json(["-n", NAMESPACE, "get", "cronjob", cronjob_name], f"cronjob/{cronjob_name}") secret, secret_error = safe_kube_json(["-n", NAMESPACE, "get", "secret", secret_name], f"secret/{secret_name}") configmap, configmap_error = safe_kube_json(["-n", NAMESPACE, "get", "configmap", configmap_name], f"configmap/{configmap_name}") state_cm, state_error = safe_kube_json(["-n", NAMESPACE, "get", "configmap", state_name], f"configmap/{state_name}") state = None if isinstance(state_cm, dict): raw_state = (state_cm.get("data") or {}).get("state.json") if isinstance(raw_state, str) and raw_state: try: state = json.loads(raw_state) except Exception as exc: state = {"parseError": str(exc)} accounts = (state.get("accounts") or {}) if isinstance(state, dict) else {} quarantined = [] recent_accounts = [] for name, account_state in accounts.items(): if not isinstance(account_state, dict): continue quarantine = account_state.get("quarantine") if isinstance(quarantine, dict) and quarantine.get("active") is True: quarantined.append({ "accountName": name, "until": quarantine.get("until"), "applied": quarantine.get("applied"), "reason": quarantine.get("reason"), "failureKind": quarantine.get("failureKind"), "errorDetails": quarantine.get("errorDetails"), "intervalMinutes": quarantine.get("intervalMinutes"), "sentinelProtect": quarantine.get("sentinelProtect"), }) last_probe = account_state.get("lastProbe") if isinstance(last_probe, dict): recent_accounts.append({ "accountName": name, "lastProbeAt": account_state.get("lastProbeAt"), "lastStatus": account_state.get("lastStatus"), "nextProbeAfter": account_state.get("nextProbeAfter"), "ok": last_probe.get("ok"), "purpose": last_probe.get("purpose"), "httpStatus": last_probe.get("httpStatus"), "durationMs": last_probe.get("durationMs"), "markerMatched": last_probe.get("markerMatched"), "sentinelProtect": last_probe.get("sentinelProtect") or account_state.get("sentinelProtectConfig"), "outputHash": last_probe.get("outputHash"), "outputPreview": last_probe.get("outputPreview"), "responseBodyHash": last_probe.get("responseBodyHash"), "responseBodyPreview": last_probe.get("responseBodyPreview"), "error": last_probe.get("error"), "errorDetails": last_probe.get("errorDetails"), "usage": last_probe.get("usage"), "failureKind": last_probe.get("failureKind"), "requestShape": last_probe.get("requestShape"), "action": last_probe.get("action"), }) recent_accounts.sort(key=lambda item: item.get("lastProbeAt") or "") last_run = state.get("lastRun") if isinstance(state, dict) else None cronjob_spec = cronjob.get("spec") if isinstance(cronjob, dict) else {} secret_data = secret.get("data") if isinstance(secret, dict) else {} configmap_data = configmap.get("data") if isinstance(configmap, dict) else {} ok = cronjob is not None and secret is not None and configmap is not None return { "ok": ok, "desired": { "monitorEnabled": cfg.get("monitor", {}).get("enabled"), "actionsEnabled": cfg.get("actions", {}).get("enabled"), "schedule": cfg.get("schedule"), "cronJobName": cronjob_name, "configMapName": configmap_name, "credentialsSecretName": secret_name, "stateConfigMapName": state_name, }, "cronJob": { "exists": cronjob is not None, "schedule": cronjob_spec.get("schedule") if isinstance(cronjob_spec, dict) else None, "suspend": cronjob_spec.get("suspend") if isinstance(cronjob_spec, dict) else None, "lastScheduleTime": (cronjob.get("status") or {}).get("lastScheduleTime") if isinstance(cronjob, dict) else None, "active": len((cronjob.get("status") or {}).get("active") or []) if isinstance(cronjob, dict) else None, "error": cronjob_error, }, "secret": { "exists": secret is not None, "profileSecretPresent": isinstance(secret_data, dict) and "profiles.json" in secret_data, "valuesPrinted": False, "error": secret_error, }, "configMap": { "exists": configmap is not None, "configPresent": isinstance(configmap_data, dict) and "config.json" in configmap_data, "runnerPresent": isinstance(configmap_data, dict) and "sentinel.py" in configmap_data, "error": configmap_error, }, "state": { "exists": state_cm is not None, "accountCount": len(accounts), "quarantinedCount": len(quarantined), "quarantined": quarantined[-10:], "recentAccounts": recent_accounts[-12:], "lastRun": last_run, "error": state_error, }, "valuesPrinted": False, } def parse_epoch_z(value): if not isinstance(value, str) or not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() except Exception: return None def sentinel_state_object(): state_name = SENTINEL_CONFIG.get("stateConfigMapName") if not state_name: return None, None obj, err = safe_kube_json(["-n", NAMESPACE, "get", "configmap", state_name], f"configmap/{state_name}") if not isinstance(obj, dict): return None, None raw_state = (obj.get("data") or {}).get("state.json") if not isinstance(raw_state, str) or not raw_state: return obj, None try: return obj, json.loads(raw_state) except Exception: return obj, None def active_sentinel_quarantine_names(): _, state = sentinel_state_object() if not isinstance(state, dict): return set() accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {} names = set() for name, account_state in accounts_state.items(): if not isinstance(name, str) or not isinstance(account_state, dict): continue quarantine = account_state.get("quarantine") if isinstance(quarantine, dict) and quarantine.get("active") is True and quarantine.get("applied") is True: names.add(name) return names def default_sentinel_state(): return {"version": 1, "accounts": {}, "ledger": {}, "history": []} def clamp_sentinel_freezes_for_config(state, now): freeze_config = SENTINEL_CONFIG.get("freeze") if isinstance(SENTINEL_CONFIG.get("freeze"), dict) else {} try: max_interval = int(freeze_config.get("maxTtlMinutes") or 10) except Exception: max_interval = 10 accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {} now_epoch = time.time() items = [] for name, account_state in accounts_state.items(): if not isinstance(name, str) or not isinstance(account_state, dict): continue quarantine = account_state.get("quarantine") if not isinstance(quarantine, dict) or quarantine.get("active") is not True or quarantine.get("applied") is not True: continue try: interval = int(quarantine.get("intervalMinutes") or 0) except Exception: interval = 0 until_epoch = parse_epoch_z(quarantine.get("until")) old_until = quarantine.get("until") if interval <= max_interval and (until_epoch is None or until_epoch <= now_epoch + max_interval * 60): continue quarantine["previousIntervalMinutes"] = interval quarantine["intervalMinutes"] = max_interval quarantine["until"] = now quarantine["clampedAt"] = now quarantine["clampedBy"] = "sync-freeze-max-ttl" account_state["nextProbeAfter"] = now items.append({ "accountName": name, "previousIntervalMinutes": interval, "maxIntervalMinutes": max_interval, "previousUntil": old_until, "nextProbeAfter": now, }) return items def parse_iso_epoch(value): if not isinstance(value, str) or not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() except Exception: return None def profile_success_max_interval(profile): cadence = SENTINEL_CONFIG.get("cadence") if isinstance(SENTINEL_CONFIG.get("cadence"), dict) else {} legacy = cadence.get("successMaxIntervalMinutes") if legacy is None: legacy = cadence.get("trustedSuccessMaxIntervalMinutes") or cadence.get("untrustedSuccessMaxIntervalMinutes") or 1 if profile.get("trustUpstream") is True: value = cadence.get("trustedSuccessMaxIntervalMinutes") or legacy else: value = cadence.get("untrustedSuccessMaxIntervalMinutes") or legacy try: return int(value) except Exception: return int(legacy) def clamp_sentinel_success_cadence_for_config(state, profiles, now): accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {} profile_map = {item.get("accountName"): item for item in profiles if isinstance(item, dict) and isinstance(item.get("accountName"), str)} now_epoch = time.time() items = [] for name, profile in profile_map.items(): account_state = accounts_state.get(name) if not isinstance(account_state, dict): continue quarantine = account_state.get("quarantine") if isinstance(quarantine, dict) and quarantine.get("active") is True: account_state["trustUpstream"] = profile.get("trustUpstream") is True account_state["sentinelProtectConfig"] = profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False} account_state["successMaxIntervalMinutes"] = profile_success_max_interval(profile) continue try: interval = int(account_state.get("successIntervalMinutes") or 0) except Exception: interval = 0 next_epoch = parse_iso_epoch(account_state.get("nextProbeAfter")) max_interval = profile_success_max_interval(profile) account_state["trustUpstream"] = profile.get("trustUpstream") is True account_state["sentinelProtectConfig"] = profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False} account_state["successMaxIntervalMinutes"] = max_interval if interval <= max_interval and (next_epoch is None or next_epoch <= now_epoch + max_interval * 60): continue old_next = account_state.get("nextProbeAfter") account_state["previousSuccessIntervalMinutes"] = interval account_state["successIntervalMinutes"] = min(interval, max_interval) if interval > 0 else interval account_state["nextProbeAfter"] = now account_state["cadenceClampedAt"] = now account_state["cadenceClampedBy"] = "sync-success-max-interval" items.append({ "accountName": name, "trustUpstream": profile.get("trustUpstream") is True, "previousSuccessIntervalMinutes": interval, "maxIntervalMinutes": max_interval, "previousNextProbeAfter": old_next, "nextProbeAfter": now, }) return items def update_sentinel_state_configmap(obj, state): state_name = SENTINEL_CONFIG.get("stateConfigMapName") if not state_name: return {"ok": False, "reason": "state-configmap-missing"} state_json = json.dumps(state, ensure_ascii=False, indent=2) manifest = { "apiVersion": "v1", "kind": "ConfigMap", "metadata": { "name": state_name, "namespace": NAMESPACE, "labels": { "app.kubernetes.io/name": SERVICE_NAME, "app.kubernetes.io/component": "account-sentinel", "app.kubernetes.io/managed-by": "unidesk-platform-infra", }, }, "data": {"state.json": state_json}, } proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], json.dumps(manifest)) action = "applied" if isinstance(obj, dict) else "created" if proc.returncode != 0: return {"ok": False, "reason": f"{action}-failed", "stderrTail": text(proc.stderr, 2000)} return {"ok": True, "action": action} def ensure_sentinel_state_for_sync(account_results, pending_only=False): if not sentinel_quality_gate_enabled(): return {"ok": True, "skipped": True, "reason": "sentinel-quality-gate-disabled", "changedCount": 0, "items": [], "valuesPrinted": False} state_obj, state = sentinel_state_object() if not isinstance(state, dict): state = default_sentinel_state() state.setdefault("version", 1) accounts_state = state.setdefault("accounts", {}) if not isinstance(accounts_state, dict): accounts_state = {} state["accounts"] = accounts_state now = utc_iso() items = [] clamped_items = [] if pending_only else clamp_sentinel_freezes_for_config(state, now) cadence_clamped_items = [] if pending_only else clamp_sentinel_success_cadence_for_config(state, [item.get("profileConfig") for item in account_results if isinstance(item.get("profileConfig"), dict)], now) changed_count = 0 fingerprint_only_count = 0 for item in account_results: name = item.get("accountName") if not isinstance(name, str) or not name: continue account_state = accounts_state.setdefault(name, {}) if not isinstance(account_state, dict): account_state = {} accounts_state[name] = account_state fingerprint_value = item.get("sentinelProbeConfigFingerprint") if isinstance(fingerprint_value, str) and fingerprint_value: account_state["probeConfigFingerprint"] = fingerprint_value if item.get("sentinelProbeRequired") is not True: fingerprint_only_count += 1 continue changed_count += 1 reasons = item.get("sentinelChangeReasons") if isinstance(item.get("sentinelChangeReasons"), list) else [] quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else None if not (isinstance(quarantine, dict) and quarantine.get("active") is True): account_state["quarantine"] = { "active": False, "reason": "yaml-account-change-pending-sentinel-probe", "lastPendingAt": now, "changeReasons": reasons, } account_state["nextProbeAfter"] = now account_state["successStreak"] = 0 account_state["successIntervalMinutes"] = 0 profile_config = item.get("profileConfig") if isinstance(item.get("profileConfig"), dict) else {} account_state["trustUpstream"] = profile_config.get("trustUpstream") is True account_state["sentinelProtectConfig"] = profile_config.get("sentinelProtect") if isinstance(profile_config.get("sentinelProtect"), dict) else {"enabled": False} account_state["successMaxIntervalMinutes"] = profile_success_max_interval(profile_config) account_state["lastStatus"] = "pending-sentinel-quality-gate" account_state["qualityGate"] = { "pending": True, "reason": "yaml-account-change", "changeReasons": reasons, "markedAt": now, "pendingOnly": pending_only, "defaultFrozen": False, } items.append({"accountName": name, "changeReasons": reasons, "nextProbeAfter": now, "defaultFrozen": False, "defaultSchedulable": True, "pendingOnly": pending_only}) if changed_count <= 0 and len(clamped_items) <= 0 and len(cadence_clamped_items) <= 0: return {"ok": True, "skipped": False, "reason": "no-new-or-changed-accounts", "changedCount": 0, "fingerprintOnlyCount": fingerprint_only_count, "clampedCount": 0, "cadenceClampedCount": 0, "items": [], "valuesPrinted": False} update = update_sentinel_state_configmap(state_obj, state) if pending_only and changed_count > 0: reason = "new-or-changed-accounts-pending-probe-prepared-default-schedulable" elif changed_count > 0 and (len(clamped_items) > 0 or len(cadence_clamped_items) > 0): reason = "new-or-changed-accounts-default-schedulable-and-sentinel-cadence-clamped" elif changed_count > 0: reason = "new-or-changed-accounts-default-schedulable" elif len(cadence_clamped_items) > 0: reason = "success-cadence-clamped-to-current-config" else: reason = "freeze-backoff-clamped-to-current-config" return { "ok": update.get("ok") is True, "skipped": False, "reason": reason, "changedCount": changed_count, "fingerprintOnlyCount": fingerprint_only_count, "clampedCount": len(clamped_items), "cadenceClampedCount": len(cadence_clamped_items), "pendingOnly": pending_only, "items": items, "clampedItems": clamped_items, "cadenceClampedItems": cadence_clamped_items, "update": update, "valuesPrinted": False, } def sentinel_state_summary(): _, state = sentinel_state_object() if not isinstance(state, dict): return {"exists": False, "valuesPrinted": False} accounts = state.get("accounts") if isinstance(state.get("accounts"), dict) else {} quarantined = [] recent_accounts = [] for name, account_state in accounts.items(): if not isinstance(account_state, dict): continue quarantine = account_state.get("quarantine") if isinstance(quarantine, dict) and quarantine.get("active") is True: quarantined.append({ "accountName": name, "until": quarantine.get("until"), "applied": quarantine.get("applied"), "reason": quarantine.get("reason"), "failureKind": quarantine.get("failureKind"), "errorDetails": quarantine.get("errorDetails"), "intervalMinutes": quarantine.get("intervalMinutes"), "sentinelProtect": quarantine.get("sentinelProtect"), }) last_probe = account_state.get("lastProbe") if isinstance(last_probe, dict): recent_accounts.append({ "accountName": name, "lastProbeAt": account_state.get("lastProbeAt"), "lastStatus": account_state.get("lastStatus"), "nextProbeAfter": account_state.get("nextProbeAfter"), "ok": last_probe.get("ok"), "purpose": last_probe.get("purpose"), "httpStatus": last_probe.get("httpStatus"), "durationMs": last_probe.get("durationMs"), "markerMatched": last_probe.get("markerMatched"), "sentinelProtect": last_probe.get("sentinelProtect") or account_state.get("sentinelProtectConfig"), "usage": last_probe.get("usage"), "responseBodyHash": last_probe.get("responseBodyHash"), "responseBodyPreview": last_probe.get("responseBodyPreview"), "error": last_probe.get("error"), "errorDetails": last_probe.get("errorDetails"), "failureKind": last_probe.get("failureKind"), "requestShape": last_probe.get("requestShape"), "action": last_probe.get("action"), }) recent_accounts.sort(key=lambda item: item.get("lastProbeAt") or "") return { "exists": True, "accountCount": len(accounts), "quarantinedCount": len(quarantined), "quarantined": quarantined[-10:], "recentAccounts": recent_accounts[-12:], "lastRun": state.get("lastRun"), "valuesPrinted": False, } def reassert_sentinel_freezes_after_sync(token): if (SENTINEL_CONFIG.get("actions") or {}).get("enabled") is not True: return {"ok": True, "skipped": True, "reason": "actions-disabled", "items": [], "valuesPrinted": False} _, state = sentinel_state_object() if not isinstance(state, dict): return {"ok": True, "skipped": True, "reason": "state-missing", "items": [], "valuesPrinted": False} accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {} accounts = list_accounts(token) by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)} items = [] for name, account_state in accounts_state.items(): if not isinstance(account_state, dict): continue quarantine = account_state.get("quarantine") if not isinstance(quarantine, dict) or quarantine.get("active") is not True or quarantine.get("applied") is not True: continue account = by_name.get(name) if not account or account.get("id") is None: items.append({"accountName": name, "ok": False, "reason": "account-not-found"}) continue try: ensure_success( curl_api("POST", f"/api/v1/admin/accounts/{account['id']}/schedulable", bearer=token, payload={"schedulable": False}), f"reassert sentinel freeze for {name}", ) items.append({"accountName": name, "accountId": account.get("id"), "ok": True, "until": quarantine.get("until")}) except Exception as exc: items.append({"accountName": name, "accountId": account.get("id"), "ok": False, "error": str(exc)}) return { "ok": all(item.get("ok") is True for item in items), "skipped": False, "items": items, "valuesPrinted": False, } def list_user_keys(token): data = ensure_success(curl_api("GET", "/api/v1/keys?page=1&page_size=200", bearer=token), "list user keys") return extract_items(data) def ensure_sub2api_api_key(token, api_key, group_id): keys = list_user_keys(token) existing = next((item for item in keys if item.get("key") == api_key), None) action = "kept-existing" if existing is None: existing = next((item for item in keys if item.get("name") == POOL_API_KEY_NAME), None) if existing is None or existing.get("key") != api_key: payload = { "name": POOL_API_KEY_NAME, "group_id": group_id, "custom_key": api_key, "quota": 0, "rate_limit_5h": 0, "rate_limit_1d": 0, "rate_limit_7d": 0, } created = ensure_success(curl_api("POST", "/api/v1/keys", bearer=token, payload=payload), "create pool API key") existing = created if isinstance(created, dict) else existing action = "created" elif existing.get("id") is not None and existing.get("group_id") != group_id: updated = ensure_success(curl_api("PUT", f"/api/v1/keys/{existing['id']}", bearer=token, payload={"name": POOL_API_KEY_NAME, "group_id": group_id}), "update pool API key group") existing = updated if isinstance(updated, dict) else existing action = "updated-group" return { "action": action, "id": existing.get("id") if isinstance(existing, dict) else None, "name": existing.get("name") if isinstance(existing, dict) else POOL_API_KEY_NAME, "groupId": existing.get("group_id") if isinstance(existing, dict) else group_id, "userId": existing.get("user_id") if isinstance(existing, dict) else None, } def get_admin_user(token, user_id): data = ensure_success(curl_api("GET", f"/api/v1/admin/users/{user_id}", bearer=token), "get API key owner") if not isinstance(data, dict): raise RuntimeError("API key owner response is not an object") return data def ensure_pool_owner_balance(token, user_id): user = get_admin_user(token, user_id) current = float(user.get("balance") or 0) if current >= MIN_OWNER_BALANCE_USD: return { "action": "kept-existing", "userId": user_id, "balanceBefore": current, "balanceAfter": current, "minimumBalanceUsd": MIN_OWNER_BALANCE_USD, } updated = ensure_success(curl_api("POST", f"/api/v1/admin/users/{user_id}/balance", bearer=token, payload={ "balance": MIN_OWNER_BALANCE_USD, "operation": "set", "notes": "UniDesk Sub2API Codex pool internal API key bootstrap balance.", }), "set API key owner balance") after = float(updated.get("balance") or MIN_OWNER_BALANCE_USD) if isinstance(updated, dict) else MIN_OWNER_BALANCE_USD return { "action": "set", "userId": user_id, "balanceBefore": current, "balanceAfter": after, "minimumBalanceUsd": MIN_OWNER_BALANCE_USD, } def ensure_pool_owner_concurrency(token, user_id): user = get_admin_user(token, user_id) try: current = int(user.get("concurrency") or 0) except Exception: current = 0 if current >= MIN_OWNER_CONCURRENCY: return { "ok": True, "action": "kept-existing", "userId": user_id, "concurrencyBefore": current, "concurrencyAfter": current, "minimumConcurrency": MIN_OWNER_CONCURRENCY, "minimumConcurrencySource": MIN_OWNER_CONCURRENCY_SOURCE, } updated = ensure_success(curl_api("PUT", f"/api/v1/admin/users/{user_id}", bearer=token, payload={ "concurrency": MIN_OWNER_CONCURRENCY, }), "set API key owner concurrency") try: after = int(updated.get("concurrency") or MIN_OWNER_CONCURRENCY) if isinstance(updated, dict) else MIN_OWNER_CONCURRENCY except Exception: after = MIN_OWNER_CONCURRENCY return { "ok": after >= MIN_OWNER_CONCURRENCY, "action": "set", "userId": user_id, "concurrencyBefore": current, "concurrencyAfter": after, "minimumConcurrency": MIN_OWNER_CONCURRENCY, "minimumConcurrencySource": MIN_OWNER_CONCURRENCY_SOURCE, } 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 = 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 recent_compact_gateway_evidence(): proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", "--since=6h", "--tail=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 = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", "--since=6h", "--tail=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" 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 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(): 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 run_sync(): global MANUAL_ACCOUNT_PROTECTIONS payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) manual_accounts_payload = payload.get("manualAccounts") if isinstance(payload.get("manualAccounts"), dict) else {} resolved_manual_protections = manual_accounts_payload.get("protected") if not isinstance(resolved_manual_protections, list): raise RuntimeError("sync payload has no manualAccounts.protected binding plan") MANUAL_ACCOUNT_PROTECTIONS = resolved_manual_protections profiles = payload.get("profiles") or [] prune_removed = bool(payload.get("pruneRemoved")) sentinel_payload = payload.get("sentinel") if isinstance(payload.get("sentinel"), dict) else {} if not profiles: raise RuntimeError("sync payload has no profiles") admin_email, token, admin_compliance = login() group, group_action = ensure_group(token) group_id = group.get("id") if isinstance(group, dict) else None if group_id is None: raise RuntimeError("pool group id missing after ensure") existing_accounts = list_accounts(token) planned_account_results = planned_sentinel_account_results(profiles, existing_accounts) sentinel_quality_prepare = ensure_sentinel_state_for_sync(planned_account_results, True) if sentinel_quality_prepare.get("ok") is not True: raise RuntimeError("prepare sentinel pending probe failed: " + json.dumps(sentinel_quality_prepare, ensure_ascii=False)) protected_frozen_names = active_sentinel_quarantine_names() account_results, pruned_account_results = ensure_accounts(token, profiles, group_id, prune_removed, protected_frozen_names, existing_accounts) manual_account_proxy_bindings = ensure_manual_account_proxy_bindings(token) manual_account_group_bindings = ensure_manual_account_group_bindings(token, group_id) manual_account_protections = manual_account_protection_status(token, 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) api_key_result = ensure_sub2api_api_key(token, api_key, group_id) 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) compact_evidence = recent_compact_gateway_evidence() responses_evidence = recent_responses_gateway_evidence() runtime_capabilities = validate_runtime_capabilities(token) sentinel = apply_sentinel_manifest(sentinel_payload.get("manifest")) sentinel_quality = ensure_sentinel_state_for_sync(account_results) sentinel_reassert = reassert_sentinel_freezes_after_sync(token) 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 load_factor_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True and manual_account_proxy_bindings.get("ok") is True and manual_account_group_bindings.get("ok") is True and manual_account_protections.get("ok") is True and sentinel.get("ok") is True and sentinel_quality_prepare.get("ok") is True and sentinel_quality.get("ok") is True and sentinel_reassert.get("ok") is True and runtime_capabilities.get("ok") is True, "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")) or bool(responses_evidence.get("degraded")) or runtime_capabilities.get("ok") is not True, "mode": "sync", "namespace": NAMESPACE, "serviceDns": SERVICE_DNS, "appPod": APP_POD, "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance}, "pool": {"name": POOL_GROUP_NAME, "id": group_id, "action": group_action, "platform": group.get("platform") if isinstance(group, dict) else "openai"}, "accounts": { "desired": len(profiles), "created": sum(1 for item in account_results if item["action"] == "created"), "updated": sum(1 for item in account_results if item["action"] == "updated"), "pruned": len(pruned_account_results), "pruneMode": "explicit" if prune_removed else "disabled-by-default", "items": account_results, "prunedItems": pruned_account_results, "processControl": {"schedulableRestore": "sentinel marker probe only; sync does not restore schedulable for existing accounts", "durableConfig": False}, "valuesPrinted": False, }, "manualAccounts": {**manual_account_protections, "proxySync": manual_account_proxy_bindings, "groupSync": manual_account_group_bindings}, "capacity": capacity_status, "loadFactor": load_factor_status, "webSocketsV2": ws_v2_status, "tempUnschedulable": temp_unschedulable_status, "apiKey": { "name": POOL_API_KEY_NAME, "secret": f"{NAMESPACE}/{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY}", "secretAction": secret_action, "secretApply": secret_apply_stdout, "sub2apiAction": api_key_result["action"], "sub2apiId": api_key_result["id"], "groupId": api_key_result["groupId"], "userId": api_key_result["userId"], "keyPreview": api_key_preview(api_key), "valuesPrinted": False, }, "ownerBalance": owner_balance, "ownerConcurrency": owner_concurrency, "sentinel": {**sentinel, "qualityGatePrepare": sentinel_quality_prepare, "qualityGate": sentinel_quality, "freezeReassert": sentinel_reassert}, "runtimeCapabilities": runtime_capabilities, "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayResponsesRecent": responses_evidence, "gatewayCompactRecent": compact_evidence}, } def run_validate(): api_key = decode_secret_value(POOL_API_KEY_SECRET_NAME, POOL_API_KEY_SECRET_KEY) if not api_key: raise RuntimeError(f"{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY} missing") admin_email, token, admin_compliance = login() key_item = next((item for item in list_user_keys(token) if item.get("key") == api_key), None) owner_balance = None owner_concurrency = None if key_item is not None and key_item.get("user_id") is not None: 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) pool_group_id = key_item.get("group_id") if isinstance(key_item, dict) else None manual_account_protections = manual_account_protection_status(token, pool_group_id) gateway = validate_gateway(api_key) responses_smoke = validate_gateway_responses(api_key) compact_evidence = recent_compact_gateway_evidence() responses_evidence = recent_responses_gateway_evidence() runtime_capabilities = validate_runtime_capabilities(token) sentinel = sentinel_runtime_status() 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 load_factor_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True and manual_account_protections.get("ok") is True and sentinel.get("ok") is True and runtime_capabilities.get("ok") is True, "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")) or bool(responses_evidence.get("degraded")) or runtime_capabilities.get("ok") is not True, "mode": "validate", "namespace": NAMESPACE, "serviceDns": SERVICE_DNS, "appPod": APP_POD, "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance}, "apiKey": { "secret": f"{NAMESPACE}/{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY}", "sub2apiId": key_item.get("id") if isinstance(key_item, dict) else None, "userId": key_item.get("user_id") if isinstance(key_item, dict) else None, "groupId": key_item.get("group_id") if isinstance(key_item, dict) else None, "keyPreview": api_key_preview(api_key), "valuesPrinted": False, }, "ownerBalance": owner_balance, "ownerConcurrency": owner_concurrency, "capacity": capacity_status, "loadFactor": load_factor_status, "webSocketsV2": ws_v2_status, "tempUnschedulable": temp_unschedulable_status, "manualAccounts": manual_account_protections, "sentinel": sentinel, "runtimeCapabilities": runtime_capabilities, "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayResponsesRecent": responses_evidence, "gatewayCompactRecent": compact_evidence}, } def parse_log_line(line): json_start = line.find("{") if json_start < 0: return None prefix = line[:json_start].rstrip() try: item = json.loads(line[json_start:]) except Exception: return None if not isinstance(item, dict): return None at = None parts = prefix.split() if parts: at = parts[0] message = "" if len(parts) >= 4: message = " ".join(parts[3:]) elif len(parts) >= 3: message = parts[2] elif len(parts) >= 1: message = parts[-1] item["_at"] = at item["_message"] = message item["_line"] = line return item def log_time_epoch(item): at = item.get("_at") if isinstance(item, dict) else None if not isinstance(at, str) or not at: return None try: return datetime.strptime(at, "%Y-%m-%dT%H:%M:%S.%f%z").timestamp() except Exception: try: return datetime.fromisoformat(at.replace("Z", "+00:00")).timestamp() except Exception: return None def event_base(item, account_names_by_id): account_id = item.get("account_id") if isinstance(account_id, str) and account_id.isdigit(): account_id = int(account_id) account_name = account_names_by_id.get(account_id) return { "at": item.get("_at"), "message": item.get("_message"), "requestId": item.get("request_id"), "clientRequestId": item.get("client_request_id"), "path": item.get("path"), "method": item.get("method"), "model": item.get("model"), "accountId": account_id, "accountName": account_name, "statusCode": item.get("status_code"), "upstreamStatus": item.get("upstream_status"), "latencyMs": item.get("latency_ms"), } def classify_trace_event(item, account_names_by_id): message = str(item.get("_message") or "") event = event_base(item, account_names_by_id) if "content_moderation.gateway_check_start" in message: event.update({ "type": "request-start", "stream": item.get("stream"), "bodyBytes": item.get("body_bytes"), "groupId": item.get("group_id"), "groupName": item.get("group_name"), "apiKeyName": item.get("api_key_name"), }) elif "content_moderation.gateway_check_done" in message: event.update({ "type": "gateway-check", "allowed": item.get("allowed"), "blocked": item.get("blocked"), "action": item.get("action"), }) elif "openai.upstream_failover_switching" in message: event.update({ "type": "failover", "switchCount": item.get("switch_count"), "maxSwitches": item.get("max_switches"), }) elif "openai.account_select_failed" in message: event.update({ "type": "select-failed", "error": item.get("error"), "excludedAccountCount": item.get("excluded_account_count"), }) elif "account_upstream_error" in message: event.update({ "type": "upstream-error", "error": item.get("error"), }) elif "account_temp_unschedulable" in message: event.update({ "type": "temp-unschedulable", "until": item.get("until") or item.get("temp_unschedulable_until"), "ruleIndex": item.get("rule_index"), "matchedKeyword": item.get("matched_keyword"), "reason": item.get("reason") or item.get("error"), }) elif "http request completed" in message: event.update({ "type": "final", "clientIp": item.get("client_ip"), "protocol": item.get("protocol"), "platform": item.get("platform"), "completedAt": item.get("completed_at"), }) elif "admin account schedulable updated" in message or "account schedulable updated" in message or "/schedulable" in str(item.get("path") or ""): event.update({ "type": "admin-schedulable", "schedulable": item.get("schedulable"), }) else: event.update({"type": "other"}) return event def with_trace_phase(event, first_epoch, last_epoch): epoch = None at = event.get("at") if isinstance(event, dict) else None if isinstance(at, str) and at: epoch = log_time_epoch({"_at": at}) if epoch is None or first_epoch is None: phase = "unknown" elif epoch < first_epoch: phase = "before-request" elif last_epoch is not None and epoch > last_epoch: phase = "after-request" else: phase = "during-request" event["phase"] = phase return event def account_snapshot_from_runtime(token): try: accounts = list_accounts(token) except Exception as exc: return [], {"error": str(exc)} rows = [] for item in accounts: if not isinstance(item, dict): continue rows.append({ "accountId": item.get("id"), "accountName": item.get("name"), "schedulable": item.get("schedulable"), "status": item.get("status"), "concurrency": item.get("concurrency"), "priority": item.get("priority"), "tempUnschedulableUntil": item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil"), "tempUnschedulableSet": (item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil")) is not None or bool(item.get("temp_unschedulable_reason") or item.get("tempUnschedulableReason")), }) rows.sort(key=lambda row: (str(row.get("accountName") or ""), int(row.get("accountId") or 0))) return rows, None def trace_reason(events, final_event): failovers = [item for item in events if item.get("type") == "failover"] select_failures = [item for item in events if item.get("type") == "select-failed"] upstream_errors = [item for item in events if item.get("type") == "upstream-error"] if failover_budget_exhausted(failovers, final_event): return "failover-budget-exhausted" if failovers and select_failures: return "failover-attempted-no-candidate" if failovers: return "failover-attempted" if select_failures: return "account-select-failed" if upstream_errors: return "upstream-error" if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") >= 400: return "final-http-error" if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int): return "completed" return "unknown" def failover_budget_exhausted(failovers, final_event): if not failovers or not isinstance(final_event, dict): return False last = failovers[-1] switch_count = last.get("switchCount") max_switches = last.get("maxSwitches") final_status = final_event.get("statusCode") return ( 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 ) def trace_untried_schedulable_accounts(failovers, final_event, account_snapshot): if not failover_budget_exhausted(failovers, final_event): return [] tried = set() for item in failovers: account_id = item.get("accountId") if isinstance(account_id, int): tried.add(account_id) final_account = final_event.get("accountId") if isinstance(final_event, dict) else None if isinstance(final_account, int): tried.add(final_account) result = [] for item in account_snapshot: account_id = item.get("accountId") if not isinstance(account_id, int) or account_id in tried: continue if item.get("schedulable") is True and item.get("status") == "active" and item.get("tempUnschedulableSet") is not True: result.append({ "accountId": account_id, "accountName": item.get("accountName"), "priority": item.get("priority"), "concurrency": item.get("concurrency"), }) return result def run_trace(): payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {} request_id = payload.get("requestId") since = payload.get("since") or "24h" tail = int(payload.get("tail") or 20000) context_seconds = int(payload.get("contextSeconds") or 300) show_lines = bool(payload.get("showLines")) if not isinstance(request_id, str) or not request_id: raise RuntimeError("trace payload missing requestId") admin_email, token, admin_compliance = login() account_snapshot, account_snapshot_error = account_snapshot_from_runtime(token) account_names_by_id = {} for row in account_snapshot: account_id = row.get("accountId") if isinstance(account_id, str) and account_id.isdigit(): account_id = int(account_id) if isinstance(account_id, int) and isinstance(row.get("accountName"), str): account_names_by_id[account_id] = row.get("accountName") proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", f"--since={since}", f"--tail={tail}"]) stdout = proc.stdout.decode("utf-8", errors="replace") parsed_lines = [] matched = [] for line in stdout.splitlines(): parsed = parse_log_line(line) if parsed is None: continue parsed_lines.append(parsed) if request_id in line: matched.append(parsed) first_epoch = None last_epoch = None for item in matched: epoch = log_time_epoch(item) if epoch is None: continue first_epoch = epoch if first_epoch is None else min(first_epoch, epoch) last_epoch = epoch if last_epoch is None else max(last_epoch, epoch) window_lines = [] if first_epoch is not None: start_epoch = first_epoch - context_seconds end_epoch = (last_epoch if last_epoch is not None else first_epoch) + context_seconds for item in parsed_lines: epoch = log_time_epoch(item) if epoch is not None and start_epoch <= epoch <= end_epoch: window_lines.append(item) else: window_lines = matched events = [classify_trace_event(item, account_names_by_id) for item in matched] request_start = next((item for item in events if item.get("type") == "request-start"), None) final_event = next((item for item in reversed(events) if item.get("type") == "final"), None) failovers = [item for item in events if item.get("type") == "failover"] select_failures = [item for item in events if item.get("type") == "select-failed"] upstream_errors = [item for item in events if item.get("type") == "upstream-error"] temp_unsched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if "account_temp_unschedulable" in str(item.get("_message") or "")] admin_sched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if ("schedulable" in str(item.get("_message") or "") or "/schedulable" in str(item.get("path") or ""))] window_events = [classify_trace_event(item, account_names_by_id) for item in window_lines] final_errors = [item for item in window_events if item.get("type") == "final" and isinstance(item.get("statusCode"), int) and item.get("statusCode") >= 400] window_failovers = [item for item in window_events if item.get("type") == "failover"] window_select_failures = [item for item in window_events if item.get("type") == "select-failed"] untried_schedulable_accounts = trace_untried_schedulable_accounts(failovers, final_event or {}, account_snapshot) reason = trace_reason(events, final_event) if not matched: outcome = "not-found" elif isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") < 400: outcome = "succeeded" elif isinstance(final_event, dict): outcome = "failed" else: outcome = "incomplete" return { "ok": proc.returncode == 0 and len(matched) > 0, "mode": "trace", "namespace": NAMESPACE, "serviceDns": SERVICE_DNS, "appPod": APP_POD, "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance}, "requestId": request_id, "summary": { "outcome": outcome, "reason": reason, "eventCount": len(events), "matchedLineCount": len(matched), "firstAt": events[0].get("at") if events else None, "lastAt": events[-1].get("at") if events else None, }, "window": { "since": since, "tail": tail, "beforeSeconds": context_seconds, "afterSeconds": context_seconds, "lineCount": len(window_lines), }, "request": request_start or {}, "final": final_event or {}, "events": events, "failovers": failovers, "selectFailures": select_failures, "upstreamErrors": upstream_errors, "tempUnschedulable": temp_unsched, "adminSchedulable": admin_sched[-20:], "windowStats": { "matchedLines": len(matched), "eventCount": len(window_events), "finalErrorCount": len(final_errors), "failoverCount": len(window_failovers), "selectFailedCount": len(window_select_failures), "tempUnschedulableCount": len(temp_unsched), "adminSchedulableCount": len(admin_sched), }, "diagnostics": { "failoverBudgetExhausted": failover_budget_exhausted(failovers, final_event or {}), "untriedSchedulableAccounts": untried_schedulable_accounts, }, "accountSnapshot": account_snapshot, "accountSnapshotError": account_snapshot_error, "rawLines": [{"line": item.get("_line")} for item in matched[-30:]] if show_lines else [], "showLines": show_lines, "logs": { "exitCode": proc.returncode, "stderrTail": text(proc.stderr, 1000), "stdoutLineCount": len(stdout.splitlines()), }, "valuesPrinted": False, } def parse_embedded_json(stdout): if not isinstance(stdout, str) or not stdout.strip(): return None start = stdout.find("{") end = stdout.rfind("}") if start < 0 or end <= start: return None try: return json.loads(stdout[start:end + 1]) except Exception: return None def inject_env(template, name, value): spec = template.setdefault("spec", {}) containers = spec.setdefault("containers", []) if not containers: raise RuntimeError("sentinel job template has no containers") container = containers[0] env = container.setdefault("env", []) env[:] = [item for item in env if not (isinstance(item, dict) and item.get("name") == name)] env.append({"name": name, "value": value}) def sentinel_probe_job_manifest(accounts): cronjob_name = SENTINEL_CONFIG.get("cronJobName") if not isinstance(cronjob_name, str) or not cronjob_name: raise RuntimeError("sentinel cronJobName missing from config") cronjob = kube_json(["-n", NAMESPACE, "get", "cronjob", cronjob_name], f"cronjob/{cronjob_name}") job_template = ((cronjob.get("spec") or {}).get("jobTemplate") or {}) if isinstance(cronjob, dict) else {} job_spec = copy_json(job_template.get("spec") or {}) if not isinstance(job_spec, dict) or not job_spec: raise RuntimeError("sentinel CronJob jobTemplate.spec missing") template = job_spec.get("template") if not isinstance(template, dict): raise RuntimeError("sentinel CronJob jobTemplate.spec.template missing") inject_env(template, "SENTINEL_ACCOUNT_NAMES", ",".join(accounts)) job_spec["ttlSecondsAfterFinished"] = int(job_spec.get("ttlSecondsAfterFinished") or 3600) suffix = str(int(time.time()))[-8:] + "-" + "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(5)) job_name = ("sub2api-sentinel-probe-" + suffix)[:63] return job_name, { "apiVersion": "batch/v1", "kind": "Job", "metadata": { "name": job_name, "namespace": NAMESPACE, "labels": { "app.kubernetes.io/name": cronjob_name, "app.kubernetes.io/part-of": "platform-infra", "app.kubernetes.io/managed-by": "unidesk", "unidesk.ai/job-purpose": "sub2api-account-sentinel-manual-probe", }, }, "spec": job_spec, } def copy_json(value): return json.loads(json.dumps(value)) def job_condition(job, cond_type): for item in ((job.get("status") or {}).get("conditions") or []): if item.get("type") == cond_type and item.get("status") == "True": return item return None def wait_sentinel_probe_job(job_name, timeout_seconds): deadline = time.time() + timeout_seconds latest = None while time.time() < deadline: latest, err = safe_kube_json(["-n", NAMESPACE, "get", "job", job_name], f"job/{job_name}") if isinstance(latest, dict): complete = job_condition(latest, "Complete") failed = job_condition(latest, "Failed") if complete is not None: return "succeeded", latest, complete if failed is not None: return "failed", latest, failed time.sleep(2) return "timeout", latest, None def job_logs(job_name): proc = kubectl(["-n", NAMESPACE, "logs", f"job/{job_name}", "--tail=4000"]) return { "exitCode": proc.returncode, "stdout": proc.stdout.decode("utf-8", errors="replace"), "stderr": text(proc.stderr, 4000), } def run_sentinel_probe(): payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {} accounts = payload.get("accounts") if isinstance(payload, dict) else None if not isinstance(accounts, list) or not accounts or not all(isinstance(item, str) and item for item in accounts): raise RuntimeError("sentinel-probe payload requires non-empty accounts") job_name, manifest = sentinel_probe_job_manifest(accounts) proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], json.dumps(manifest)) if proc.returncode != 0: raise RuntimeError(f"apply sentinel probe job failed: {text(proc.stderr, 2000)}") timeout_seconds = max(300, int(SENTINEL_CONFIG.get("probe", {}).get("timeoutSeconds") or 30) + 240) status, job, condition = wait_sentinel_probe_job(job_name, timeout_seconds) logs = job_logs(job_name) parsed = parse_embedded_json(logs.get("stdout") or "") results = parsed.get("results") if isinstance(parsed, dict) and isinstance(parsed.get("results"), list) else [] requested = set(accounts) measured = {item.get("accountName") for item in results if isinstance(item, dict)} missing = sorted(name for name in requested if name not in measured) marker_ok = len(missing) == 0 and all(isinstance(item, dict) and item.get("accountName") in requested and item.get("markerMatched") is True for item in results if isinstance(item, dict) and item.get("accountName") in requested) job_ok = status == "succeeded" and isinstance(parsed, dict) and parsed.get("ok") is True return { "ok": job_ok and marker_ok, "jobExecutionOk": job_ok, "markerOk": marker_ok, "mode": "sentinel-probe", "namespace": NAMESPACE, "requestedAccounts": accounts, "missingAccounts": missing, "job": { "name": job_name, "status": status, "condition": condition, "timeoutSeconds": timeout_seconds, "applyStdoutTail": text(proc.stdout, 1200), "logsExitCode": logs.get("exitCode"), "logsStderrTail": logs.get("stderr"), }, "probe": parsed, "sentinelState": sentinel_state_summary(), "valuesPrinted": False, } def run_cleanup_probes(): admin_email, token, admin_compliance = login() cleanup = cleanup_probe_resources(token) return { "ok": cleanup.get("ok") is True, "mode": "cleanup-probes", "namespace": NAMESPACE, "serviceDns": SERVICE_DNS, "appPod": APP_POD, "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance}, "cleanup": cleanup, "valuesPrinted": False, } try: if MODE == "sync": result = run_sync() elif MODE == "trace": result = run_trace() elif MODE == "cleanup-probes": result = run_cleanup_probes() elif MODE == "sentinel-probe": result = run_sentinel_probe() else: result = run_validate() except Exception as exc: result = { "ok": False, "mode": MODE, "namespace": NAMESPACE, "serviceDns": SERVICE_DNS, "appPod": globals().get("APP_POD"), "error": str(exc), "valuesPrinted": False, } print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0 if result.get("ok") else 1) PY `; }