import { Buffer } from "node:buffer"; export interface CodexPoolSentinelConfig { monitor: { enabled: boolean; }; actions: { enabled: boolean; }; schedule: string; image: string; serviceAccountName: string; configMapName: string; credentialsSecretName: string; stateConfigMapName: string; cronJobName: string; model: string; endpoint: "responses"; marker: { prefix: string; exact: boolean; }; probe: { timeoutSeconds: number; maxOutputTokens: number; transportRetryMinutes: number; userAgent: string; }; sdk: { openaiPythonVersion: string; }; cadence: { successInitialIntervalMinutes: number; successMaxIntervalMinutes: number; successBackoffMultiplier: number; jitterPercent: number; }; freeze: { initialTtlMinutes: number; maxTtlMinutes: number; backoffMultiplier: number; jitterPercent: number; }; pricing: { usdPer1MInputTokens: number; usdPer1MOutputTokens: number; }; historyLimit: number; } export interface CodexPoolSentinelImageTarget { baseImage: string; runtimeImage: string; repository: string; tag: string; } export interface CodexPoolSentinelProfileSecret { accountName: string; profile: string; baseUrl: string; apiKey: string; upstreamUserAgent: string | null; } export interface CodexPoolSentinelManifestOptions { namespace: string; serviceName: string; serviceDns: string; appSecretName: string; } export function defaultCodexPoolSentinelConfig(): CodexPoolSentinelConfig { return { monitor: { enabled: true, }, actions: { enabled: false, }, schedule: "*/1 * * * *", image: "python:3.12-alpine", serviceAccountName: "sub2api-account-sentinel", configMapName: "sub2api-account-sentinel-config", credentialsSecretName: "sub2api-account-sentinel-profiles", stateConfigMapName: "sub2api-account-sentinel-state", cronJobName: "sub2api-account-sentinel", model: "gpt-5.5", endpoint: "responses", marker: { prefix: "UDSG_OK", exact: true, }, probe: { timeoutSeconds: 30, maxOutputTokens: 16, transportRetryMinutes: 5, userAgent: "Go-http-client/1.1", }, sdk: { openaiPythonVersion: "2.41.1", }, cadence: { successInitialIntervalMinutes: 1, successMaxIntervalMinutes: 20, successBackoffMultiplier: 2, jitterPercent: 10, }, freeze: { initialTtlMinutes: 1, maxTtlMinutes: 10, backoffMultiplier: 2, jitterPercent: 10, }, pricing: { usdPer1MInputTokens: 1.25, usdPer1MOutputTokens: 10, }, historyLimit: 200, }; } export function codexPoolSentinelRuntimeImage(config: CodexPoolSentinelConfig): CodexPoolSentinelImageTarget { const baseTag = config.image .replace(/[^A-Za-z0-9_.-]+/gu, "-") .replace(/^-+|-+$/gu, "") .slice(0, 80) || "python"; const tag = `${baseTag}-openai-${config.sdk.openaiPythonVersion}`; const repository = "127.0.0.1:5000/platform-infra/sub2api-account-sentinel"; return { baseImage: config.image, runtimeImage: `${repository}:${tag}`, repository, tag, }; } export function readCodexPoolSentinelConfig(value: unknown, defaults: CodexPoolSentinelConfig, sourcePath: string): CodexPoolSentinelConfig { if (!isRecord(value)) return defaults; const monitor = isRecord(value.monitor) ? value.monitor : {}; const actions = isRecord(value.actions) ? value.actions : {}; const marker = isRecord(value.marker) ? value.marker : {}; const probe = isRecord(value.probe) ? value.probe : {}; const sdk = isRecord(value.sdk) ? value.sdk : {}; const cadence = isRecord(value.cadence) ? value.cadence : {}; const freeze = isRecord(value.freeze) ? value.freeze : {}; const pricing = isRecord(value.pricing) ? value.pricing : {}; const config: CodexPoolSentinelConfig = { monitor: { enabled: readBoolean(valueAt(monitor, "enabled"), `${sourcePath}.sentinel.monitor.enabled`, defaults.monitor.enabled), }, actions: { enabled: readBoolean(valueAt(actions, "enabled"), `${sourcePath}.sentinel.actions.enabled`, defaults.actions.enabled), }, schedule: readString(valueAt(value, "schedule"), `${sourcePath}.sentinel.schedule`, defaults.schedule), image: readString(valueAt(value, "image"), `${sourcePath}.sentinel.image`, defaults.image), serviceAccountName: readDnsName(valueAt(value, "serviceAccountName"), `${sourcePath}.sentinel.serviceAccountName`, defaults.serviceAccountName), configMapName: readDnsName(valueAt(value, "configMapName"), `${sourcePath}.sentinel.configMapName`, defaults.configMapName), credentialsSecretName: readDnsName(valueAt(value, "credentialsSecretName"), `${sourcePath}.sentinel.credentialsSecretName`, defaults.credentialsSecretName), stateConfigMapName: readDnsName(valueAt(value, "stateConfigMapName"), `${sourcePath}.sentinel.stateConfigMapName`, defaults.stateConfigMapName), cronJobName: readDnsName(valueAt(value, "cronJobName"), `${sourcePath}.sentinel.cronJobName`, defaults.cronJobName), model: readModelName(valueAt(value, "model"), `${sourcePath}.sentinel.model`, defaults.model), endpoint: "responses", marker: { prefix: readMarkerPrefix(valueAt(marker, "prefix"), `${sourcePath}.sentinel.marker.prefix`, defaults.marker.prefix), exact: readBoolean(valueAt(marker, "exact"), `${sourcePath}.sentinel.marker.exact`, defaults.marker.exact), }, probe: { timeoutSeconds: readInt(valueAt(probe, "timeoutSeconds"), `${sourcePath}.sentinel.probe.timeoutSeconds`, defaults.probe.timeoutSeconds, 3, 300), maxOutputTokens: readInt(valueAt(probe, "maxOutputTokens"), `${sourcePath}.sentinel.probe.maxOutputTokens`, defaults.probe.maxOutputTokens, 1, 128), transportRetryMinutes: readInt(valueAt(probe, "transportRetryMinutes"), `${sourcePath}.sentinel.probe.transportRetryMinutes`, defaults.probe.transportRetryMinutes, 1, 120), userAgent: readUserAgent(valueAt(probe, "userAgent"), `${sourcePath}.sentinel.probe.userAgent`, defaults.probe.userAgent), }, sdk: { openaiPythonVersion: readOpenAiPythonVersion(valueAt(sdk, "openaiPythonVersion"), `${sourcePath}.sentinel.sdk.openaiPythonVersion`, defaults.sdk.openaiPythonVersion), }, cadence: { successInitialIntervalMinutes: readInt(valueAt(cadence, "successInitialIntervalMinutes"), `${sourcePath}.sentinel.cadence.successInitialIntervalMinutes`, defaults.cadence.successInitialIntervalMinutes, 1, 1440), successMaxIntervalMinutes: readInt(valueAt(cadence, "successMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.successMaxIntervalMinutes`, defaults.cadence.successMaxIntervalMinutes, 1, 1440), successBackoffMultiplier: readInt(valueAt(cadence, "successBackoffMultiplier"), `${sourcePath}.sentinel.cadence.successBackoffMultiplier`, defaults.cadence.successBackoffMultiplier, 1, 10), jitterPercent: readInt(valueAt(cadence, "jitterPercent"), `${sourcePath}.sentinel.cadence.jitterPercent`, defaults.cadence.jitterPercent, 0, 50), }, freeze: { initialTtlMinutes: readInt(valueAt(freeze, "initialTtlMinutes"), `${sourcePath}.sentinel.freeze.initialTtlMinutes`, defaults.freeze.initialTtlMinutes, 1, 1440), maxTtlMinutes: readInt(valueAt(freeze, "maxTtlMinutes"), `${sourcePath}.sentinel.freeze.maxTtlMinutes`, defaults.freeze.maxTtlMinutes, 1, 1440), backoffMultiplier: readInt(valueAt(freeze, "backoffMultiplier"), `${sourcePath}.sentinel.freeze.backoffMultiplier`, defaults.freeze.backoffMultiplier, 1, 10), jitterPercent: readInt(valueAt(freeze, "jitterPercent"), `${sourcePath}.sentinel.freeze.jitterPercent`, defaults.freeze.jitterPercent, 0, 50), }, pricing: { usdPer1MInputTokens: readNumber(valueAt(pricing, "usdPer1MInputTokens"), `${sourcePath}.sentinel.pricing.usdPer1MInputTokens`, defaults.pricing.usdPer1MInputTokens, 0, 100000), usdPer1MOutputTokens: readNumber(valueAt(pricing, "usdPer1MOutputTokens"), `${sourcePath}.sentinel.pricing.usdPer1MOutputTokens`, defaults.pricing.usdPer1MOutputTokens, 0, 100000), }, historyLimit: readInt(valueAt(value, "historyLimit"), `${sourcePath}.sentinel.historyLimit`, defaults.historyLimit, 1, 2000), }; if (config.actions.enabled && !config.monitor.enabled) { throw new Error(`${sourcePath}.sentinel.actions.enabled requires sentinel.monitor.enabled=true`); } if (config.cadence.successMaxIntervalMinutes < config.cadence.successInitialIntervalMinutes) { throw new Error(`${sourcePath}.sentinel.cadence.successMaxIntervalMinutes must be >= successInitialIntervalMinutes`); } if (config.freeze.maxTtlMinutes < config.freeze.initialTtlMinutes) { throw new Error(`${sourcePath}.sentinel.freeze.maxTtlMinutes must be >= initialTtlMinutes`); } if (!/^[-0-9A-Za-z_/*,\s]+$/u.test(config.schedule)) { throw new Error(`${sourcePath}.sentinel.schedule has an unsupported cron format`); } if (!/^[A-Za-z0-9._:/@-]+$/u.test(config.image)) { throw new Error(`${sourcePath}.sentinel.image has an unsupported image format`); } return config; } export function codexPoolSentinelSummary(config: CodexPoolSentinelConfig): Record { return { monitorEnabled: config.monitor.enabled, actionsEnabled: config.actions.enabled, schedule: config.schedule, cronJobName: config.cronJobName, configMapName: config.configMapName, credentialsSecretName: config.credentialsSecretName, stateConfigMapName: config.stateConfigMapName, image: config.image, model: config.model, endpoint: config.endpoint, probe: config.probe, sdk: config.sdk, cadence: config.cadence, freeze: config.freeze, accounting: { mode: "record-only", pricing: config.pricing, }, marker: { prefix: config.marker.prefix, exact: config.marker.exact, }, valuesPrinted: false, }; } export function renderCodexPoolSentinelManifest( config: CodexPoolSentinelConfig, profiles: CodexPoolSentinelProfileSecret[], options: CodexPoolSentinelManifestOptions, ): string { const profilesJson = JSON.stringify({ profiles }, null, 2); const runnerConfig = { monitor: config.monitor, actions: config.actions, service: { baseUrl: `http://${options.serviceDns}`, adminEmailDefault: "admin@sub2api.platform-infra.local", }, model: config.model, endpoint: config.endpoint, marker: config.marker, probe: config.probe, sdk: config.sdk, cadence: config.cadence, freeze: config.freeze, pricing: config.pricing, state: { configMapName: config.stateConfigMapName, historyLimit: config.historyLimit, }, }; const suspend = config.monitor.enabled ? "false" : "true"; const activeDeadlineSeconds = Math.max(300, Math.min(3600, config.probe.timeoutSeconds + 240)); const command = sentinelContainerShellCommand(config); const runtimeImage = codexPoolSentinelRuntimeImage(config).runtimeImage; return `apiVersion: v1 kind: Secret metadata: name: ${config.credentialsSecretName} namespace: ${options.namespace} labels: app.kubernetes.io/name: ${config.cronJobName} app.kubernetes.io/part-of: platform-infra app.kubernetes.io/managed-by: unidesk unidesk.ai/secret-purpose: sub2api-account-sentinel-profiles type: Opaque data: profiles.json: ${Buffer.from(profilesJson, "utf8").toString("base64")} --- apiVersion: v1 kind: ConfigMap metadata: name: ${config.configMapName} namespace: ${options.namespace} labels: app.kubernetes.io/name: ${config.cronJobName} app.kubernetes.io/part-of: platform-infra app.kubernetes.io/managed-by: unidesk data: config.json: | ${indentBlock(JSON.stringify(runnerConfig, null, 2), 4)} sentinel.py: | ${indentBlock(sentinelRunnerPython(), 4)} --- apiVersion: v1 kind: ServiceAccount metadata: name: ${config.serviceAccountName} namespace: ${options.namespace} labels: app.kubernetes.io/name: ${config.cronJobName} app.kubernetes.io/part-of: platform-infra app.kubernetes.io/managed-by: unidesk --- apiVersion: rbac.authorization.k8s.io/v1 kind: Role metadata: name: ${config.serviceAccountName} namespace: ${options.namespace} labels: app.kubernetes.io/name: ${config.cronJobName} app.kubernetes.io/part-of: platform-infra app.kubernetes.io/managed-by: unidesk rules: - apiGroups: [""] resources: ["configmaps"] verbs: ["get", "create", "update", "patch"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding metadata: name: ${config.serviceAccountName} namespace: ${options.namespace} labels: app.kubernetes.io/name: ${config.cronJobName} app.kubernetes.io/part-of: platform-infra app.kubernetes.io/managed-by: unidesk subjects: - kind: ServiceAccount name: ${config.serviceAccountName} namespace: ${options.namespace} roleRef: kind: Role name: ${config.serviceAccountName} apiGroup: rbac.authorization.k8s.io --- apiVersion: batch/v1 kind: CronJob metadata: name: ${config.cronJobName} namespace: ${options.namespace} labels: app.kubernetes.io/name: ${config.cronJobName} app.kubernetes.io/part-of: platform-infra app.kubernetes.io/managed-by: unidesk spec: schedule: "${config.schedule}" concurrencyPolicy: Forbid suspend: ${suspend} successfulJobsHistoryLimit: 3 failedJobsHistoryLimit: 5 jobTemplate: spec: ttlSecondsAfterFinished: 3600 activeDeadlineSeconds: ${activeDeadlineSeconds} template: metadata: labels: app.kubernetes.io/name: ${config.cronJobName} app.kubernetes.io/part-of: platform-infra app.kubernetes.io/managed-by: unidesk spec: serviceAccountName: ${config.serviceAccountName} restartPolicy: Never containers: - name: sentinel image: ${runtimeImage} imagePullPolicy: IfNotPresent command: ["sh", "-c"] args: - ${JSON.stringify(command)} env: - name: ADMIN_EMAIL valueFrom: configMapKeyRef: name: sub2api-config key: ADMIN_EMAIL optional: true - name: ADMIN_PASSWORD valueFrom: secretKeyRef: name: ${options.appSecretName} key: ADMIN_PASSWORD optional: true - name: POD_NAMESPACE valueFrom: fieldRef: fieldPath: metadata.namespace volumeMounts: - name: sentinel-config mountPath: /opt/sentinel readOnly: true - name: sentinel-profiles mountPath: /opt/sentinel-secrets readOnly: true volumes: - name: sentinel-config configMap: name: ${config.configMapName} defaultMode: 0555 - name: sentinel-profiles secret: secretName: ${config.credentialsSecretName} `; } export function sentinelContainerShellCommand(config: CodexPoolSentinelConfig): string { return [ "set -eu", "python3 - <<'PY'", "import importlib.metadata", `expected = ${JSON.stringify(config.sdk.openaiPythonVersion)}`, "try:", " current = importlib.metadata.version('openai')", "except importlib.metadata.PackageNotFoundError:", " current = None", "if current != expected:", " raise SystemExit(f'openai-python-version-mismatch expected={expected} current={current}')", "PY", "exec python3 /opt/sentinel/sentinel.py", ].join("\n"); } export function sentinelRunnerPython(): string { return String.raw`#!/usr/bin/env python3 import base64 import hashlib import json import math import os import random import ssl import time import traceback from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone, timedelta from urllib import error, parse, request from openai import APIConnectionError, APIStatusError, APITimeoutError, OpenAI CONFIG_PATH = "/opt/sentinel/config.json" PROFILES_PATH = "/opt/sentinel-secrets/profiles.json" STATE_KEY = "state.json" def utc_now(): return datetime.now(timezone.utc) def iso(dt): return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def parse_iso(value): if not isinstance(value, str) or not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except Exception: return None def load_json(path): with open(path, encoding="utf-8") as handle: return json.load(handle) def sha(value): return hashlib.sha256(str(value).encode("utf-8", errors="replace")).hexdigest()[:16] def preview(value, limit=160): if not isinstance(value, str): value = str(value) value = " ".join(value.replace("\r", " ").replace("\n", " ").split()) return value[:limit] def day_key(now): return now.strftime("%Y-%m-%d") def add_minutes(now, minutes, jitter_percent=0): minutes = float(minutes) if jitter_percent: factor = 1 + random.uniform(-jitter_percent, jitter_percent) / 100 minutes = max(1, minutes * factor) return now + timedelta(minutes=minutes) def estimate_tokens(text): if not isinstance(text, str) or not text: return 0 return max(1, math.ceil(len(text) / 4)) class KubeClient: def __init__(self, namespace): self.namespace = namespace with open("/var/run/secrets/kubernetes.io/serviceaccount/token", encoding="utf-8") as handle: self.token = handle.read().strip() self.base = "https://kubernetes.default.svc" self.context = ssl.create_default_context(cafile="/var/run/secrets/kubernetes.io/serviceaccount/ca.crt") def api(self, method, path, payload=None): body = None if payload is None else json.dumps(payload).encode("utf-8") req = request.Request( self.base + path, data=body, method=method, headers={ "Authorization": "Bearer " + self.token, "Accept": "application/json", "Content-Type": "application/json", }, ) try: with request.urlopen(req, timeout=15, context=self.context) as resp: raw = resp.read() return resp.status, json.loads(raw.decode("utf-8")) if raw else None except error.HTTPError as exc: raw = exc.read() try: parsed = json.loads(raw.decode("utf-8")) if raw else None except Exception: parsed = {"message": raw.decode("utf-8", errors="replace")} return exc.code, parsed def get_configmap(self, name): status, data = self.api("GET", f"/api/v1/namespaces/{self.namespace}/configmaps/{name}") if status == 404: return None if status >= 300: raise RuntimeError(f"get configmap {name} failed: {status} {data}") return data def create_configmap(self, name, state): payload = { "apiVersion": "v1", "kind": "ConfigMap", "metadata": { "name": name, "namespace": self.namespace, "labels": { "app.kubernetes.io/part-of": "platform-infra", "app.kubernetes.io/managed-by": "unidesk", "unidesk.ai/state-purpose": "sub2api-account-sentinel", }, }, "data": {STATE_KEY: json.dumps(state, ensure_ascii=False, indent=2)}, } status, data = self.api("POST", f"/api/v1/namespaces/{self.namespace}/configmaps", payload) if status >= 300 and status != 409: raise RuntimeError(f"create state configmap {name} failed: {status} {data}") return self.get_configmap(name) def update_configmap_state(self, obj, state): obj.setdefault("data", {})[STATE_KEY] = json.dumps(state, ensure_ascii=False, indent=2) name = obj["metadata"]["name"] status, data = self.api("PUT", f"/api/v1/namespaces/{self.namespace}/configmaps/{name}", obj) if status >= 300: raise RuntimeError(f"update state configmap {name} failed: {status} {data}") return data def default_state(): return { "version": 1, "accounts": {}, "ledger": {}, "history": [], } def load_state(kube, config): name = config["state"]["configMapName"] obj = kube.get_configmap(name) if obj is None: obj = kube.create_configmap(name, default_state()) raw = (obj.get("data") or {}).get(STATE_KEY) try: state = json.loads(raw) if raw else default_state() except Exception: state = default_state() state["stateLoadWarning"] = "invalid-state-json-reset" state.setdefault("version", 1) state.setdefault("accounts", {}) state.setdefault("ledger", {}) state.setdefault("history", []) return obj, state def http_json(method, url, headers=None, payload=None, timeout=30, max_bytes=65536): body = None if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8") req = request.Request(url, data=body, method=method, headers=headers or {}) started = time.time() try: with request.urlopen(req, timeout=timeout) as resp: raw = resp.read(max_bytes + 1) too_large = len(raw) > max_bytes if too_large: raw = raw[:max_bytes] text = raw.decode("utf-8", errors="replace") parsed = None try: parsed = json.loads(text) if text.strip() else None except Exception: parsed = None app_success = not (isinstance(parsed, dict) and parsed.get("code") not in (None, 0)) return { "ok": 200 <= resp.status < 300 and not too_large and app_success, "status": resp.status, "json": parsed, "text": text, "tooLarge": too_large, "durationMs": int((time.time() - started) * 1000), } except error.HTTPError as exc: raw = exc.read(max_bytes + 1) text = raw.decode("utf-8", errors="replace") parsed = None try: parsed = json.loads(text) if text.strip() else None except Exception: parsed = None return { "ok": False, "status": exc.code, "json": parsed, "text": text, "tooLarge": len(raw) > max_bytes, "durationMs": int((time.time() - started) * 1000), "error": str(exc), } except Exception as exc: return { "ok": False, "status": 0, "json": None, "text": "", "tooLarge": False, "durationMs": int((time.time() - started) * 1000), "error": str(exc), } def find_token(value): if isinstance(value, dict): for key in ("access_token", "token"): if isinstance(value.get(key), str) and value[key]: return value[key] for item in value.values(): found = find_token(item) if found: return found if isinstance(value, list): for item in value: found = find_token(item) if found: return found return None def url_quote(value): return parse.quote(str(value), safe="") class Sub2ApiAdmin: def __init__(self, config): self.base = config["service"]["baseUrl"].rstrip("/") self.email = os.environ.get("ADMIN_EMAIL") or config["service"]["adminEmailDefault"] self.password = os.environ.get("ADMIN_PASSWORD") or "" self.token = None self.accounts_by_name = None def login(self): if self.token: return self.token if not self.password: raise RuntimeError("ADMIN_PASSWORD is missing") resp = http_json("POST", self.base + "/api/v1/auth/login", {"Content-Type": "application/json"}, {"email": self.email, "password": self.password}, timeout=15) token = find_token(resp.get("json")) if not resp["ok"] or not token: raise RuntimeError("admin login failed") self.token = token return token def request(self, method, path, payload=None): token = self.login() headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"} resp = http_json(method, self.base + path, headers, payload, timeout=20) if not resp["ok"]: raise RuntimeError(f"admin {method} {path} failed: {resp.get('status')} {preview(resp.get('text', ''), 300)}") parsed = resp.get("json") if isinstance(parsed, dict) and parsed.get("code") not in (None, 0): raise RuntimeError(f"admin {method} {path} failed: code={parsed.get('code')} message={parsed.get('message')}") if isinstance(parsed, dict) and "data" in parsed: return parsed["data"] return parsed def accounts(self): if self.accounts_by_name is not None: return self.accounts_by_name data = self.request("GET", "/api/v1/admin/accounts?page=1&page_size=500&platform=openai&type=apikey&search=unidesk-codex-") items = [] if isinstance(data, list): items = data elif isinstance(data, dict): for key in ("items", "accounts"): if isinstance(data.get(key), list): items = data[key] break self.accounts_by_name = {item.get("name"): item for item in items if isinstance(item, dict) and isinstance(item.get("name"), str)} return self.accounts_by_name def account(self, account_name): if self.accounts_by_name is not None and account_name in self.accounts_by_name: return self.accounts_by_name[account_name] data = self.request("GET", "/api/v1/admin/accounts?page=1&page_size=20&platform=openai&type=apikey&search=" + url_quote(account_name)) items = data if isinstance(data, list) else [] if isinstance(data, dict): for key in ("items", "accounts"): if isinstance(data.get(key), list): items = data[key] break for item in items: if isinstance(item, dict) and item.get("name") == account_name: if self.accounts_by_name is not None: self.accounts_by_name[account_name] = item return item return None def set_schedulable(self, account_name, schedulable): account = self.account(account_name) if not account or account.get("id") is None: raise RuntimeError(f"account {account_name} not found") self.request("POST", f"/api/v1/admin/accounts/{account['id']}/schedulable", {"schedulable": bool(schedulable)}) return {"accountId": account.get("id"), "schedulable": bool(schedulable)} def recover_state(self, account_name): account = self.account(account_name) if not account or account.get("id") is None: return {"skipped": True, "reason": "account-not-found"} try: self.request("POST", f"/api/v1/admin/accounts/{account['id']}/recover-state", {}) return {"ok": True, "accountId": account.get("id")} except Exception as exc: return {"ok": False, "accountId": account.get("id"), "error": str(exc)} def upstream_base_url(base_url): base = str(base_url).rstrip("/") return base if base.endswith("/v1") else base + "/v1" def output_text(parsed): if isinstance(parsed, dict) and isinstance(parsed.get("output_text"), str): return parsed["output_text"] parts = [] output = parsed.get("output") if isinstance(parsed, dict) else None if isinstance(output, list): 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 isinstance(block, dict) and isinstance(block.get("text"), str): parts.append(block["text"]) return "\n".join(parts) def model_dump(value): if hasattr(value, "model_dump"): return value.model_dump() if isinstance(value, dict): return value return {} def body_text(value): if isinstance(value, bytes): return value.decode("utf-8", errors="replace") if isinstance(value, str): return value try: return json.dumps(value, ensure_ascii=False) except Exception: return str(value) def redact_diagnostic(value): if isinstance(value, dict): redacted = {} for key, item in value.items(): key_text = str(key) if any(token in key_text.lower() for token in ("key", "token", "secret", "password", "credential", "authorization")): redacted[key_text] = "[redacted]" else: redacted[key_text] = redact_diagnostic(item) return redacted if isinstance(value, list): return [redact_diagnostic(item) for item in value[:20]] if isinstance(value, str): return value if len(value) <= 2000 else value[:2000] + "...[truncated]" if isinstance(value, (int, float, bool)) or value is None: return value return str(value) def selected_headers(headers): if headers is None: return {} selected = {} for key in ( "content-type", "x-request-id", "x-ratelimit-limit-requests", "x-ratelimit-remaining-requests", "x-ratelimit-reset-requests", "cf-ray", "server", ): try: value = headers.get(key) except Exception: value = None if value: selected[key] = str(value) return selected def openai_error_fields(body): if not isinstance(body, dict): return {} error_obj = body.get("error") if isinstance(error_obj, dict): return { "message": error_obj.get("message"), "type": error_obj.get("type"), "param": error_obj.get("param"), "code": error_obj.get("code"), } return { "message": body.get("message"), "type": body.get("type"), "param": body.get("param"), "code": body.get("code"), } def error_details(kind, status, body=None, message=None, headers=None): text = body_text(body) return { "kind": kind, "statusCode": status, "message": str(message) if message else None, "openaiError": openai_error_fields(body), "body": redact_diagnostic(body) if isinstance(body, (dict, list)) else None, "bodyHash": sha(text), "bodyPreview": preview(text, 1000), "headers": selected_headers(headers), } def sub2api_style_input(prompt): return [{ "role": "user", "content": [{ "type": "input_text", "text": prompt, }], }] def sub2api_style_instructions(): return ( "You are Codex, based on GPT-5. You are running as a coding agent in the Codex CLI on a user's computer." ) def event_type(event): if isinstance(event, dict): return event.get("type") return getattr(event, "type", None) def event_delta(event): if isinstance(event, dict): value = event.get("delta") return value if isinstance(value, str) else "" value = getattr(event, "delta", "") return value if isinstance(value, str) else "" def event_error_message(event): data = model_dump(event) if isinstance(data, dict): if isinstance(data.get("error"), dict): message = data["error"].get("message") if isinstance(message, str) and message: return message if isinstance(data.get("response"), dict) and isinstance(data["response"].get("error"), dict): message = data["response"]["error"].get("message") if isinstance(message, str) and message: return message return None def openai_responses_create(profile, config, marker, prompt): headers = { "User-Agent": profile.get("upstreamUserAgent") or config["probe"].get("userAgent") or "Go-http-client/1.1", "X-Request-ID": "unidesk-account-sentinel-" + hashlib.sha256(marker.encode()).hexdigest()[:16], } client = OpenAI( api_key=profile["apiKey"], base_url=upstream_base_url(profile["baseUrl"]), timeout=float(config["probe"]["timeoutSeconds"]), max_retries=0, ) started = time.time() try: stream = client.responses.create( model=config["model"], input=sub2api_style_input(prompt), instructions=sub2api_style_instructions(), stream=True, extra_headers=headers, ) deltas = [] events = [] seen_completed = False max_chars = max(32, int(config["probe"]["maxOutputTokens"]) * 12) for event in stream: event_data = model_dump(event) etype = event_type(event) events.append({"type": etype, "preview": preview(body_text(event_data), 240)}) if etype == "response.output_text.delta": delta = event_delta(event) if delta: deltas.append(delta) if len("".join(deltas)) > max_chars: break elif etype in ("response.completed", "response.done"): seen_completed = True break elif etype in ("response.failed", "error"): message = event_error_message(event) or "OpenAI response failed" raise RuntimeError(message) out = "".join(deltas) parsed = {"stream": True, "completed": seen_completed, "events": events[-20:], "output_text": out} if not seen_completed: parsed["streamError"] = "stream ended before response.completed" return { "ok": seen_completed, "status": 200 if seen_completed else 0, "json": parsed, "outputText": out, "text": body_text(parsed), "tooLarge": not seen_completed and len(out) > max_chars, "durationMs": int((time.time() - started) * 1000), "sdk": "openai-python", "requestShape": "sub2api-account-test-streaming-responses", } except APIStatusError as exc: status = getattr(exc, "status_code", 0) or 0 body = getattr(exc, "body", None) response = getattr(exc, "response", None) return { "ok": False, "status": status, "json": body if isinstance(body, dict) else None, "text": body_text(body or response or ""), "tooLarge": False, "durationMs": int((time.time() - started) * 1000), "error": str(exc), "errorDetails": error_details("APIStatusError", status, body, str(exc), getattr(response, "headers", None)), "sdk": "openai-python", "requestShape": "sub2api-account-test-streaming-responses", } except (APITimeoutError, APIConnectionError) as exc: return { "ok": False, "status": 0, "json": None, "text": "", "tooLarge": False, "durationMs": int((time.time() - started) * 1000), "error": str(exc), "errorDetails": error_details(exc.__class__.__name__, 0, None, str(exc), None), "sdk": "openai-python", "requestShape": "sub2api-account-test-streaming-responses", } except Exception as exc: return { "ok": False, "status": 0, "json": None, "text": "", "tooLarge": False, "durationMs": int((time.time() - started) * 1000), "error": str(exc), "errorDetails": error_details(exc.__class__.__name__, 0, None, str(exc), None), "sdk": "openai-python", "requestShape": "sub2api-account-test-streaming-responses", } def usage_from(parsed, prompt, out, config): usage = parsed.get("usage") if isinstance(parsed, dict) and isinstance(parsed.get("usage"), dict) else {} input_tokens = usage.get("input_tokens") output_tokens = usage.get("output_tokens") estimated = False if not isinstance(input_tokens, int): input_tokens = estimate_tokens(prompt) estimated = True if not isinstance(output_tokens, int): output_tokens = estimate_tokens(out) estimated = True total = usage.get("total_tokens") if not isinstance(total, int): total = input_tokens + output_tokens estimated = True cost = ( input_tokens * float(config["pricing"]["usdPer1MInputTokens"]) + output_tokens * float(config["pricing"]["usdPer1MOutputTokens"]) ) / 1000000 return { "inputTokens": input_tokens, "outputTokens": output_tokens, "totalTokens": total, "estimated": estimated, "estimatedCostUsd": cost, } def probe_account(profile, config, purpose): marker = config["marker"]["prefix"] + "_" + hashlib.sha256((profile["accountName"] + str(time.time()) + str(random.random())).encode()).hexdigest()[:10] prompt = "Return exactly this marker and no other text: " + marker resp = openai_responses_create(profile, config, marker, prompt) parsed = resp.get("json") out = resp.get("outputText") if isinstance(resp.get("outputText"), str) else output_text(parsed) trimmed = out.strip() marker_matched = trimmed == marker if config["marker"].get("exact", True) else marker in trimmed usage = usage_from(parsed if isinstance(parsed, dict) else {}, prompt, out or resp.get("text", ""), config) http_success = isinstance(resp.get("status"), int) and 200 <= resp.get("status") < 300 ok = marker_matched mismatch = not marker_matched if marker_matched: failure_kind = "none" elif resp.get("tooLarge"): failure_kind = "response-too-large" elif not resp["ok"]: failure_kind = "transport-or-http-failure" elif http_success: failure_kind = "success-body-mismatch" else: failure_kind = "unknown-marker-mismatch" return { "accountName": profile["accountName"], "profile": profile.get("profile"), "purpose": purpose, "ok": ok, "markerMatched": marker_matched, "markerHash": sha(marker), "httpStatus": resp.get("status"), "transportOk": resp["ok"], "tooLarge": resp.get("tooLarge"), "durationMs": resp.get("durationMs"), "outputHash": sha(out), "outputPreview": "" if marker_matched else preview(out or resp.get("text", ""), 160), "responseBodyHash": sha(resp.get("text", "")), "responseBodyPreview": "" if marker_matched else preview(resp.get("text", ""), 1000), "error": resp.get("error"), "errorDetails": resp.get("errorDetails"), "usage": usage, "mismatch": mismatch, "transportFailure": not resp["ok"], "failureKind": failure_kind, "sdk": resp.get("sdk"), "requestShape": resp.get("requestShape"), } def ledger_for(state, now): day = day_key(now) ledger = state.setdefault("ledger", {}).setdefault(day, {"inputTokens": 0, "outputTokens": 0, "totalTokens": 0, "estimatedCostUsd": 0, "requestCount": 0}) return day, ledger def account_day(account_state, day): return account_state.setdefault("daily", {}).setdefault(day, {"inputTokens": 0, "outputTokens": 0, "totalTokens": 0, "estimatedCostUsd": 0, "requestCount": 0}) def add_usage(state, account_state, now, usage): day, ledger = ledger_for(state, now) daily = account_day(account_state, day) for target in (ledger, daily): target["inputTokens"] = int(target.get("inputTokens") or 0) + int(usage.get("inputTokens") or 0) target["outputTokens"] = int(target.get("outputTokens") or 0) + int(usage.get("outputTokens") or 0) target["totalTokens"] = int(target.get("totalTokens") or 0) + int(usage.get("totalTokens") or 0) target["estimatedCostUsd"] = round(float(target.get("estimatedCostUsd") or 0) + float(usage.get("estimatedCostUsd") or 0), 8) target["requestCount"] = int(target.get("requestCount") or 0) + 1 def due_time(account_state): quarantine = account_state.get("quarantine") if isinstance(quarantine, dict) and quarantine.get("active") is True: return parse_iso(quarantine.get("until")) return parse_iso(account_state.get("nextProbeAfter")) def choose_due_profiles(profiles, state, config, now): day, ledger = ledger_for(state, now) due = [] accounts = state.setdefault("accounts", {}) for profile in profiles: name = profile["accountName"] account_state = accounts.setdefault(name, {}) when = due_time(account_state) if when is None or when <= now: quarantine = account_state.get("quarantine") purpose = "recovery" if isinstance(quarantine, dict) and quarantine.get("active") is True else "health" due.append({"profile": profile, "purpose": purpose, "dueAt": iso(when) if when else None}) due.sort(key=lambda item: item["dueAt"] or "") return due, {"selected": len(due), "due": len(due), "limit": "all-due", "budgetMode": "record-only", "ledger": ledger} def forced_account_names(): raw = os.environ.get("SENTINEL_ACCOUNT_NAMES") or "" names = [item.strip() for item in raw.split(",") if item.strip()] return set(names) def choose_forced_profiles(profiles, state, config, now, names): accounts = state.setdefault("accounts", {}) found = [] missing = sorted(names) due = [] for profile in profiles: name = profile["accountName"] if name not in names: continue account_state = accounts.setdefault(name, {}) quarantine = account_state.get("quarantine") purpose = "manual-recovery" if isinstance(quarantine, dict) and quarantine.get("active") is True else "manual-health" due.append({"profile": profile, "purpose": purpose, "dueAt": "forced"}) found.append(name) missing = sorted(name for name in names if name not in set(found)) return due, {"selected": len(due), "due": len(due), "limit": "forced-accounts", "budgetMode": "record-only", "ledger": ledger_for(state, now)[1], "requestedAccounts": sorted(names), "missingAccounts": missing} def next_success_interval(account_state, config): streak = int(account_state.get("successStreak") or 0) previous = int(account_state.get("successIntervalMinutes") or 0) initial = int(config["cadence"]["successInitialIntervalMinutes"]) maximum = int(config["cadence"]["successMaxIntervalMinutes"]) multiplier = int(config["cadence"]["successBackoffMultiplier"]) return initial if streak <= 0 or previous <= 0 else min(maximum, max(initial, previous * multiplier)) def next_freeze_interval(account_state, config, was_recovery): quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else {} previous = int(quarantine.get("intervalMinutes") or 0) initial = int(config["freeze"]["initialTtlMinutes"]) maximum = int(config["freeze"]["maxTtlMinutes"]) multiplier = int(config["freeze"]["backoffMultiplier"]) if was_recovery and previous > 0: return min(maximum, max(initial, previous * multiplier)) return initial def apply_result(result, state, config, now, admin): name = result["accountName"] account_state = state.setdefault("accounts", {}).setdefault(name, {}) add_usage(state, account_state, now, result.get("usage") or {}) actions_enabled = bool(config["actions"]["enabled"]) quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else None was_recovery = bool(quarantine and quarantine.get("active") is True) action = {"taken": False, "type": None} if result.get("ok") is True: if was_recovery: if actions_enabled and quarantine.get("applied") is True: try: action = {"taken": True, "type": "restore", "result": admin.set_schedulable(name, True), "recoverState": admin.recover_state(name)} except Exception as exc: action = {"taken": False, "type": "restore-failed", "error": str(exc)} account_state["quarantine"] = {"active": False, "clearedAt": iso(now), "lastApplied": quarantine.get("applied") is True} quality_gate = account_state.get("qualityGate") if isinstance(account_state.get("qualityGate"), dict) else None if quality_gate and quality_gate.get("pending") is True: account_state["qualityGate"] = {**quality_gate, "pending": False, "clearedAt": iso(now)} account_state["successStreak"] = 0 account_state["successIntervalMinutes"] = 0 interval = next_success_interval(account_state, config) account_state["successStreak"] = int(account_state.get("successStreak") or 0) + 1 account_state["successIntervalMinutes"] = interval account_state["nextProbeAfter"] = iso(add_minutes(now, interval, int(config["cadence"]["jitterPercent"]))) account_state["lastOkAt"] = iso(now) account_state["lastStatus"] = "ok" else: should_freeze = result.get("markerMatched") is not True if should_freeze: interval = next_freeze_interval(account_state, config, was_recovery) until = add_minutes(now, interval, int(config["freeze"]["jitterPercent"])) applied = False if actions_enabled: try: action = {"taken": True, "type": "freeze", "result": admin.set_schedulable(name, False)} applied = True except Exception as exc: action = {"taken": False, "type": "freeze-failed", "error": str(exc)} else: action = {"taken": False, "type": "would-freeze"} account_state["quarantine"] = { "active": True, "applied": applied, "until": iso(until), "intervalMinutes": interval, "reason": "marker-not-matched", "failureKind": result.get("failureKind"), "markerHash": result.get("markerHash"), "outputHash": result.get("outputHash"), "responseBodyHash": result.get("responseBodyHash"), "errorDetails": result.get("errorDetails"), "lastBadAt": iso(now), } account_state["nextProbeAfter"] = iso(until) account_state["successStreak"] = 0 account_state["successIntervalMinutes"] = 0 account_state["lastStatus"] = "quarantined" else: retry = int(config["probe"]["transportRetryMinutes"]) account_state["nextProbeAfter"] = iso(add_minutes(now, retry, int(config["cadence"]["jitterPercent"]))) account_state["lastStatus"] = "marker-not-matched-no-freeze" account_state["lastFailureAt"] = iso(now) account_state["lastProbeAt"] = iso(now) account_state["lastProbe"] = { "ok": result.get("ok"), "purpose": result.get("purpose"), "httpStatus": result.get("httpStatus"), "durationMs": result.get("durationMs"), "markerMatched": result.get("markerMatched"), "transportOk": result.get("transportOk"), "outputHash": result.get("outputHash"), "outputPreview": result.get("outputPreview"), "responseBodyHash": result.get("responseBodyHash"), "responseBodyPreview": result.get("responseBodyPreview"), "error": result.get("error"), "errorDetails": result.get("errorDetails"), "usage": result.get("usage"), "failureKind": result.get("failureKind"), "sdk": result.get("sdk"), "requestShape": result.get("requestShape"), "action": action, } return action def reconcile_active_quarantines(state, config, now, admin): actions = [] if not config["actions"]["enabled"]: return actions for name, account_state in state.setdefault("accounts", {}).items(): quarantine = account_state.get("quarantine") if not isinstance(quarantine, dict) or quarantine.get("active") is not True: continue until = parse_iso(quarantine.get("until")) if until is not None and until <= now: continue if quarantine.get("applied") is not True: try: admin.set_schedulable(name, False) quarantine["applied"] = True quarantine["appliedAt"] = iso(now) actions.append({"accountName": name, "type": "apply-pending-freeze", "ok": True}) except Exception as exc: actions.append({"accountName": name, "type": "apply-pending-freeze", "ok": False, "error": str(exc)}) continue try: admin.set_schedulable(name, False) actions.append({"accountName": name, "type": "reassert-freeze", "ok": True}) except Exception as exc: actions.append({"accountName": name, "type": "reassert-freeze", "ok": False, "error": str(exc)}) return actions def main(): now = utc_now() config = load_json(CONFIG_PATH) profiles = load_json(PROFILES_PATH).get("profiles") or [] namespace = os.environ.get("POD_NAMESPACE") or "platform-infra" kube = KubeClient(namespace) state_obj, state = load_state(kube, config) admin = Sub2ApiAdmin(config) reconcile = reconcile_active_quarantines(state, config, now, admin) forced_names = forced_account_names() if forced_names: due, selection = choose_forced_profiles(profiles, state, config, now, forced_names) else: due, selection = choose_due_profiles(profiles, state, config, now) results = [] actions = [] if (config["monitor"]["enabled"] or forced_names) and due: with ThreadPoolExecutor(max_workers=max(1, len(due))) as executor: futures = [executor.submit(probe_account, item["profile"], config, item["purpose"]) for item in due] for future in as_completed(futures): result = future.result() results.append(result) actions.append({"accountName": result["accountName"], **apply_result(result, state, config, now, admin)}) history = state.setdefault("history", []) run_summary = { "at": iso(now), "monitorEnabled": bool(config["monitor"]["enabled"]), "actionsEnabled": bool(config["actions"]["enabled"]), "profileCount": len(profiles), "selected": len(due), "okCount": sum(1 for item in results if item.get("ok") is True), "mismatchCount": sum(1 for item in results if item.get("markerMatched") is not True), "markerMismatchCount": sum(1 for item in results if item.get("markerMatched") is not True), "transportFailureCount": sum(1 for item in results if item.get("transportFailure") is True), "actionsTaken": sum(1 for item in actions if item.get("taken") is True), "selection": selection, "reconcile": reconcile[-20:], } history.append(run_summary) del history[:-int(config["state"]["historyLimit"])] state["lastRun"] = run_summary kube.update_configmap_state(state_obj, state) print(json.dumps({ "ok": True, "summary": run_summary, "results": [{ "accountName": item.get("accountName"), "purpose": item.get("purpose"), "ok": item.get("ok"), "markerMatched": item.get("markerMatched"), "httpStatus": item.get("httpStatus"), "durationMs": item.get("durationMs"), "usage": item.get("usage"), "outputHash": item.get("outputHash"), "outputPreview": item.get("outputPreview"), "responseBodyHash": item.get("responseBodyHash"), "responseBodyPreview": item.get("responseBodyPreview"), "error": item.get("error"), "errorDetails": item.get("errorDetails"), "failureKind": item.get("failureKind"), "sdk": item.get("sdk"), "requestShape": item.get("requestShape"), } for item in results], "actions": actions, "valuesPrinted": False, }, ensure_ascii=False)) if __name__ == "__main__": try: main() except Exception as exc: print(json.dumps({ "ok": False, "error": str(exc), "traceback": traceback.format_exc()[-4000:], "valuesPrinted": False, }, ensure_ascii=False)) raise `; } function indentBlock(value: string, spaces: number): string { const prefix = " ".repeat(spaces); return value.split("\n").map((line) => `${prefix}${line}`).join("\n"); } function valueAt(value: Record, key: string): unknown { return Object.prototype.hasOwnProperty.call(value, key) ? value[key] : undefined; } function readBoolean(value: unknown, key: string, fallback: boolean): boolean { if (value === undefined || value === null) return fallback; if (typeof value === "boolean") return value; if (typeof value === "string" && value.trim() === "true") return true; if (typeof value === "string" && value.trim() === "false") return false; throw new Error(`${key} must be a boolean`); } function readString(value: unknown, key: string, fallback: string): string { if (value === undefined || value === null) return fallback; if (typeof value !== "string" || value.trim().length === 0) throw new Error(`${key} must be a non-empty string`); if (/[\r\n]/u.test(value)) throw new Error(`${key} must not contain newlines`); return value.trim(); } function readDnsName(value: unknown, key: string, fallback: string): string { const text = readString(value, key, fallback); if (!/^[a-z0-9]([-a-z0-9]*[a-z0-9])?$/u.test(text)) throw new Error(`${key} must be a Kubernetes DNS label`); return text; } function readModelName(value: unknown, key: string, fallback: string): string { const text = readString(value, key, fallback); if (!/^[A-Za-z0-9._:-]+$/u.test(text)) throw new Error(`${key} has an unsupported model name`); return text; } function readMarkerPrefix(value: unknown, key: string, fallback: string): string { const text = readString(value, key, fallback); if (!/^[A-Za-z0-9_-]{2,32}$/u.test(text)) throw new Error(`${key} must be 2-32 chars of letters, digits, _ or -`); return text; } function readUserAgent(value: unknown, key: string, fallback: string): string { const text = readString(value, key, fallback); if (/[\r\n]/u.test(text)) throw new Error(`${key} must not contain newlines`); if (Buffer.byteLength(text, "utf8") > 200) throw new Error(`${key} must be at most 200 bytes`); return text; } function readOpenAiPythonVersion(value: unknown, key: string, fallback: string): string { const text = readString(value, key, fallback); if (!/^[0-9]+[.][0-9]+[.][0-9]+$/u.test(text)) throw new Error(`${key} must be a pinned semver version like 2.41.1`); return text; } function readInt(value: unknown, key: string, fallback: number, min: number, max: number): number { if (value === undefined || value === null) return fallback; const parsed = typeof value === "number" ? value : typeof value === "string" && value.trim() ? Number(value) : Number.NaN; if (!Number.isInteger(parsed) || parsed < min || parsed > max) throw new Error(`${key} must be an integer from ${min} to ${max}`); return parsed; } function readNumber(value: unknown, key: string, fallback: number, min: number, max: number): number { if (value === undefined || value === null) return fallback; const parsed = typeof value === "number" ? value : typeof value === "string" && value.trim() ? Number(value) : Number.NaN; if (!Number.isFinite(parsed) || parsed < min || parsed > max) throw new Error(`${key} must be a number from ${min} to ${max}`); return parsed; } function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); }