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; }; gatewayFailureMonitor: { enabled: boolean; lookbackSeconds: number; tailLines: number; initialTtlMinutes: number; maxTtlMinutes: number; backoffMultiplier: number; paths: string[]; }; sdk: { openaiPythonVersion: string; }; cadence: { successInitialIntervalMinutes: number; successMaxIntervalMinutes: number; trustedSuccessMaxIntervalMinutes: number; untrustedSuccessMaxIntervalMinutes: 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; trustUpstream: boolean; } export interface CodexPoolSentinelManifestOptions { namespace: string; serviceName: string; serviceDns: string; appSecretName: string; proxy?: { httpProxy: string; noProxy: string; } | null; } 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", }, gatewayFailureMonitor: { enabled: false, lookbackSeconds: 900, tailLines: 4000, initialTtlMinutes: 5, maxTtlMinutes: 30, backoffMultiplier: 2, paths: ["/responses", "/v1/responses", "/responses/compact", "/v1/responses/compact"], }, sdk: { openaiPythonVersion: "2.41.1", }, cadence: { successInitialIntervalMinutes: 1, successMaxIntervalMinutes: 20, trustedSuccessMaxIntervalMinutes: 20, untrustedSuccessMaxIntervalMinutes: 2, 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 gatewayFailureMonitor = isRecord(value.gatewayFailureMonitor) ? value.gatewayFailureMonitor : {}; 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 legacySuccessMax = readInt(valueAt(cadence, "successMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.successMaxIntervalMinutes`, defaults.cadence.successMaxIntervalMinutes, 1, 1440); const trustedSuccessMax = readInt(valueAt(cadence, "trustedSuccessMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.trustedSuccessMaxIntervalMinutes`, legacySuccessMax, 1, 1440); const untrustedSuccessMax = readInt(valueAt(cadence, "untrustedSuccessMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.untrustedSuccessMaxIntervalMinutes`, legacySuccessMax, 1, 1440); 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), }, gatewayFailureMonitor: { enabled: readBoolean(valueAt(gatewayFailureMonitor, "enabled"), `${sourcePath}.sentinel.gatewayFailureMonitor.enabled`, defaults.gatewayFailureMonitor.enabled), lookbackSeconds: readInt(valueAt(gatewayFailureMonitor, "lookbackSeconds"), `${sourcePath}.sentinel.gatewayFailureMonitor.lookbackSeconds`, defaults.gatewayFailureMonitor.lookbackSeconds, 60, 7200), tailLines: readInt(valueAt(gatewayFailureMonitor, "tailLines"), `${sourcePath}.sentinel.gatewayFailureMonitor.tailLines`, defaults.gatewayFailureMonitor.tailLines, 100, 50000), initialTtlMinutes: readInt(valueAt(gatewayFailureMonitor, "initialTtlMinutes"), `${sourcePath}.sentinel.gatewayFailureMonitor.initialTtlMinutes`, defaults.gatewayFailureMonitor.initialTtlMinutes, 1, 1440), maxTtlMinutes: readInt(valueAt(gatewayFailureMonitor, "maxTtlMinutes"), `${sourcePath}.sentinel.gatewayFailureMonitor.maxTtlMinutes`, defaults.gatewayFailureMonitor.maxTtlMinutes, 1, 1440), backoffMultiplier: readInt(valueAt(gatewayFailureMonitor, "backoffMultiplier"), `${sourcePath}.sentinel.gatewayFailureMonitor.backoffMultiplier`, defaults.gatewayFailureMonitor.backoffMultiplier, 1, 10), paths: readPathList(valueAt(gatewayFailureMonitor, "paths"), `${sourcePath}.sentinel.gatewayFailureMonitor.paths`, defaults.gatewayFailureMonitor.paths), }, 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: legacySuccessMax, trustedSuccessMaxIntervalMinutes: trustedSuccessMax, untrustedSuccessMaxIntervalMinutes: untrustedSuccessMax, 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.cadence.trustedSuccessMaxIntervalMinutes < config.cadence.successInitialIntervalMinutes) { throw new Error(`${sourcePath}.sentinel.cadence.trustedSuccessMaxIntervalMinutes must be >= successInitialIntervalMinutes`); } if (config.cadence.untrustedSuccessMaxIntervalMinutes < config.cadence.successInitialIntervalMinutes) { throw new Error(`${sourcePath}.sentinel.cadence.untrustedSuccessMaxIntervalMinutes must be >= successInitialIntervalMinutes`); } if (config.freeze.maxTtlMinutes < config.freeze.initialTtlMinutes) { throw new Error(`${sourcePath}.sentinel.freeze.maxTtlMinutes must be >= initialTtlMinutes`); } if (config.gatewayFailureMonitor.maxTtlMinutes < config.gatewayFailureMonitor.initialTtlMinutes) { throw new Error(`${sourcePath}.sentinel.gatewayFailureMonitor.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, gatewayFailureMonitor: config.gatewayFailureMonitor, 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, gatewayFailureMonitor: config.gatewayFailureMonitor, 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; const proxyEnv = options.proxy?.httpProxy ? ` - name: HTTP_PROXY value: ${JSON.stringify(options.proxy.httpProxy)} - name: HTTPS_PROXY value: ${JSON.stringify(options.proxy.httpProxy)} - name: ALL_PROXY value: ${JSON.stringify(options.proxy.httpProxy)} - name: http_proxy value: ${JSON.stringify(options.proxy.httpProxy)} - name: https_proxy value: ${JSON.stringify(options.proxy.httpProxy)} - name: all_proxy value: ${JSON.stringify(options.proxy.httpProxy)} - name: NO_PROXY value: ${JSON.stringify(options.proxy.noProxy)} - name: no_proxy value: ${JSON.stringify(options.proxy.noProxy)} ` : ""; 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"] - apiGroups: [""] resources: ["pods"] verbs: ["get", "list"] - apiGroups: [""] resources: ["pods/log"] verbs: ["get"] --- 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 ${proxyEnv} 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() if not raw: return resp.status, None text = raw.decode("utf-8", errors="replace") try: return resp.status, json.loads(text) except Exception: return resp.status, text 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 list_pods(self, label_selector): query = parse.urlencode({"labelSelector": label_selector}) status, data = self.api("GET", f"/api/v1/namespaces/{self.namespace}/pods?{query}") if status >= 300: raise RuntimeError(f"list pods failed: {status} {data}") return data.get("items") if isinstance(data, dict) and isinstance(data.get("items"), list) else [] def pod_logs(self, pod_name, container, since_seconds, tail_lines): query = parse.urlencode({ "container": container, "sinceSeconds": int(since_seconds), "tailLines": int(tail_lines), }) status, data = self.api("GET", f"/api/v1/namespaces/{self.namespace}/pods/{url_quote(pod_name)}/log?{query}") if status >= 300: raise RuntimeError(f"get pod logs failed: {status} {data}") return data if isinstance(data, str) else "" 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 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"), "trustUpstream": profile.get("trustUpstream") is True, "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 success_max_interval(profile, config): cadence = config["cadence"] if profile.get("trustUpstream") is True: return int(cadence.get("trustedSuccessMaxIntervalMinutes") or cadence.get("successMaxIntervalMinutes")) return int(cadence.get("untrustedSuccessMaxIntervalMinutes") or cadence.get("successMaxIntervalMinutes")) def next_success_interval(account_state, config, profile): streak = int(account_state.get("successStreak") or 0) previous = int(account_state.get("successIntervalMinutes") or 0) initial = int(config["cadence"]["successInitialIntervalMinutes"]) maximum = success_max_interval(profile, config) 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, profile): 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: quality_gate = account_state.get("qualityGate") if isinstance(account_state.get("qualityGate"), dict) else None if was_recovery: if actions_enabled and quarantine.get("applied") is True: try: action = {"taken": True, "type": "restore", "result": admin.set_schedulable(name, True)} 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} account_state["successStreak"] = 0 account_state["successIntervalMinutes"] = 0 elif isinstance(quarantine, dict) and quarantine.get("active") is not True: account_state["quarantine"] = {"active": False, "clearedAt": iso(now), "lastApplied": quarantine.get("applied") is True} if quality_gate and quality_gate.get("pending") is True: account_state["qualityGate"] = {**quality_gate, "pending": False, "clearedAt": iso(now)} interval = next_success_interval(account_state, config, profile) account_state["successStreak"] = int(account_state.get("successStreak") or 0) + 1 account_state["successIntervalMinutes"] = interval account_state["successMaxIntervalMinutes"] = success_max_interval(profile, config) 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["successMaxIntervalMinutes"] = success_max_interval(profile, config) 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["trustUpstream"] = profile.get("trustUpstream") is True account_state["lastProbe"] = { "ok": result.get("ok"), "purpose": result.get("purpose"), "trustUpstream": result.get("trustUpstream"), "successMaxIntervalMinutes": success_max_interval(profile, config), "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 log_line_payload(line): pos = line.find("{") if pos < 0: return None, "", None prefix = line[:pos].rstrip("\t ") parts = prefix.split("\t") ts = parts[0] if parts else "" message = parts[3] if len(parts) >= 4 else "" try: return ts, message, json.loads(line[pos:]) except Exception: return ts, message, None def gateway_monitor_paths(config): cfg = config.get("gatewayFailureMonitor") if isinstance(config.get("gatewayFailureMonitor"), dict) else {} paths = cfg.get("paths") if isinstance(paths, list) and paths: return set(str(item) for item in paths if isinstance(item, str) and item) return {"/responses", "/v1/responses", "/responses/compact", "/v1/responses/compact"} def gateway_failure_kind(message, payload, config): if not isinstance(payload, dict): return None path = payload.get("path") if path not in gateway_monitor_paths(config): return None if payload.get("account_id") is None: return None if "codex.remote_compact.failed" in message: status = payload.get("status_code") if isinstance(status, int) and status >= 500: return "gateway-compact-final-failure" return None if "openai.upstream_failover_switching" in message and path in ("/responses/compact", "/v1/responses/compact"): upstream_status = payload.get("upstream_status") if isinstance(upstream_status, int) and upstream_status >= 500: return "gateway-compact-upstream-failover" return None if "openai.forward_failed" not in message: return None error_text = str(payload.get("error") or "").lower() fallback_written = payload.get("fallback_error_response_written") is True upstream_already_written = payload.get("upstream_error_response_already_written") is True stream_failure = any(token in error_text for token in ( "stream usage incomplete", "missing terminal event", "stream read error", "stream data interval timeout", )) if fallback_written or upstream_already_written or stream_failure: return "gateway-stream-forward-failure" session_affinity_failure = any(token in error_text for token in ( "encrypted content could not be decrypted", "could not be verified", "invalid_encrypted_content", )) if session_affinity_failure: return "gateway-session-affinity-failure" final_compatibility_failure = any(token in error_text for token in ( "bad_response_status_code", "model_not_found", "no available channel for model", "unsupported model", "not support", "not supported", "payload too large", "request too large", "context length", "context window", "maximum context", )) if final_compatibility_failure: return "gateway-final-compatibility-failure" final_5xx_failure = any(token in error_text for token in ( "upstream error: 500", "upstream error: 502", "upstream error: 503", "upstream error: 504", "upstream error: 524", "gateway timeout", "bad gateway", "upstream request failed", "unknown error", "context deadline exceeded", "context canceled", )) if final_5xx_failure: return "gateway-final-transient-failure" return None def gateway_failure_is_observe_only(failure_kind): return failure_kind in {"gateway-session-affinity-failure", "gateway-compact-final-failure", "gateway-compact-upstream-failover"} def gateway_failure_item(ts, pod_name, payload, failure_kind): request_id = payload.get("request_id") or sha(json.dumps(payload, sort_keys=True, ensure_ascii=False)) try: account_id = int(payload.get("account_id")) except Exception: account_id = None return { "at": ts, "pod": pod_name, "requestId": request_id, "clientRequestId": payload.get("client_request_id"), "accountId": account_id, "failureKind": failure_kind, "path": payload.get("path"), "errorPreview": preview(payload.get("error"), 240), "fallbackErrorResponseWritten": payload.get("fallback_error_response_written") is True, "upstreamErrorResponseAlreadyWritten": payload.get("upstream_error_response_already_written") is True, "bodyBytes": payload.get("body_bytes"), "latencyMs": payload.get("latency_ms"), "statusCode": payload.get("status_code"), "upstreamStatus": payload.get("upstream_status"), } def trim_gateway_seen(monitor_state, now, lookback_seconds): seen = monitor_state.setdefault("seenRequestIds", {}) if not isinstance(seen, dict): seen = {} monitor_state["seenRequestIds"] = seen cutoff = now - timedelta(seconds=max(int(lookback_seconds) * 4, 3600)) for request_id, seen_at in list(seen.items()): parsed = parse_iso(seen_at) if parsed is None or parsed < cutoff: seen.pop(request_id, None) return seen def gateway_failure_account_map(admin): by_id = {} for name, account in admin.accounts().items(): try: account_id = int(account.get("id")) except Exception: continue by_id[account_id] = name return by_id def next_gateway_freeze_interval(account_state, config): monitor_cfg = config.get("gatewayFailureMonitor") if isinstance(config.get("gatewayFailureMonitor"), dict) else {} initial = int(monitor_cfg.get("initialTtlMinutes") or 5) maximum = int(monitor_cfg.get("maxTtlMinutes") or 30) multiplier = int(monitor_cfg.get("backoffMultiplier") or 2) previous = int(account_state.get("gatewayFailureBackoffIntervalMinutes") or 0) quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else {} if quarantine.get("reason") == "gateway-forward-failure": previous = max(previous, int(quarantine.get("intervalMinutes") or 0)) if previous <= 0: return initial return min(maximum, max(initial, previous * multiplier)) def apply_gateway_failure(account_name, failures, state, config, now, admin, profile): latest = failures[-1] account_state = state.setdefault("accounts", {}).setdefault(account_name, {}) interval = next_gateway_freeze_interval(account_state, config) until = add_minutes(now, interval, int(config["freeze"]["jitterPercent"])) actions_enabled = bool(config["actions"]["enabled"]) applied = False action = {"taken": False, "type": "would-freeze"} if actions_enabled: try: action = {"taken": True, "type": "freeze", "result": admin.set_schedulable(account_name, False)} applied = True except Exception as exc: action = {"taken": False, "type": "freeze-failed", "error": str(exc)} account_state["quarantine"] = { "active": True, "applied": applied, "until": iso(until), "intervalMinutes": interval, "reason": "gateway-forward-failure", "failureKind": latest.get("failureKind") or "gateway-forward-failure", "errorDetails": { "kind": "Sub2APIGatewayForwardFailure", "requestId": latest.get("requestId"), "clientRequestId": latest.get("clientRequestId"), "failureKind": latest.get("failureKind"), "path": latest.get("path"), "errorPreview": latest.get("errorPreview"), "fallbackErrorResponseWritten": latest.get("fallbackErrorResponseWritten"), "upstreamErrorResponseAlreadyWritten": latest.get("upstreamErrorResponseAlreadyWritten"), "bodyBytes": latest.get("bodyBytes"), "latencyMs": latest.get("latencyMs"), "countInRun": len(failures), }, "lastBadAt": iso(now), } account_state["nextProbeAfter"] = iso(until) account_state["successStreak"] = 0 account_state["successIntervalMinutes"] = 0 account_state["successMaxIntervalMinutes"] = success_max_interval(profile, config) account_state["lastStatus"] = "quarantined" account_state["lastFailureAt"] = iso(now) account_state["lastGatewayFailureAt"] = iso(now) account_state["gatewayFailureBackoffIntervalMinutes"] = interval account_state["lastGatewayFailure"] = { "accountName": account_name, "accountId": latest.get("accountId"), "requestId": latest.get("requestId"), "clientRequestId": latest.get("clientRequestId"), "failureKind": latest.get("failureKind"), "path": latest.get("path"), "errorPreview": latest.get("errorPreview"), "fallbackErrorResponseWritten": latest.get("fallbackErrorResponseWritten"), "upstreamErrorResponseAlreadyWritten": latest.get("upstreamErrorResponseAlreadyWritten"), "bodyBytes": latest.get("bodyBytes"), "latencyMs": latest.get("latencyMs"), "countInRun": len(failures), "firstAt": failures[0].get("at"), "lastAt": latest.get("at"), "intervalMinutes": interval, "freezeUntil": iso(until), "action": action, } return action def record_gateway_observation(account_name, failures, state, now): latest = failures[-1] account_state = state.setdefault("accounts", {}).setdefault(account_name, {}) account_state["lastGatewayAffinityFailureAt"] = iso(now) account_state["lastGatewayAffinityFailure"] = { "accountName": account_name, "accountId": latest.get("accountId"), "requestId": latest.get("requestId"), "clientRequestId": latest.get("clientRequestId"), "failureKind": latest.get("failureKind"), "path": latest.get("path"), "errorPreview": latest.get("errorPreview"), "bodyBytes": latest.get("bodyBytes"), "latencyMs": latest.get("latencyMs"), "countInRun": len(failures), "firstAt": failures[0].get("at"), "lastAt": latest.get("at"), "action": { "taken": False, "type": "observe-session-affinity-failure", }, } return {"taken": False, "type": "observe-session-affinity-failure"} def run_gateway_failure_monitor(state, config, now, kube, admin, profiles): cfg = config.get("gatewayFailureMonitor") if isinstance(config.get("gatewayFailureMonitor"), dict) else {} if cfg.get("enabled") is not True: return {"enabled": False, "scanned": 0, "newFailures": 0, "actions": []} lookback_seconds = int(cfg.get("lookbackSeconds") or 900) tail_lines = int(cfg.get("tailLines") or 4000) monitor_state = state.setdefault("gatewayFailureMonitor", {}) if not isinstance(monitor_state, dict): monitor_state = {} state["gatewayFailureMonitor"] = monitor_state seen = trim_gateway_seen(monitor_state, now, lookback_seconds) pods = kube.list_pods("app.kubernetes.io/name=sub2api") candidates = [] log_errors = [] for pod in pods: metadata = pod.get("metadata") if isinstance(pod, dict) else {} status = pod.get("status") if isinstance(pod, dict) else {} pod_name = metadata.get("name") if not isinstance(pod_name, str) or not pod_name: continue if status.get("phase") not in (None, "Running"): continue try: logs = kube.pod_logs(pod_name, "sub2api", lookback_seconds, tail_lines) except Exception as exc: log_errors.append({"pod": pod_name, "error": str(exc)}) continue for line in str(logs).splitlines(): ts, message, payload = log_line_payload(line) failure_kind = gateway_failure_kind(message, payload, config) if failure_kind is not None: candidates.append(gateway_failure_item(ts, pod_name, payload, failure_kind)) by_id = gateway_failure_account_map(admin) if candidates else {} profile_by_name = {item.get("accountName"): item for item in profiles if isinstance(item, dict) and isinstance(item.get("accountName"), str)} by_account = {} skipped = [] new_failures = [] for item in candidates: request_id = item.get("requestId") if not isinstance(request_id, str) or not request_id: request_id = sha(json.dumps(item, sort_keys=True, ensure_ascii=False)) item["requestId"] = request_id if request_id in seen: continue seen[request_id] = iso(now) account_name = by_id.get(item.get("accountId")) if not account_name: skipped.append({"requestId": request_id, "accountId": item.get("accountId"), "reason": "account-id-not-managed"}) continue if account_name not in profile_by_name: skipped.append({"requestId": request_id, "accountId": item.get("accountId"), "accountName": account_name, "reason": "account-not-in-profiles"}) continue item["accountName"] = account_name by_account.setdefault(account_name, []).append(item) new_failures.append(item) actions = [] for account_name, failures in sorted(by_account.items()): failures.sort(key=lambda item: item.get("at") or "") profile = profile_by_name[account_name] freeze_failures = [item for item in failures if not gateway_failure_is_observe_only(item.get("failureKind"))] observed_failures = [item for item in failures if gateway_failure_is_observe_only(item.get("failureKind"))] if freeze_failures: action = apply_gateway_failure(account_name, freeze_failures, state, config, now, admin, profile) actions.append({ "accountName": account_name, "accountId": freeze_failures[-1].get("accountId"), "failureCount": len(freeze_failures), "requestId": freeze_failures[-1].get("requestId"), "failureKind": freeze_failures[-1].get("failureKind"), "path": freeze_failures[-1].get("path"), "errorPreview": freeze_failures[-1].get("errorPreview"), "taken": action.get("taken"), "type": action.get("type"), "error": action.get("error"), }) if observed_failures: action = record_gateway_observation(account_name, observed_failures, state, now) actions.append({ "accountName": account_name, "accountId": observed_failures[-1].get("accountId"), "failureCount": len(observed_failures), "requestId": observed_failures[-1].get("requestId"), "failureKind": observed_failures[-1].get("failureKind"), "path": observed_failures[-1].get("path"), "errorPreview": observed_failures[-1].get("errorPreview"), "taken": action.get("taken"), "type": action.get("type"), "error": action.get("error"), }) monitor_state["lastRunAt"] = iso(now) monitor_state["lastScannedPods"] = [((pod.get("metadata") or {}).get("name")) for pod in pods if isinstance(pod, dict)] monitor_state["lastCandidateCount"] = len(candidates) monitor_state["lastNewFailureCount"] = len(new_failures) monitor_state["lastActionCount"] = len(actions) monitor_state["lastFailures"] = new_failures[-20:] monitor_state["lastSkipped"] = skipped[-20:] monitor_state["lastLogErrors"] = log_errors[-10:] return { "enabled": True, "lookbackSeconds": lookback_seconds, "tailLines": tail_lines, "scannedPods": len(pods), "candidates": len(candidates), "newFailures": len(new_failures), "actionsTaken": sum(1 for item in actions if item.get("taken") is True), "actions": actions[-20:], "skipped": skipped[-20:], "logErrors": log_errors[-10:], } 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() gateway_monitor = {"enabled": False, "skipped": "forced-manual-probe"} if forced_names else run_gateway_failure_monitor(state, config, now, kube, admin, profiles) 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"]): item["profile"] for item in due} for future in as_completed(futures): result = future.result() results.append(result) profile = futures[future] actions.append({"accountName": result["accountName"], **apply_result(result, state, config, now, admin, profile)}) 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), "gatewayFailureMonitor": { "enabled": gateway_monitor.get("enabled") is True, "newFailures": gateway_monitor.get("newFailures", 0), "actionsTaken": gateway_monitor.get("actionsTaken", 0), "skipped": gateway_monitor.get("skipped"), "logErrors": gateway_monitor.get("logErrors"), }, "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"), "trustUpstream": item.get("trustUpstream"), "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, "gatewayFailureMonitor": gateway_monitor, "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 readPathList(value: unknown, key: string, fallback: string[]): string[] { if (value === undefined || value === null) return fallback; if (!Array.isArray(value) || value.length === 0) throw new Error(`${key} must be a non-empty string array`); const paths = value.map((item, index) => { if (typeof item !== "string" || item.trim().length === 0) throw new Error(`${key}[${index}] must be a non-empty string`); const path = item.trim(); if (!/^\/[A-Za-z0-9._~!$&'()*+,;=:@/-]*$/u.test(path)) throw new Error(`${key}[${index}] has an unsupported HTTP path`); return path; }); return [...new Set(paths)]; } 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); }