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;
  roleName: string;
  roleBindingName: 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;
  protectedManualAccounts?: string[];
}

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;
  sentinelProtect: CodexPoolSentinelProtectPolicy;
}

export interface CodexPoolSentinelProtectPolicy {
  enabled: boolean;
  consecutiveFailures: number;
  initialRetryDelaySeconds: number;
  maxRetryDelaySeconds: number;
  backoffMultiplier: number;
}

export interface CodexPoolSentinelManifestOptions {
  namespace: string;
  serviceName: string;
  serviceDns: string;
  appSecretName: string;
  adminEmailDefault: string;
  proxy?: {
    httpProxy: string;
    noProxy: string;
  } | null;
}

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, sourcePath: string): CodexPoolSentinelConfig {
  if (!isRecord(value)) throw new Error(`${sourcePath}.sentinel must be a YAML object`);
  const monitor = readRequiredRecord(valueAt(value, "monitor"), `${sourcePath}.sentinel.monitor`);
  const actions = readRequiredRecord(valueAt(value, "actions"), `${sourcePath}.sentinel.actions`);
  const marker = readRequiredRecord(valueAt(value, "marker"), `${sourcePath}.sentinel.marker`);
  const probe = readRequiredRecord(valueAt(value, "probe"), `${sourcePath}.sentinel.probe`);
  const gatewayFailureMonitor = readRequiredRecord(valueAt(value, "gatewayFailureMonitor"), `${sourcePath}.sentinel.gatewayFailureMonitor`);
  const sdk = readRequiredRecord(valueAt(value, "sdk"), `${sourcePath}.sentinel.sdk`);
  const cadence = readRequiredRecord(valueAt(value, "cadence"), `${sourcePath}.sentinel.cadence`);
  const freeze = readRequiredRecord(valueAt(value, "freeze"), `${sourcePath}.sentinel.freeze`);
  const pricing = readRequiredRecord(valueAt(value, "pricing"), `${sourcePath}.sentinel.pricing`);
  const config: CodexPoolSentinelConfig = {
    monitor: {
      enabled: readRequiredBoolean(valueAt(monitor, "enabled"), `${sourcePath}.sentinel.monitor.enabled`),
    },
    actions: {
      enabled: readRequiredBoolean(valueAt(actions, "enabled"), `${sourcePath}.sentinel.actions.enabled`),
    },
    schedule: readRequiredString(valueAt(value, "schedule"), `${sourcePath}.sentinel.schedule`),
    image: readRequiredImage(valueAt(value, "image"), `${sourcePath}.sentinel.image`),
    serviceAccountName: readRequiredDnsName(valueAt(value, "serviceAccountName"), `${sourcePath}.sentinel.serviceAccountName`),
    configMapName: readRequiredDnsName(valueAt(value, "configMapName"), `${sourcePath}.sentinel.configMapName`),
    credentialsSecretName: readRequiredDnsName(valueAt(value, "credentialsSecretName"), `${sourcePath}.sentinel.credentialsSecretName`),
    stateConfigMapName: readRequiredDnsName(valueAt(value, "stateConfigMapName"), `${sourcePath}.sentinel.stateConfigMapName`),
    cronJobName: readRequiredDnsName(valueAt(value, "cronJobName"), `${sourcePath}.sentinel.cronJobName`),
    roleName: readRequiredDnsName(valueAt(value, "roleName"), `${sourcePath}.sentinel.roleName`),
    roleBindingName: readRequiredDnsName(valueAt(value, "roleBindingName"), `${sourcePath}.sentinel.roleBindingName`),
    model: readRequiredModelName(valueAt(value, "model"), `${sourcePath}.sentinel.model`),
    endpoint: readRequiredEndpoint(valueAt(value, "endpoint"), `${sourcePath}.sentinel.endpoint`),
    marker: {
      prefix: readRequiredMarkerPrefix(valueAt(marker, "prefix"), `${sourcePath}.sentinel.marker.prefix`),
      exact: readRequiredBoolean(valueAt(marker, "exact"), `${sourcePath}.sentinel.marker.exact`),
    },
    probe: {
      timeoutSeconds: readRequiredInt(valueAt(probe, "timeoutSeconds"), `${sourcePath}.sentinel.probe.timeoutSeconds`, 3, 300),
      maxOutputTokens: readRequiredInt(valueAt(probe, "maxOutputTokens"), `${sourcePath}.sentinel.probe.maxOutputTokens`, 1, 128),
      transportRetryMinutes: readRequiredInt(valueAt(probe, "transportRetryMinutes"), `${sourcePath}.sentinel.probe.transportRetryMinutes`, 1, 120),
      userAgent: readRequiredUserAgent(valueAt(probe, "userAgent"), `${sourcePath}.sentinel.probe.userAgent`),
    },
    gatewayFailureMonitor: {
      enabled: readRequiredBoolean(valueAt(gatewayFailureMonitor, "enabled"), `${sourcePath}.sentinel.gatewayFailureMonitor.enabled`),
      lookbackSeconds: readRequiredInt(valueAt(gatewayFailureMonitor, "lookbackSeconds"), `${sourcePath}.sentinel.gatewayFailureMonitor.lookbackSeconds`, 60, 7200),
      tailLines: readRequiredInt(valueAt(gatewayFailureMonitor, "tailLines"), `${sourcePath}.sentinel.gatewayFailureMonitor.tailLines`, 100, 50000),
      initialTtlMinutes: readRequiredInt(valueAt(gatewayFailureMonitor, "initialTtlMinutes"), `${sourcePath}.sentinel.gatewayFailureMonitor.initialTtlMinutes`, 1, 1440),
      maxTtlMinutes: readRequiredInt(valueAt(gatewayFailureMonitor, "maxTtlMinutes"), `${sourcePath}.sentinel.gatewayFailureMonitor.maxTtlMinutes`, 1, 1440),
      backoffMultiplier: readRequiredInt(valueAt(gatewayFailureMonitor, "backoffMultiplier"), `${sourcePath}.sentinel.gatewayFailureMonitor.backoffMultiplier`, 1, 10),
      paths: readRequiredPathList(valueAt(gatewayFailureMonitor, "paths"), `${sourcePath}.sentinel.gatewayFailureMonitor.paths`),
    },
    sdk: {
      openaiPythonVersion: readRequiredOpenAiPythonVersion(valueAt(sdk, "openaiPythonVersion"), `${sourcePath}.sentinel.sdk.openaiPythonVersion`),
    },
    cadence: {
      successInitialIntervalMinutes: readRequiredInt(valueAt(cadence, "successInitialIntervalMinutes"), `${sourcePath}.sentinel.cadence.successInitialIntervalMinutes`, 1, 1440),
      successMaxIntervalMinutes: readRequiredInt(valueAt(cadence, "successMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.successMaxIntervalMinutes`, 1, 1440),
      trustedSuccessMaxIntervalMinutes: readRequiredInt(valueAt(cadence, "trustedSuccessMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.trustedSuccessMaxIntervalMinutes`, 1, 1440),
      untrustedSuccessMaxIntervalMinutes: readRequiredInt(valueAt(cadence, "untrustedSuccessMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.untrustedSuccessMaxIntervalMinutes`, 1, 1440),
      successBackoffMultiplier: readRequiredInt(valueAt(cadence, "successBackoffMultiplier"), `${sourcePath}.sentinel.cadence.successBackoffMultiplier`, 1, 10),
      jitterPercent: readRequiredInt(valueAt(cadence, "jitterPercent"), `${sourcePath}.sentinel.cadence.jitterPercent`, 0, 50),
    },
    freeze: {
      initialTtlMinutes: readRequiredInt(valueAt(freeze, "initialTtlMinutes"), `${sourcePath}.sentinel.freeze.initialTtlMinutes`, 1, 1440),
      maxTtlMinutes: readRequiredInt(valueAt(freeze, "maxTtlMinutes"), `${sourcePath}.sentinel.freeze.maxTtlMinutes`, 1, 1440),
      backoffMultiplier: readRequiredInt(valueAt(freeze, "backoffMultiplier"), `${sourcePath}.sentinel.freeze.backoffMultiplier`, 1, 10),
      jitterPercent: readRequiredInt(valueAt(freeze, "jitterPercent"), `${sourcePath}.sentinel.freeze.jitterPercent`, 0, 50),
    },
    pricing: {
      usdPer1MInputTokens: readRequiredNumber(valueAt(pricing, "usdPer1MInputTokens"), `${sourcePath}.sentinel.pricing.usdPer1MInputTokens`, 0, 100000),
      usdPer1MOutputTokens: readRequiredNumber(valueAt(pricing, "usdPer1MOutputTokens"), `${sourcePath}.sentinel.pricing.usdPer1MOutputTokens`, 0, 100000),
    },
    historyLimit: readRequiredInt(valueAt(value, "historyLimit"), `${sourcePath}.sentinel.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<string, unknown> {
  return {
    monitorEnabled: config.monitor.enabled,
    actionsEnabled: config.actions.enabled,
    schedule: config.schedule,
    cronJobName: config.cronJobName,
    roleName: config.roleName,
    roleBindingName: config.roleBindingName,
    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,
    protectedManualAccountCount: config.protectedManualAccounts?.length ?? 0,
    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: options.adminEmailDefault,
    },
    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,
    protectedManualAccounts: config.protectedManualAccounts ?? [],
    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.roleName}
  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.roleBindingName}
  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.roleName}
  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 ""
        protected = config.get("protectedManualAccounts") if isinstance(config.get("protectedManualAccounts"), list) else []
        self.protected_manual_accounts = set(str(item) for item in protected if isinstance(item, str) and item)
        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, max_bytes=2 * 1024 * 1024)
        if not resp["ok"]:
            parsed = resp.get("json")
            code = parsed.get("code") if isinstance(parsed, dict) else None
            message = parsed.get("message") if isinstance(parsed, dict) else None
            detail = f"status={resp.get('status')} tooLarge={resp.get('tooLarge')}"
            if code is not None:
                detail += f" code={code}"
            if message:
                detail += f" message={preview(message, 200)}"
            raise RuntimeError(f"admin {method} {path} failed: {detail}")
        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):
        if account_name in self.protected_manual_accounts:
            return {
                "accountId": None,
                "previousSchedulable": None,
                "schedulable": None,
                "skipped": True,
                "reason": "protected-manual-account",
            }
        account = self.account(account_name)
        if not account or account.get("id") is None:
            raise RuntimeError(f"account {account_name} not found")
        previous = account.get("schedulable")
        self.request("POST", f"/api/v1/admin/accounts/{account['id']}/schedulable", {"schedulable": bool(schedulable)})
        account["schedulable"] = bool(schedulable)
        if self.accounts_by_name is not None:
            self.accounts_by_name[account_name] = account
        return {"accountId": account.get("id"), "previousSchedulable": previous, "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 protect_policy(profile):
    policy = profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {}
    return policy if policy.get("enabled") is True else None

def protect_failure_threshold(policy):
    try:
        return max(1, int(policy.get("consecutiveFailures") or 1))
    except Exception:
        return 1

def protect_retry_delay_seconds(policy, retry_index):
    try:
        initial = max(1, int(policy.get("initialRetryDelaySeconds") or 2))
    except Exception:
        initial = 2
    try:
        maximum = max(initial, int(policy.get("maxRetryDelaySeconds") or initial))
    except Exception:
        maximum = initial
    try:
        multiplier = max(1, int(policy.get("backoffMultiplier") or 2))
    except Exception:
        multiplier = 2
    if retry_index <= 0:
        return 0
    return min(maximum, initial * (multiplier ** (retry_index - 1)))

def probe_account_with_protection(profile, config, purpose):
    policy = protect_policy(profile)
    if policy is None:
        return probe_account(profile, config, purpose)
    threshold = protect_failure_threshold(policy)
    attempts = []
    last_result = None
    for index in range(threshold):
        delay = protect_retry_delay_seconds(policy, index)
        if delay > 0:
            time.sleep(delay)
        attempt_purpose = purpose if index == 0 else purpose + "-protect-retry"
        result = probe_account(profile, config, attempt_purpose)
        attempt = {
            "attempt": index + 1,
            "delaySeconds": delay,
            "ok": result.get("ok"),
            "markerMatched": result.get("markerMatched"),
            "httpStatus": result.get("httpStatus"),
            "durationMs": result.get("durationMs"),
            "failureKind": result.get("failureKind"),
            "outputHash": result.get("outputHash"),
            "responseBodyHash": result.get("responseBodyHash"),
            "errorDetails": result.get("errorDetails"),
        }
        attempts.append(attempt)
        last_result = result
        if result.get("markerMatched") is True:
            result["sentinelProtect"] = {
                "enabled": True,
                "threshold": threshold,
                "attempts": attempts,
                "failureCount": index,
                "protected": index > 0,
                "decision": "pass",
            }
            if index > 0:
                result["purpose"] = purpose + "-protect-recovered"
            return result
    if last_result is None:
        last_result = probe_account(profile, config, purpose)
    last_result["sentinelProtect"] = {
        "enabled": True,
        "threshold": threshold,
        "attempts": attempts,
        "failureCount": len(attempts),
        "protected": False,
        "decision": "fail",
    }
    last_result["purpose"] = purpose + "-protect-exhausted"
    return last_result

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 runtime_temp_unschedulable_until(account):
    if not isinstance(account, dict):
        return None
    return account.get("temp_unschedulable_until") or account.get("tempUnschedulableUntil")

def runtime_temp_unschedulable_reason(account):
    if not isinstance(account, dict):
        return None
    return account.get("temp_unschedulable_reason") or account.get("tempUnschedulableReason")

def runtime_recovery_due(account_state):
    return account_state.get("runtimeSchedulable") is False

def sync_runtime_schedulable_state(state, profiles, now, admin):
    accounts_state = state.setdefault("accounts", {})
    synced_at = iso(now)
    summary = {
        "ok": False,
        "syncedAt": synced_at,
        "managedAccountCount": len(profiles),
        "schedulableAccounts": [],
        "unschedulableAccounts": [],
        "missingAccounts": [],
    }
    try:
        runtime_accounts = admin.accounts()
    except Exception as exc:
        error_text = str(exc)
        summary["error"] = error_text
        state["runtimeSchedulable"] = summary
        for profile in profiles:
            name = profile.get("accountName") if isinstance(profile, dict) else None
            if isinstance(name, str) and name:
                account_state = accounts_state.setdefault(name, {})
                account_state["runtimeSchedulable"] = None
                account_state["runtimeSyncedAt"] = synced_at
                account_state["runtimeSyncError"] = error_text
        return [{"type": "runtime-sync", "ok": False, "error": error_text}]
    for profile in profiles:
        name = profile.get("accountName") if isinstance(profile, dict) else None
        if not isinstance(name, str) or not name:
            continue
        account_state = accounts_state.setdefault(name, {})
        account = runtime_accounts.get(name)
        if not isinstance(account, dict):
            account_state["runtimeMissing"] = True
            account_state["runtimeAccountId"] = None
            account_state["runtimeStatus"] = None
            account_state["runtimeSchedulable"] = None
            account_state["runtimeTempUnschedulableUntil"] = None
            account_state["runtimeTempUnschedulableSet"] = None
            account_state["runtimeSyncedAt"] = synced_at
            account_state.pop("runtimeSyncError", None)
            summary["missingAccounts"].append(name)
            continue
        temp_until = runtime_temp_unschedulable_until(account)
        temp_reason = runtime_temp_unschedulable_reason(account)
        temp_set = temp_until is not None or bool(temp_reason)
        schedulable = account.get("schedulable")
        if not isinstance(schedulable, bool):
            schedulable = None
        account_state["runtimeMissing"] = False
        account_state["runtimeAccountId"] = account.get("id")
        account_state["runtimeStatus"] = account.get("status")
        account_state["runtimeSchedulable"] = schedulable
        account_state["runtimeTempUnschedulableUntil"] = temp_until
        account_state["runtimeTempUnschedulableSet"] = temp_set
        account_state["runtimeSyncedAt"] = synced_at
        account_state.pop("runtimeSyncError", None)
        if temp_reason:
            account_state["runtimeTempUnschedulableReasonHash"] = sha(temp_reason)
            account_state["runtimeTempUnschedulableReasonPreview"] = preview(temp_reason, 240)
        else:
            account_state.pop("runtimeTempUnschedulableReasonHash", None)
            account_state.pop("runtimeTempUnschedulableReasonPreview", None)
        if schedulable is True:
            summary["schedulableAccounts"].append(name)
        elif schedulable is False:
            summary["unschedulableAccounts"].append({
                "accountName": name,
                "status": account.get("status"),
                "tempUnschedulableSet": temp_set,
            })
    summary["ok"] = True
    state["runtimeSchedulable"] = summary
    return [{
        "type": "runtime-sync",
        "ok": True,
        "schedulableCount": len(summary["schedulableAccounts"]),
        "unschedulableCount": len(summary["unschedulableAccounts"]),
        "missingCount": len(summary["missingAccounts"]),
    }]

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"))
    if runtime_recovery_due(account_state):
        return None
    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")
            active_quarantine = isinstance(quarantine, dict) and quarantine.get("active") is True
            purpose = "recovery" if active_quarantine else "runtime-recovery" if runtime_recovery_due(account_state) 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")
        active_quarantine = isinstance(quarantine, dict) and quarantine.get("active") is True
        purpose = "manual-recovery" if active_quarantine else "manual-runtime-recovery" if runtime_recovery_due(account_state) 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)
    was_runtime_recovery = runtime_recovery_due(account_state)
    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 or was_runtime_recovery:
            if actions_enabled:
                try:
                    action_type = "restore" if was_recovery else "restore-runtime-unschedulable"
                    action = {"taken": True, "type": action_type, "result": admin.set_schedulable(name, True)}
                    account_state["runtimeSchedulable"] = True
                    account_state["runtimeSyncedAt"] = iso(now)
                    account_state.pop("runtimeSyncError", None)
                except Exception as exc:
                    action = {"taken": False, "type": "restore-failed", "error": str(exc)}
            if was_recovery:
                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
                    account_state["runtimeSchedulable"] = False
                    account_state["runtimeSyncedAt"] = iso(now)
                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"),
        "sentinelProtect": result.get("sentinelProtect"),
        "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, {})
    policy = protect_policy(profile)
    protected_probe = None
    if policy is not None:
        protected_probe = probe_account_with_protection(profile, config, "gateway-failure-confirm")
        protected_probe["sourceGatewayFailure"] = {
            "requestId": latest.get("requestId"),
            "clientRequestId": latest.get("clientRequestId"),
            "failureKind": latest.get("failureKind"),
            "path": latest.get("path"),
            "countInRun": len(failures),
        }
        account_state["lastProbeAt"] = iso(now)
        account_state["lastProbe"] = {
            "ok": protected_probe.get("ok"),
            "purpose": protected_probe.get("purpose"),
            "trustUpstream": protected_probe.get("trustUpstream"),
            "sentinelProtect": protected_probe.get("sentinelProtect"),
            "successMaxIntervalMinutes": success_max_interval(profile, config),
            "httpStatus": protected_probe.get("httpStatus"),
            "durationMs": protected_probe.get("durationMs"),
            "markerMatched": protected_probe.get("markerMatched"),
            "transportOk": protected_probe.get("transportOk"),
            "outputHash": protected_probe.get("outputHash"),
            "outputPreview": protected_probe.get("outputPreview"),
            "responseBodyHash": protected_probe.get("responseBodyHash"),
            "responseBodyPreview": protected_probe.get("responseBodyPreview"),
            "error": protected_probe.get("error"),
            "errorDetails": protected_probe.get("errorDetails"),
            "usage": protected_probe.get("usage"),
            "failureKind": protected_probe.get("failureKind"),
            "sdk": protected_probe.get("sdk"),
            "requestShape": protected_probe.get("requestShape"),
            "action": {"taken": False, "type": "protect-confirm-pass" if protected_probe.get("markerMatched") is True else "protect-confirm-fail"},
            "sourceGatewayFailure": protected_probe.get("sourceGatewayFailure"),
        }
        add_usage(state, account_state, now, protected_probe.get("usage") or {})
        account_state["sentinelProtect"] = protected_probe.get("sentinelProtect")
        account_state["trustUpstream"] = profile.get("trustUpstream") is True
        account_state["successMaxIntervalMinutes"] = success_max_interval(profile, config)
        if protected_probe.get("markerMatched") is True:
            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["nextProbeAfter"] = iso(add_minutes(now, interval, int(config["cadence"]["jitterPercent"])))
            account_state["lastOkAt"] = iso(now)
            account_state["lastStatus"] = "gateway-failure-protect-confirmed-ok"
            account_state["lastGatewayFailureAt"] = iso(now)
            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"),
                "countInRun": len(failures),
                "firstAt": failures[0].get("at"),
                "lastAt": latest.get("at"),
                "action": {"taken": False, "type": "protect-confirm-pass"},
                "sentinelProtect": protected_probe.get("sentinelProtect"),
            }
            return {"taken": False, "type": "protect-confirm-pass", "sentinelProtect": protected_probe.get("sentinelProtect")}
    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),
        "sentinelProtect": protected_probe.get("sentinelProtect") if isinstance(protected_probe, dict) else None,
    }
    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,
        "sentinelProtect": protected_probe.get("sentinelProtect") if isinstance(protected_probe, dict) else None,
    }
    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)
    runtime_sync = sync_runtime_schedulable_state(state, profiles, now, admin)
    reconcile = runtime_sync + 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_with_protection, 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"),
        },
        "runtimeSchedulable": state.get("runtimeSchedulable"),
        "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"),
            "sentinelProtect": item.get("sentinelProtect"),
            "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<string, unknown>, key: string): unknown {
  return Object.prototype.hasOwnProperty.call(value, key) ? value[key] : undefined;
}

function readRequiredRecord(value: unknown, key: string): Record<string, unknown> {
  if (!isRecord(value)) throw new Error(`${key} must be a YAML object`);
  return value;
}

function readRequiredBoolean(value: unknown, key: string): boolean {
  if (value === undefined || value === null) throw new Error(`${key} is required`);
  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 readRequiredDnsName(value: unknown, key: string): string {
  const text = readRequiredString(value, key);
  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 readRequiredModelName(value: unknown, key: string): string {
  const text = readRequiredString(value, key);
  if (!/^[A-Za-z0-9._:-]+$/u.test(text)) throw new Error(`${key} has an unsupported model name`);
  return text;
}

function readRequiredEndpoint(value: unknown, key: string): "responses" {
  const text = readRequiredString(value, key);
  if (text !== "responses") throw new Error(`${key} must be responses`);
  return text;
}

function readRequiredImage(value: unknown, key: string): string {
  const text = readRequiredString(value, key);
  if (!/^[A-Za-z0-9._:/@-]+$/u.test(text)) throw new Error(`${key} has an unsupported image format`);
  return text;
}

function readRequiredOpenAiPythonVersion(value: unknown, key: string): string {
  const text = readRequiredString(value, key);
  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 readRequiredString(value: unknown, key: string): string {
  if (value === undefined || value === null) throw new Error(`${key} is required`);
  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 readRequiredMarkerPrefix(value: unknown, key: string): string {
  const text = readRequiredString(value, key);
  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 readRequiredUserAgent(value: unknown, key: string): string {
  const text = readRequiredString(value, key);
  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 readRequiredPathList(value: unknown, key: string): string[] {
  if (value === undefined || value === null) throw new Error(`${key} is required`);
  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 readRequiredInt(value: unknown, key: string, min: number, max: number): number {
  if (value === undefined || value === null) throw new Error(`${key} is required`);
  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 readRequiredNumber(value: unknown, key: string, min: number, max: number): number {
  if (value === undefined || value === null) throw new Error(`${key} is required`);
  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<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}
