pikasTech-unidesk/scripts/src/platform-infra-sub2api-codex/remote-python-sync.ts

// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. remote-python-sync module for scripts/src/platform-infra-sub2api-codex.ts.

// Moved mechanically from scripts/src/platform-infra-sub2api-codex.ts:4265-5400 for #903.

import { chmodSync, copyFileSync, existsSync, mkdirSync, readFileSync, statSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import { join } from "node:path";
import type { UniDeskConfig } from "../config";
import { rootPath } from "../config";
import type { RenderedCliResult } from "../output";
import { applyPk01CaddyBlock, prepareFrpcSecret, renderFrpcManifest, type PublicServiceExposure, type PublicServiceTarget } from "../platform-infra-public-service";
import { shortSha256Fingerprint } from "../platform-infra-ops-library";
import {
  codexPoolSentinelSummary,
  codexPoolSentinelRuntimeImage,
  readCodexPoolSentinelConfig,
  renderCodexPoolSentinelManifest,
  type CodexPoolSentinelConfig,
  type CodexPoolSentinelProfileSecret,
} from "../platform-infra-sub2api-codex-sentinel";
import { parseEnvFile, readTextFile, redactRepoPath, requiredEnvValue } from "../secrets";
import { runSshCommandCapture, type SshCaptureResult } from "../ssh";

import type { CodexPoolConfig, CodexPoolRuntimeTarget } from "./types";
import { desiredAccountCapacityMap, desiredAccountLoadFactorMap, desiredAccountTempUnschedulableMap, desiredAccountWebSocketsV2ModeMap } from "./accounts";
import { resolvedManualAccountProtections } from "./public-exposure";
import { fieldManager } from "./types";

function pyJson(value: unknown): string {
  return `json.loads(${JSON.stringify(JSON.stringify(value))})`;
}

export function remotePythonScript(mode: "sync" | "validate" | "trace" | "cleanup-probes" | "sentinel-probe", encodedPayload: string, pool: CodexPoolConfig, target: CodexPoolRuntimeTarget): string {
  const hostDockerEnvPath = target.runtimeMode === "host-docker" ? target.hostDockerEnvPath : null;
  return `
set -u
python3 - <<'PY'
import base64
import hashlib
import json
import os
import re
import secrets
import string
import subprocess
import sys
import time
from datetime import datetime, timezone, timedelta
from urllib.parse import quote

TARGET_ID = ${pyJson(target.id)}
RUNTIME_MODE = ${pyJson(target.runtimeMode)}
NAMESPACE = ${pyJson(target.namespace)}
SERVICE_NAME = ${pyJson(target.serviceName)}
SERVICE_DNS = ${pyJson(target.serviceDns)}
HOST_DOCKER_APP_PORT = ${pyJson(target.hostDockerAppPort)}
HOST_DOCKER_ENV_PATH = ${pyJson(hostDockerEnvPath)}
HOST_DOCKER_APP_CONTAINER = "sub2api-app"
FIELD_MANAGER = "${fieldManager}"
APP_SECRET_NAME = ${pyJson(target.appSecretName)}
POOL_GROUP_NAME = "${pool.groupName}"
POOL_GROUP_DESCRIPTION = ${pyJson(pool.groupDescription)}
POOL_API_KEY_NAME = "${pool.apiKeyName}"
POOL_API_KEY_SECRET_NAME = "${pool.apiKeySecretName}"
POOL_API_KEY_SECRET_KEY = "${pool.apiKeySecretKey}"
POOL_ADMIN_EMAIL_DEFAULT = ${pyJson(pool.adminEmailDefault)}
MIN_OWNER_BALANCE_USD = ${pyJson(pool.minOwnerBalanceUsd)}
MIN_OWNER_CONCURRENCY = ${pyJson(pool.minOwnerConcurrency)}
MIN_OWNER_CONCURRENCY_SOURCE = ${pyJson(pool.minOwnerConcurrencySource)}
POOL_DEFAULT_ACCOUNT_PRIORITY = ${pyJson(pool.defaultAccountPriority)}
POOL_DEFAULT_ACCOUNT_CAPACITY = ${pyJson(pool.defaultAccountCapacity)}
POOL_DEFAULT_ACCOUNT_LOAD_FACTOR = ${pyJson(pool.defaultAccountLoadFactor)}
RESPONSES_SMOKE_MODEL = ${pyJson(pool.localCodex.responsesSmokeModel)}
EXPECTED_ACCOUNT_CAPACITIES = ${pyJson(desiredAccountCapacityMap(pool))}
EXPECTED_ACCOUNT_LOAD_FACTORS = ${pyJson(desiredAccountLoadFactorMap(pool))}
EXPECTED_ACCOUNT_WS_MODES = json.loads(${JSON.stringify(JSON.stringify(desiredAccountWebSocketsV2ModeMap(pool)))})
EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE = json.loads(${JSON.stringify(JSON.stringify(desiredAccountTempUnschedulableMap(pool)))})
MANUAL_ACCOUNT_PROTECTIONS = json.loads(${JSON.stringify(JSON.stringify(resolvedManualAccountProtections(pool, target)))})
SENTINEL_CONFIG = json.loads(${JSON.stringify(JSON.stringify(pool.sentinel))})
TARGET_EGRESS_PROXY = json.loads(${JSON.stringify(JSON.stringify(target.egressProxy))})
TARGET_SENTINEL_ENABLED = ${target.sentinelEnabled ? "True" : "False"}
MODE = "${mode}"
PAYLOAD_B64 = "${encodedPayload}"

def run(cmd, input_bytes=None):
    return subprocess.run(cmd, input=input_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

def text(data, limit=4000):
    if isinstance(data, bytes):
        data = data.decode("utf-8", errors="replace")
    return data[-limit:]

def read_host_env():
    if RUNTIME_MODE != "host-docker":
        return {}
    if not isinstance(HOST_DOCKER_ENV_PATH, str) or not HOST_DOCKER_ENV_PATH:
        raise RuntimeError("host-docker env source path missing")
    values = {}
    lines = read_host_env_lines()
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith("#") or "=" not in stripped:
            continue
        key, value = stripped.split("=", 1)
        key = key.strip()
        value = value.strip()
        if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
            value = value[1:-1]
        if key:
            values[key] = value
    return values

def read_host_env_lines():
    try:
        with open(HOST_DOCKER_ENV_PATH, "r", encoding="utf-8") as handle:
            return handle.read().splitlines()
    except FileNotFoundError:
        raise RuntimeError(f"host-docker env source missing: {HOST_DOCKER_ENV_PATH}")
    except PermissionError:
        proc = run(["sudo", "-n", "cat", HOST_DOCKER_ENV_PATH])
        if proc.returncode != 0:
            raise RuntimeError("read host-docker env source failed: " + text(proc.stderr, 1000))
        return proc.stdout.decode("utf-8", errors="replace").splitlines()

def write_host_env_value(key, value):
    if RUNTIME_MODE != "host-docker":
        raise RuntimeError("write_host_env_value is only valid for host-docker")
    if not isinstance(HOST_DOCKER_ENV_PATH, str) or not HOST_DOCKER_ENV_PATH:
        raise RuntimeError("host-docker env source path missing")
    if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key):
        raise RuntimeError(f"unsupported env key: {key}")
    os.makedirs(os.path.dirname(HOST_DOCKER_ENV_PATH), exist_ok=True)
    try:
        lines = read_host_env_lines()
    except RuntimeError as exc:
        if "missing" not in str(exc):
            raise
        lines = []
    next_lines = []
    replaced = False
    for line in lines:
        stripped = line.strip()
        if stripped.startswith("#") or "=" not in stripped:
            next_lines.append(line)
            continue
        current_key = stripped.split("=", 1)[0].strip()
        if current_key == key:
            next_lines.append(f"{key}={value}")
            replaced = True
        else:
            next_lines.append(line)
    if not replaced:
        next_lines.append(f"{key}={value}")
    content = "\\n".join(next_lines).rstrip() + "\\n"
    tmp_path = HOST_DOCKER_ENV_PATH + ".tmp"
    try:
        with open(tmp_path, "w", encoding="utf-8") as handle:
            handle.write(content)
        os.chmod(tmp_path, 0o600)
        os.replace(tmp_path, HOST_DOCKER_ENV_PATH)
    except PermissionError:
        try:
            os.unlink(tmp_path)
        except Exception:
            pass
        script = r'''
set -eu
path="$1"
dir="$(dirname "$path")"
mkdir -p "$dir"
tmp="$path.tmp.$$"
umask 077
cat > "$tmp"
mv "$tmp" "$path"
chmod 600 "$path"
'''
        proc = run(["sudo", "-n", "sh", "-c", script, "sh", HOST_DOCKER_ENV_PATH], content.encode("utf-8"))
        if proc.returncode != 0:
            raise RuntimeError("write host-docker env source failed: " + text(proc.stderr, 1000))
    return "updated" if replaced else "created"

def docker(args):
    proc = run(["docker", *args])
    if proc.returncode == 0:
        return proc
    sudo_proc = run(["sudo", "-n", "docker", *args])
    return sudo_proc if sudo_proc.returncode == 0 else proc

def runtime_logs(since, tail):
    if RUNTIME_MODE == "host-docker":
        return docker(["logs", f"--since={since}", f"--tail={tail}", HOST_DOCKER_APP_CONTAINER])
    return kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", f"--since={since}", f"--tail={tail}"])

def kubectl(args, input_obj=None):
    if isinstance(input_obj, str):
        input_bytes = input_obj.encode("utf-8")
    else:
        input_bytes = input_obj
    return run(["kubectl", *args], input_bytes)

def require_kubectl(args, input_obj=None, label="kubectl"):
    proc = kubectl(args, input_obj)
    if proc.returncode != 0:
        raise RuntimeError(f"{label} failed: {text(proc.stderr, 1000)}")
    return proc.stdout

def kube_json(args, label):
    raw = require_kubectl([*args, "-o", "json"], label=label)
    return json.loads(raw.decode("utf-8"))

def decode_secret_value(name, key):
    if RUNTIME_MODE == "host-docker":
        return read_host_env().get(key)
    data = kube_json(["-n", NAMESPACE, "get", "secret", name], f"secret/{name}").get("data") or {}
    if key not in data:
        return None
    return base64.b64decode(data[key]).decode("utf-8")

def get_config_value(name, key):
    if RUNTIME_MODE == "host-docker":
        return read_host_env().get(key)
    data = kube_json(["-n", NAMESPACE, "get", "configmap", name], f"configmap/{name}").get("data") or {}
    value = data.get(key)
    return value if isinstance(value, str) and value else None

def select_app_pod():
    if RUNTIME_MODE == "host-docker":
        return HOST_DOCKER_APP_CONTAINER
    pods = kube_json(["-n", NAMESPACE, "get", "pods", "-l", "app.kubernetes.io/name=sub2api"], "sub2api pods").get("items") or []
    for pod in pods:
        status = pod.get("status") or {}
        if status.get("phase") != "Running":
            continue
        statuses = status.get("containerStatuses") or []
        if statuses and all(item.get("ready") is True for item in statuses):
            return pod["metadata"]["name"]
    if pods:
        return pods[0]["metadata"]["name"]
    raise RuntimeError("sub2api app pod not found")

APP_POD = select_app_pod()

def parse_curl_output(proc):
    stdout = proc.stdout.decode("utf-8", errors="replace")
    marker = "\\n__HTTP_CODE__:"
    pos = stdout.rfind(marker)
    if pos < 0:
        return {
            "ok": False,
            "httpStatus": 0,
            "json": None,
            "body": stdout,
            "stderr": text(proc.stderr, 1000),
            "transportExitCode": proc.returncode,
        }
    body = stdout[:pos]
    status_text = stdout[pos + len(marker):].strip()
    try:
        http_status = int(status_text[-3:])
    except ValueError:
        http_status = 0
    try:
        parsed = json.loads(body) if body.strip() else None
    except json.JSONDecodeError:
        parsed = None
    return {
        "ok": proc.returncode == 0 and 200 <= http_status < 300,
        "httpStatus": http_status,
        "json": parsed,
        "body": body,
        "stderr": text(proc.stderr, 1000),
        "transportExitCode": proc.returncode,
    }

def utc_iso(offset_seconds=0):
    return (datetime.now(timezone.utc) + timedelta(seconds=offset_seconds)).strftime("%Y-%m-%dT%H:%M:%SZ")

def normalize_runtime_base_url(value):
    if not isinstance(value, str):
        return None
    value = value.strip().rstrip("/")
    return value or None

def empty_to_none(value):
    return value if isinstance(value, str) and value else None

def sentinel_quality_gate_enabled():
    return TARGET_SENTINEL_ENABLED and (SENTINEL_CONFIG.get("monitor") or {}).get("enabled") is True and (SENTINEL_CONFIG.get("actions") or {}).get("enabled") is True

def account_notes_fingerprint(account):
    notes = account.get("notes") if isinstance(account, dict) else None
    if not isinstance(notes, str):
        return None
    match = re.search(r"fingerprint=([A-Za-z0-9_-]+)", notes)
    return match.group(1) if match else None

def runtime_account_credentials(account):
    credentials = account.get("credentials") if isinstance(account, dict) and isinstance(account.get("credentials"), dict) else {}
    return credentials

def runtime_account_extra(account):
    extra = account.get("extra") if isinstance(account, dict) and isinstance(account.get("extra"), dict) else {}
    return extra

def sentinel_probe_change_reasons(current, profile):
    if not isinstance(current, dict) or current.get("id") is None:
        return ["created"]
    credentials = runtime_account_credentials(current)
    extra = runtime_account_extra(current)
    expected_base_url = normalize_runtime_base_url(profile.get("baseUrl"))
    runtime_base_url = normalize_runtime_base_url(credentials.get("base_url"))
    expected_user_agent = empty_to_none(profile.get("upstreamUserAgent"))
    runtime_user_agent = empty_to_none(credentials.get("user_agent"))
    expected_ws_mode = empty_to_none(profile.get("openaiResponsesWebSocketsV2Mode"))
    runtime_ws_mode = empty_to_none(extra.get("openai_apikey_responses_websockets_v2_mode"))
    expected_trust_upstream = profile.get("trustUpstream") is True
    runtime_trust_upstream = extra.get("unidesk_trust_upstream") is True
    expected_protect = profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False}
    runtime_protect = extra.get("unidesk_sentinel_protect") if isinstance(extra.get("unidesk_sentinel_protect"), dict) else {"enabled": False}
    reasons = []
    if empty_to_none(extra.get("unidesk_codex_profile")) != profile.get("profile"):
        reasons.append("profile")
    if runtime_base_url != expected_base_url:
        reasons.append("base-url")
    if account_notes_fingerprint(current) != profile.get("apiKeyFingerprint"):
        reasons.append("api-key-fingerprint")
    if runtime_user_agent != expected_user_agent:
        reasons.append("upstream-user-agent")
    if runtime_ws_mode != expected_ws_mode:
        reasons.append("responses-websockets-v2-mode")
    if runtime_trust_upstream != expected_trust_upstream:
        reasons.append("trust-upstream")
    if runtime_protect != expected_protect:
        reasons.append("sentinel-protect")
    return reasons

def curl_api(method, path, bearer=None, payload=None):
    body = b"" if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8")
    script = r'''
set -eu
method="$1"
url="$2"
token="\${3:-}"
tmp="$(mktemp)"
trap 'rm -f "$tmp"' EXIT
cat > "$tmp"
if [ -n "$token" ]; then
  if [ "$method" = "GET" ] && [ ! -s "$tmp" ]; then
    curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" -H "Authorization: Bearer $token" "$url"
  else
    curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' -H "Authorization: Bearer $token" --data-binary @"$tmp" "$url"
  fi
else
  if [ "$method" = "GET" ] && [ ! -s "$tmp" ]; then
    curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" "$url"
  else
    curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' --data-binary @"$tmp" "$url"
  fi
fi
'''
    if RUNTIME_MODE == "host-docker":
        if not isinstance(HOST_DOCKER_APP_PORT, int):
            raise RuntimeError("host-docker app port missing")
        proc = run(["sh", "-c", script, "sh", method, f"http://127.0.0.1:{HOST_DOCKER_APP_PORT}{path}", bearer or ""], body)
    else:
        proc = run([
            "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
            "--", "sh", "-c", script, "sh", method, f"http://127.0.0.1:8080{path}", bearer or "",
        ], body)
    return parse_curl_output(proc)

def envelope_data(parsed):
    if isinstance(parsed, dict) and "data" in parsed:
        return parsed.get("data")
    return parsed

def ensure_success(resp, label):
    parsed = resp.get("json")
    code = parsed.get("code") if isinstance(parsed, dict) else None
    if not resp.get("ok") or (code is not None and code != 0):
        message = parsed.get("message") if isinstance(parsed, dict) else text(resp.get("body", ""), 500)
        raise RuntimeError(f"{label} failed: http={resp.get('httpStatus')} message={message}")
    return envelope_data(parsed)

def ensure_admin_compliance(token):
    status_resp = curl_api("GET", "/api/v1/admin/compliance", bearer=token)
    if status_resp.get("httpStatus") == 404:
        return {
            "ok": True,
            "action": "not-supported-by-runtime",
            "valuesPrinted": False,
        }
    status = ensure_success(status_resp, "get admin compliance")
    if not isinstance(status, dict):
        return {
            "ok": True,
            "action": "status-unstructured",
            "valuesPrinted": False,
        }
    version = status.get("version")
    required = status.get("required") is True
    if not required:
        return {
            "ok": True,
            "action": "already-acknowledged",
            "version": version,
            "requiredBefore": False,
            "requiredAfter": False,
            "phrasePrinted": False,
            "valuesPrinted": False,
        }
    phrase = status.get("ack_phrase_zh") if isinstance(status.get("ack_phrase_zh"), str) and status.get("ack_phrase_zh") else status.get("ack_phrase_en")
    language = "zh" if isinstance(status.get("ack_phrase_zh"), str) and status.get("ack_phrase_zh") else "en"
    if not isinstance(phrase, str) or not phrase:
        raise RuntimeError("admin compliance acknowledgement phrase missing")
    accepted = ensure_success(
        curl_api("POST", "/api/v1/admin/compliance/accept", bearer=token, payload={"phrase": phrase, "language": language}),
        "accept admin compliance",
    )
    required_after = accepted.get("required") if isinstance(accepted, dict) else None
    return {
        "ok": required_after is False,
        "action": "accepted",
        "version": version,
        "requiredBefore": True,
        "requiredAfter": required_after,
        "phrasePrinted": False,
        "valuesPrinted": False,
    }

def extract_items(data):
    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        if isinstance(data.get("items"), list):
            return data["items"]
        for key in ("groups", "accounts", "api_keys", "keys"):
            if isinstance(data.get(key), list):
                return data[key]
    return []

def find_access_token(data):
    if isinstance(data, dict):
        for key in ("access_token", "token"):
            if isinstance(data.get(key), str) and data[key]:
                return data[key]
        for value in data.values():
            token = find_access_token(value)
            if token:
                return token
    return None

def login():
    admin_email = get_config_value("sub2api-config", "ADMIN_EMAIL") or POOL_ADMIN_EMAIL_DEFAULT
    admin_password = decode_secret_value(APP_SECRET_NAME, "ADMIN_PASSWORD")
    if not admin_password:
        raise RuntimeError("ADMIN_PASSWORD missing from sub2api-secrets")
    data = ensure_success(curl_api("POST", "/api/v1/auth/login", payload={"email": admin_email, "password": admin_password}), "admin login")
    token = find_access_token(data)
    if not token:
        raise RuntimeError("admin login response did not contain access_token")
    compliance = ensure_admin_compliance(token)
    if compliance.get("ok") is not True:
        raise RuntimeError("admin compliance acknowledgement failed")
    return admin_email, token, compliance

def group_payload():
    return {
        "name": POOL_GROUP_NAME,
        "description": POOL_GROUP_DESCRIPTION,
        "platform": "openai",
        "rate_multiplier": 1,
        "is_exclusive": False,
        "subscription_type": "standard",
        "allow_messages_dispatch": True,
        "require_oauth_only": False,
        "require_privacy_set": False,
        "rpm_limit": 0,
    }

def list_groups(token):
    data = ensure_success(curl_api("GET", "/api/v1/admin/groups/all?platform=openai", bearer=token), "list groups")
    return extract_items(data)

def ensure_group(token):
    existing = next((item for item in list_groups(token) if item.get("name") == POOL_GROUP_NAME), None)
    payload = group_payload()
    if existing is None:
        created = ensure_success(curl_api("POST", "/api/v1/admin/groups", bearer=token, payload=payload), "create group")
        return created, "created"
    group_id = existing.get("id")
    if group_id is None:
        raise RuntimeError("existing group has no id")
    payload["status"] = "active"
    updated = ensure_success(curl_api("PUT", f"/api/v1/admin/groups/{group_id}", bearer=token, payload=payload), "update group")
    return updated if isinstance(updated, dict) else existing, "updated"

def list_accounts_for_group(token, group_id):
    path = f"/api/v1/admin/accounts?group_id={group_id}&page=1&page_size=500&platform=openai"
    data = ensure_success(curl_api("GET", path, bearer=token), f"list accounts for group {group_id}")
    return extract_items(data)

def account_group_ids(token, account):
    if not isinstance(account, dict) or account.get("id") is None:
        return []
    account_id = account.get("id")
    account_name = account.get("name")
    ids = []
    for group in list_groups(token):
        group_id = group.get("id") if isinstance(group, dict) else None
        if group_id is None:
            continue
        members = list_accounts_for_group(token, group_id)
        if any(item.get("id") == account_id or item.get("name") == account_name for item in members if isinstance(item, dict)):
            ids.append(group_id)
    return sorted(set(ids))

def list_accounts(token):
    path = "/api/v1/admin/accounts?page=1&page_size=200&platform=openai&type=apikey&search=" + quote("unidesk-codex-")
    data = ensure_success(curl_api("GET", path, bearer=token), "list accounts")
    return extract_items(data)

def find_account_by_name(token, name):
    path = "/api/v1/admin/accounts?page=1&page_size=50&platform=openai&search=" + quote(str(name))
    data = ensure_success(curl_api("GET", path, bearer=token), "find account " + str(name))
    for item in extract_items(data):
        if isinstance(item, dict) and item.get("name") == name:
            return item
    return None

def list_proxy_candidates(token, search=""):
    path = "/api/v1/admin/proxies?page=1&page_size=200"
    if search:
        path += "&search=" + quote(str(search))
    data = ensure_success(curl_api("GET", path, bearer=token), "list proxies")
    return extract_items(data)

def find_proxy_by_name(token, name):
    for item in list_proxy_candidates(token, name):
        if isinstance(item, dict) and item.get("name") == name:
            return item
    return None

def manual_binding_plan(binding, expected_kind):
    if not isinstance(binding, dict):
        return None
    plan = binding.get("sourcePlan")
    if not isinstance(plan, dict):
        raise RuntimeError("manual account binding is missing resolved sourcePlan")
    if plan.get("enabled") is not True:
        raise RuntimeError(f"manual account binding source {plan.get('id')} is disabled")
    if plan.get("kind") != expected_kind:
        raise RuntimeError(f"manual account binding source {plan.get('id')} kind must be {expected_kind}")
    return plan

def desired_manual_proxy_payload(protection):
    binding = protection.get("proxyBinding") if isinstance(protection, dict) else None
    if not isinstance(binding, dict) or binding.get("enabled") is not True:
        return None
    plan = manual_binding_plan(binding, "proxy")
    if plan.get("provider") not in ("target-egress-proxy", "fixed-http-proxy"):
        raise RuntimeError(f"manual account proxyBinding source {plan.get('id')} uses unsupported provider {plan.get('provider')}")
    if plan.get("provider") == "fixed-http-proxy":
        fixed_proxy = plan.get("fixedProxy")
        if not isinstance(fixed_proxy, dict):
            raise RuntimeError(f"manual account proxyBinding source {plan.get('id')} requires fixedProxy")
        proxy_name = binding.get("proxyName")
        protocol = fixed_proxy.get("protocol")
        host = fixed_proxy.get("host")
        port = fixed_proxy.get("port")
        if not isinstance(proxy_name, str) or not proxy_name:
            raise RuntimeError("manual account proxyBinding proxyName is missing")
        if protocol != "http":
            raise RuntimeError("fixed proxy protocol must be http")
        if not isinstance(host, str) or not host:
            raise RuntimeError("fixed proxy host is missing")
        if not isinstance(port, int) or port <= 0:
            raise RuntimeError("fixed proxy port is missing")
        return {
            "name": proxy_name,
            "protocol": protocol,
            "host": host,
            "port": port,
            "fallback_mode": "none",
            "expiry_warn_days": 0,
        }
    target_proxy = plan.get("targetEgressProxy")
    if not isinstance(target_proxy, dict):
        raise RuntimeError(f"manual account proxyBinding source {plan.get('id')} requires an enabled target egressProxy")
    service_name = target_proxy.get("serviceName")
    listen_port = target_proxy.get("listenPort")
    namespace = target_proxy.get("namespace") if isinstance(target_proxy.get("namespace"), str) and target_proxy.get("namespace") else NAMESPACE
    proxy_name = binding.get("proxyName")
    if not isinstance(service_name, str) or not service_name:
        raise RuntimeError("target egressProxy serviceName is missing")
    if not isinstance(listen_port, int) or listen_port <= 0:
        raise RuntimeError("target egressProxy listenPort is missing")
    if not isinstance(proxy_name, str) or not proxy_name:
        raise RuntimeError("manual account proxyBinding proxyName is missing")
    return {
        "name": proxy_name,
        "protocol": "http",
        "host": f"{service_name}.{namespace}.svc.cluster.local",
        "port": listen_port,
        "fallback_mode": "none",
        "expiry_warn_days": 0,
    }

def proxy_needs_update(proxy, payload):
    if not isinstance(proxy, dict):
        return True
    for key in ("name", "protocol", "host", "port", "fallback_mode", "expiry_warn_days"):
        if proxy.get(key) != payload.get(key):
            return True
    return proxy.get("status") != "active"

def ensure_manual_proxy(token, payload):
    current = find_proxy_by_name(token, payload["name"])
    if current is None:
        created = ensure_success(curl_api("POST", "/api/v1/admin/proxies", bearer=token, payload=payload), f"create proxy {payload['name']}")
        return created, "created"
    if proxy_needs_update(current, payload):
        update_payload = dict(payload)
        update_payload["status"] = "active"
        updated = ensure_success(curl_api("PUT", f"/api/v1/admin/proxies/{current['id']}", bearer=token, payload=update_payload), f"update proxy {payload['name']}")
        return updated if isinstance(updated, dict) else current, "updated"
    return current, "unchanged"

def manual_proxy_status(token, account, protection):
    payload = desired_manual_proxy_payload(protection)
    if payload is None:
        return {
            "enabled": False,
            "ok": True,
            "action": "not-configured",
            "valuesPrinted": False,
        }
    binding = protection.get("proxyBinding") if isinstance(protection, dict) else None
    plan = manual_binding_plan(binding, "proxy")
    proxy = find_proxy_by_name(token, payload["name"])
    runtime_proxy = account.get("proxy") if isinstance(account, dict) and isinstance(account.get("proxy"), dict) else None
    binding_aligned = (
        isinstance(account, dict)
        and isinstance(proxy, dict)
        and account.get("proxy_id") == proxy.get("id")
    )
    return {
        "enabled": True,
        "ok": binding_aligned,
        "action": "validate",
        "source": plan.get("id"),
        "sourceProvider": plan.get("provider"),
        "expectedProxyName": payload["name"],
        "expectedProtocol": payload["protocol"],
        "expectedHost": payload["host"],
        "expectedPort": payload["port"],
        "proxyRecordExists": isinstance(proxy, dict),
        "proxyId": proxy.get("id") if isinstance(proxy, dict) else None,
        "runtimeProxyId": account.get("proxy_id") if isinstance(account, dict) else None,
        "runtimeProxyName": runtime_proxy.get("name") if isinstance(runtime_proxy, dict) else None,
        "bindingAligned": binding_aligned,
        "valuesPrinted": False,
    }

def ensure_manual_account_proxy_bindings(token):
    items = []
    for protection in MANUAL_ACCOUNT_PROTECTIONS:
        if not isinstance(protection, dict):
            continue
        name = protection.get("accountName")
        if not isinstance(name, str) or not name:
            continue
        account = find_account_by_name(token, name)
        payload = desired_manual_proxy_payload(protection)
        if payload is None:
            items.append({
                "accountName": name,
                "enabled": False,
                "action": "not-configured",
                "ok": True,
                "valuesPrinted": False,
            })
            continue
        if not isinstance(account, dict):
            items.append({
                "accountName": name,
                "enabled": True,
                "action": "account-missing",
                "ok": False,
                "expectedProxyName": payload["name"],
                "valuesPrinted": False,
            })
            continue
        proxy, proxy_action = ensure_manual_proxy(token, payload)
        binding = protection.get("proxyBinding") if isinstance(protection, dict) else None
        plan = manual_binding_plan(binding, "proxy")
        proxy_id = proxy.get("id") if isinstance(proxy, dict) else None
        if proxy_id is None:
            raise RuntimeError(f"proxy {payload['name']} has no id")
        action = "unchanged"
        if account.get("proxy_id") != proxy_id:
            updated = ensure_success(curl_api("PUT", f"/api/v1/admin/accounts/{account['id']}", bearer=token, payload={"proxy_id": proxy_id}), f"bind manual account proxy {name}")
            account = updated if isinstance(updated, dict) else account
            action = "bound"
        runtime_proxy = account.get("proxy") if isinstance(account.get("proxy"), dict) else None
        items.append({
            "accountName": name,
            "accountId": account.get("id"),
            "enabled": True,
            "action": action,
            "proxyAction": proxy_action,
            "source": plan.get("id"),
            "sourceProvider": plan.get("provider"),
            "ok": account.get("proxy_id") == proxy_id,
            "expectedProxyName": payload["name"],
            "expectedProtocol": payload["protocol"],
            "expectedHost": payload["host"],
            "expectedPort": payload["port"],
            "proxyId": proxy_id,
            "runtimeProxyId": account.get("proxy_id"),
            "runtimeProxyName": runtime_proxy.get("name") if isinstance(runtime_proxy, dict) else None,
            "bindingAligned": account.get("proxy_id") == proxy_id,
            "controlPolicy": "manual-protected: only proxy_id binding is YAML-controlled; credentials/status/schedulable are untouched",
            "valuesPrinted": False,
        })
    return {
        "ok": all(item.get("ok") is True for item in items),
        "itemCount": len(items),
        "items": items,
        "valuesPrinted": False,
    }

def manual_group_binding_plan(protection):
    binding = protection.get("groupBinding") if isinstance(protection, dict) else None
    if not isinstance(binding, dict) or binding.get("enabled") is not True:
        return None
    plan = manual_binding_plan(binding, "group")
    if plan.get("provider") != "pool-group":
        raise RuntimeError(f"manual account groupBinding source {plan.get('id')} uses unsupported provider {plan.get('provider')}")
    return plan

def manual_group_binding_enabled(protection):
    return manual_group_binding_plan(protection) is not None

def manual_group_status(token, account, protection, group_id):
    plan = manual_group_binding_plan(protection)
    if plan is None:
        return {
            "enabled": False,
            "ok": True,
            "action": "not-configured",
            "valuesPrinted": False,
        }
    group_accounts = list_accounts_for_group(token, group_id)
    account_id = account.get("id") if isinstance(account, dict) else None
    account_name = account.get("name") if isinstance(account, dict) else None
    binding_aligned = any(
        item.get("id") == account_id or item.get("name") == account_name
        for item in group_accounts
        if isinstance(item, dict)
    )
    return {
        "enabled": True,
        "ok": binding_aligned,
        "action": "validate",
        "source": plan.get("id"),
        "sourceProvider": plan.get("provider"),
        "poolGroupName": POOL_GROUP_NAME,
        "poolGroupId": group_id,
        "bindingAligned": binding_aligned,
        "valuesPrinted": False,
    }

def ensure_manual_account_group_bindings(token, group_id):
    items = []
    for protection in MANUAL_ACCOUNT_PROTECTIONS:
        if not isinstance(protection, dict):
            continue
        name = protection.get("accountName")
        if not isinstance(name, str) or not name:
            continue
        if not manual_group_binding_enabled(protection):
            items.append({
                "accountName": name,
                "enabled": False,
                "action": "not-configured",
                "ok": True,
                "valuesPrinted": False,
            })
            continue
        plan = manual_group_binding_plan(protection)
        account = find_account_by_name(token, name)
        if not isinstance(account, dict):
            items.append({
                "accountName": name,
                "enabled": True,
                "action": "account-missing",
                "ok": False,
                "poolGroupName": POOL_GROUP_NAME,
                "poolGroupId": group_id,
                "valuesPrinted": False,
            })
            continue
        existing_group_ids = account_group_ids(token, account)
        desired_group_ids = sorted(set(existing_group_ids + [group_id]))
        action = "unchanged"
        if group_id not in existing_group_ids:
            updated = ensure_success(curl_api("PUT", f"/api/v1/admin/accounts/{account['id']}", bearer=token, payload={"group_ids": desired_group_ids}), f"bind manual account group {name}")
            account = updated if isinstance(updated, dict) else account
            action = "bound"
        binding_aligned = any(
            item.get("id") == account.get("id") or item.get("name") == name
            for item in list_accounts_for_group(token, group_id)
            if isinstance(item, dict)
        )
        items.append({
            "accountName": name,
            "accountId": account.get("id"),
            "enabled": True,
            "ok": binding_aligned,
            "action": action,
            "source": plan.get("id") if isinstance(plan, dict) else None,
            "sourceProvider": plan.get("provider") if isinstance(plan, dict) else None,
            "poolGroupName": POOL_GROUP_NAME,
            "poolGroupId": group_id,
            "previousGroupIds": existing_group_ids,
            "desiredGroupIds": desired_group_ids,
            "bindingAligned": binding_aligned,
            "controlPolicy": "manual-protected: only pool group membership is YAML-controlled; credentials/status/schedulable are untouched and sentinel does not probe it",
            "valuesPrinted": False,
        })
    return {
        "ok": all(item.get("ok") is True for item in items),
        "itemCount": len(items),
        "items": items,
        "valuesPrinted": False,
    }

def manual_account_protection_status(token, group_id=None):
    items = []
    desired_names = set(EXPECTED_ACCOUNT_CAPACITIES.keys())
    for protection in MANUAL_ACCOUNT_PROTECTIONS:
        if not isinstance(protection, dict):
            continue
        name = protection.get("accountName")
        if not isinstance(name, str) or not name:
            continue
        account = find_account_by_name(token, name)
        extra = account.get("extra") if isinstance(account, dict) and isinstance(account.get("extra"), dict) else {}
        proxy_status = manual_proxy_status(token, account, protection)
        group_status = manual_group_status(token, account, protection, group_id) if group_id is not None else {"enabled": False, "ok": True, "action": "not-checked", "valuesPrinted": False}
        items.append({
            "accountName": name,
            "reason": protection.get("reason") if isinstance(protection.get("reason"), str) else None,
            "exists": isinstance(account, dict),
            "accountId": account.get("id") if isinstance(account, dict) else None,
            "status": account.get("status") if isinstance(account, dict) else None,
            "schedulable": account.get("schedulable") if isinstance(account, dict) else None,
            "inYamlProfiles": name in desired_names,
            "runtimeMarkedUnideskManaged": extra.get("unidesk_managed") is True,
            "proxyBinding": proxy_status,
            "groupBinding": group_status,
            "ok": proxy_status.get("ok") is True and group_status.get("ok") is True,
            "controlPolicy": "manual-protected: no create/update/prune/probe/freeze; optional proxy_id and pool group membership binding only when configured",
            "valuesPrinted": False,
        })
    return {
        "ok": all(item.get("ok") is True for item in items),
        "protectedCount": len(items),
        "items": items,
        "valuesPrinted": False,
    }

def list_probe_accounts(token):
    path = "/api/v1/admin/accounts?page=1&page_size=200&platform=openai&type=apikey&search=" + quote("unidesk-probe-")
    data = ensure_success(curl_api("GET", path, bearer=token), "list probe accounts")
    return [item for item in extract_items(data) if isinstance(item.get("name"), str) and item.get("name").startswith("unidesk-probe-")]

def list_probe_groups(token):
    data = ensure_success(curl_api("GET", "/api/v1/admin/groups/all?platform=openai", bearer=token), "list groups")
    return [item for item in extract_items(data) if isinstance(item.get("name"), str) and item.get("name").startswith("unidesk-probe-")]

def cleanup_probe_resources(token):
    api_key_results = []
    account_results = []
    group_results = []
    for item in list_user_keys(token):
        name = item.get("name")
        key_id = item.get("id")
        if not isinstance(name, str) or not name.startswith("unidesk-probe-") or key_id is None:
            continue
        api_key_results.append(delete_probe_resource(token, "DELETE", f"/api/v1/keys/{key_id}", "api-key"))
    for item in list_probe_accounts(token):
        account_id = item.get("id")
        if account_id is None:
            continue
        account_results.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/accounts/{account_id}", "account"))
    for item in list_probe_groups(token):
        group_id = item.get("id")
        if group_id is None:
            continue
        group_results.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/groups/{group_id}", "group"))
    return {
        "ok": all(item.get("ok") is True for item in api_key_results + account_results + group_results),
        "apiKeysDeleted": len(api_key_results),
        "accountsDeleted": len(account_results),
        "groupsDeleted": len(group_results),
        "items": {
            "apiKeys": api_key_results,
            "accounts": account_results,
            "groups": group_results,
        },
        "valuesPrinted": False,
    }

def temp_unschedulable_credentials(profile):
    credentials = profile.get("tempUnschedulableCredentials")
    if not isinstance(credentials, dict):
        credentials = {}
    return normalize_temp_unschedulable_credentials(credentials)

def account_payload(profile, group_id):
    extra = {
        "openai_responses_mode": "force_responses",
        "unidesk_codex_profile": profile["profile"],
        "unidesk_managed": True,
        "unidesk_trust_upstream": profile.get("trustUpstream") is True,
        "unidesk_sentinel_protect": profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False},
    }
    ws_mode = profile.get("openaiResponsesWebSocketsV2Mode")
    if ws_mode:
        extra["openai_apikey_responses_websockets_v2_mode"] = ws_mode
        extra["openai_apikey_responses_websockets_v2_enabled"] = ws_mode != "off"
    credentials = {
        "api_key": profile["apiKey"],
        "base_url": profile["baseUrl"],
    }
    upstream_user_agent = profile.get("upstreamUserAgent")
    if upstream_user_agent:
        credentials["user_agent"] = upstream_user_agent
    temp_unschedulable = temp_unschedulable_credentials(profile)
    if temp_unschedulable["enabled"]:
        credentials["temp_unschedulable_enabled"] = True
        credentials["temp_unschedulable_rules"] = temp_unschedulable["rules"]
    return {
        "name": profile["accountName"],
        "notes": f"UniDesk-managed Codex profile {profile['profile']} from {profile['configFile']} and {profile['authFile']}; secret source={profile['apiKeySource']}; fingerprint={profile['apiKeyFingerprint']}.",
        "platform": "openai",
        "type": "apikey",
        "credentials": credentials,
        "extra": extra,
        "concurrency": int(profile.get("capacity", 5) or 5),
        "priority": int(profile.get("priority", POOL_DEFAULT_ACCOUNT_PRIORITY) or POOL_DEFAULT_ACCOUNT_PRIORITY),
        "rate_multiplier": 1,
        "load_factor": int(profile.get("loadFactor", POOL_DEFAULT_ACCOUNT_LOAD_FACTOR) or POOL_DEFAULT_ACCOUNT_LOAD_FACTOR),
        "group_ids": [group_id],
        "confirm_mixed_channel_risk": True,
        **({"proxy_id": int(profile["proxyId"])} if profile.get("proxyId") is not None else {}),
    }

def planned_sentinel_account_results(profiles, existing_accounts):
    existing = {item.get("name"): item for item in existing_accounts if isinstance(item, dict)}
    results = []
    for profile in profiles:
        change_reasons = sentinel_probe_change_reasons(existing.get(profile["accountName"]), profile)
        quality_gate_required = sentinel_quality_gate_enabled() and len(change_reasons) > 0
        results.append({
            "profile": profile["profile"],
            "accountName": profile["accountName"],
            "profileConfig": {
                "accountName": profile["accountName"],
                "trustUpstream": profile.get("trustUpstream") is True,
                "sentinelProtect": profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False},
            },
            "sentinelProbeConfigFingerprint": profile.get("sentinelProbeConfigFingerprint"),
            "sentinelProbeRequired": quality_gate_required,
            "sentinelChangeReasons": change_reasons if quality_gate_required else [],
            "sentinelProbePending": quality_gate_required,
            "sentinelDefaultFrozen": False,
            "valuesPrinted": False,
        })
    return results

def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_frozen_names=None, existing_accounts=None):
    if not isinstance(protected_frozen_names, set):
        protected_frozen_names = set()
    if not isinstance(existing_accounts, list):
        existing_accounts = list_accounts(token)
    existing = {item.get("name"): item for item in existing_accounts}
    desired_names = {profile["accountName"] for profile in profiles}
    results = []
    for profile in profiles:
        payload = account_payload(profile, group_id)
        current = existing.get(profile["accountName"])
        change_reasons = sentinel_probe_change_reasons(current, profile)
        quality_gate_required = sentinel_quality_gate_enabled() and len(change_reasons) > 0
        keep_frozen = profile["accountName"] in protected_frozen_names
        if current and current.get("id") is not None:
            account_id = current["id"]
            update_payload = dict(payload)
            update_payload.pop("platform", None)
            update_payload["status"] = "active"
            data = ensure_success(curl_api("PUT", f"/api/v1/admin/accounts/{account_id}", bearer=token, payload=update_payload), f"update account {profile['accountName']}")
            action = "updated"
        else:
            data = ensure_success(curl_api("POST", "/api/v1/admin/accounts", bearer=token, payload=payload), f"create account {profile['accountName']}")
            action = "created"
        results.append({
            "profile": profile["profile"],
            "accountName": profile["accountName"],
            "profileConfig": {
                "accountName": profile["accountName"],
                "trustUpstream": profile.get("trustUpstream") is True,
                "sentinelProtect": profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False},
            },
            "accountId": data.get("id") if isinstance(data, dict) else None,
            "action": action,
            "baseUrl": profile["baseUrl"],
            "apiKeySource": profile["apiKeySource"],
            "apiKeyFingerprint": profile["apiKeyFingerprint"],
            "sentinelProbeConfigFingerprint": profile.get("sentinelProbeConfigFingerprint"),
            "sentinelProbeRequired": quality_gate_required,
            "sentinelChangeReasons": change_reasons if quality_gate_required else [],
            "sentinelProbePending": quality_gate_required,
            "sentinelDefaultFrozen": False,
            "sentinelFreezeProtected": keep_frozen,
            "schedulableControl": "sentinel-marker-restore",
            "openaiResponsesWebSocketsV2Mode": profile.get("openaiResponsesWebSocketsV2Mode"),
            "trustUpstream": profile.get("trustUpstream") is True,
            "priority": int(profile.get("priority", POOL_DEFAULT_ACCOUNT_PRIORITY) or POOL_DEFAULT_ACCOUNT_PRIORITY),
            "capacity": int(profile.get("capacity", 5) or 5),
            "loadFactor": int(profile.get("loadFactor", POOL_DEFAULT_ACCOUNT_LOAD_FACTOR) or POOL_DEFAULT_ACCOUNT_LOAD_FACTOR),
            "runtimeConcurrency": data.get("concurrency") if isinstance(data, dict) else None,
            "runtimeLoadFactor": data.get("load_factor") if isinstance(data, dict) else None,
            "runtimeSchedulable": data.get("schedulable") if isinstance(data, dict) else None,
            "tempUnschedulableConfigured": bool(payload["credentials"].get("temp_unschedulable_enabled")),
            "tempUnschedulableRuleCount": len(payload["credentials"].get("temp_unschedulable_rules") or []),
            "upstreamUserAgentConfigured": bool(profile.get("upstreamUserAgent")),
            "valuesPrinted": False,
        })
    prune_results = prune_removed_accounts(token, existing_accounts, desired_names) if prune_removed else []
    return results, prune_results

def prune_removed_accounts(token, existing_accounts, desired_names):
    results = []
    for account in existing_accounts:
        name = account.get("name")
        account_id = account.get("id")
        extra = account.get("extra") if isinstance(account.get("extra"), dict) else {}
        if not isinstance(name, str) or not name.startswith("unidesk-codex-"):
            continue
        if extra.get("unidesk_managed") is not True:
            continue
        if name in desired_names:
            continue
        if account_id is None:
            raise RuntimeError(f"removed account {name} has no id")
        ensure_success(curl_api("DELETE", f"/api/v1/admin/accounts/{account_id}", bearer=token), f"delete removed account {name}")
        results.append({
            "accountName": name,
            "accountId": account_id,
            "profile": extra.get("unidesk_codex_profile"),
            "action": "deleted",
            "reason": "removed-from-yaml",
            "valuesPrinted": False,
        })
    return results

def generate_api_key():
    alphabet = string.ascii_letters + string.digits
    return "sk-unidesk-codex-" + "".join(secrets.choice(alphabet) for _ in range(48))

def existing_sub2api_pool_api_key(token):
    try:
        existing = next((item for item in list_user_keys(token) if item.get("name") == POOL_API_KEY_NAME), None)
    except Exception:
        return None
    if not isinstance(existing, dict):
        return None
    key = existing.get("key")
    return key if isinstance(key, str) and key else None

def ensure_api_key_secret(group_id, token):
    existing = None
    try:
        existing = decode_secret_value(POOL_API_KEY_SECRET_NAME, POOL_API_KEY_SECRET_KEY)
    except Exception:
        existing = None
    reused = None if existing else existing_sub2api_pool_api_key(token)
    api_key = existing or reused or generate_api_key()
    if existing:
        secret_action = "kept-existing"
    elif reused:
        secret_action = "reused-existing-sub2api-key"
    else:
        secret_action = "created"
    if RUNTIME_MODE == "host-docker":
        env_action = "kept-existing" if existing else write_host_env_value(POOL_API_KEY_SECRET_KEY, api_key)
        return api_key, secret_action, f"host-docker-env:{env_action};source={HOST_DOCKER_ENV_PATH};key={POOL_API_KEY_SECRET_KEY};valuesPrinted=false"
    manifest = {
        "apiVersion": "v1",
        "kind": "Secret",
        "metadata": {
            "name": POOL_API_KEY_SECRET_NAME,
            "namespace": NAMESPACE,
            "labels": {
                "app.kubernetes.io/name": "sub2api",
                "app.kubernetes.io/part-of": "platform-infra",
                "app.kubernetes.io/managed-by": "unidesk",
                "unidesk.ai/secret-purpose": "sub2api-codex-pool-api-key",
            },
        },
        "type": "Opaque",
        "stringData": {
            POOL_API_KEY_SECRET_KEY: api_key,
            "GROUP_ID": str(group_id),
            "GROUP_NAME": POOL_GROUP_NAME,
            "SERVICE_DNS": SERVICE_DNS,
        },
    }
    proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], json.dumps(manifest))
    if proc.returncode != 0:
        raise RuntimeError(f"apply API key secret failed: {text(proc.stderr, 1000)}")
    return api_key, secret_action, text(proc.stdout, 1000)

def pool_api_key_secret_location():
    if RUNTIME_MODE == "host-docker":
        return f"{HOST_DOCKER_ENV_PATH}.{POOL_API_KEY_SECRET_KEY}"
    return f"{NAMESPACE}/{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY}"

def apply_sentinel_manifest(manifest):
    if not TARGET_SENTINEL_ENABLED:
        return {
            "ok": True,
            "action": "skipped-target-disabled",
            "valuesPrinted": False,
        }
    if not isinstance(manifest, str) or not manifest.strip():
        return {
            "ok": False,
            "action": "missing-manifest",
            "valuesPrinted": False,
        }
    proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], manifest)
    if proc.returncode != 0:
        return {
            "ok": False,
            "action": "apply-failed",
            "stdoutTail": text(proc.stdout, 2000),
            "stderrTail": text(proc.stderr, 4000),
            "valuesPrinted": False,
        }
    status = sentinel_runtime_status()
    return {
        "ok": status.get("ok") is True,
        "action": "applied",
        "stdoutTail": text(proc.stdout, 2000),
        "runtime": status,
        "valuesPrinted": False,
    }

def safe_kube_json(args, label):
    proc = kubectl([*args, "-o", "json"])
    if proc.returncode != 0:
        return None, {"label": label, "exitCode": proc.returncode, "stderrTail": text(proc.stderr, 1000)}
    try:
        return json.loads(proc.stdout.decode("utf-8")), None
    except Exception as exc:
        return None, {"label": label, "exitCode": proc.returncode, "error": str(exc), "stdoutTail": text(proc.stdout, 1000)}

def sentinel_runtime_status():
    if not TARGET_SENTINEL_ENABLED:
        return {
            "ok": True,
            "action": "skipped-target-disabled",
            "desired": {
                "monitorEnabled": SENTINEL_CONFIG.get("monitor", {}).get("enabled"),
                "actionsEnabled": SENTINEL_CONFIG.get("actions", {}).get("enabled"),
            },
            "valuesPrinted": False,
        }
    cfg = SENTINEL_CONFIG
    cronjob_name = cfg.get("cronJobName")
    secret_name = cfg.get("credentialsSecretName")
    configmap_name = cfg.get("configMapName")
    state_name = cfg.get("stateConfigMapName")
    cronjob, cronjob_error = safe_kube_json(["-n", NAMESPACE, "get", "cronjob", cronjob_name], f"cronjob/{cronjob_name}")
    secret, secret_error = safe_kube_json(["-n", NAMESPACE, "get", "secret", secret_name], f"secret/{secret_name}")
    configmap, configmap_error = safe_kube_json(["-n", NAMESPACE, "get", "configmap", configmap_name], f"configmap/{configmap_name}")
    state_cm, state_error = safe_kube_json(["-n", NAMESPACE, "get", "configmap", state_name], f"configmap/{state_name}")
    state = None
    if isinstance(state_cm, dict):
        raw_state = (state_cm.get("data") or {}).get("state.json")
        if isinstance(raw_state, str) and raw_state:
            try:
                state = json.loads(raw_state)
            except Exception as exc:
                state = {"parseError": str(exc)}
    accounts = (state.get("accounts") or {}) if isinstance(state, dict) else {}
    quarantined = []
    recent_accounts = []
    for name, account_state in accounts.items():
        if not isinstance(account_state, dict):
            continue
        quarantine = account_state.get("quarantine")
        if isinstance(quarantine, dict) and quarantine.get("active") is True:
            quarantined.append({
                "accountName": name,
                "until": quarantine.get("until"),
                "applied": quarantine.get("applied"),
                "reason": quarantine.get("reason"),
                "failureKind": quarantine.get("failureKind"),
                "errorDetails": quarantine.get("errorDetails"),
                "intervalMinutes": quarantine.get("intervalMinutes"),
                "sentinelProtect": quarantine.get("sentinelProtect"),
            })
        last_probe = account_state.get("lastProbe")
        if isinstance(last_probe, dict):
            recent_accounts.append({
                "accountName": name,
                "lastProbeAt": account_state.get("lastProbeAt"),
                "lastStatus": account_state.get("lastStatus"),
                "nextProbeAfter": account_state.get("nextProbeAfter"),
                "ok": last_probe.get("ok"),
                "purpose": last_probe.get("purpose"),
                "httpStatus": last_probe.get("httpStatus"),
                "durationMs": last_probe.get("durationMs"),
                "markerMatched": last_probe.get("markerMatched"),
                "sentinelProtect": last_probe.get("sentinelProtect") or account_state.get("sentinelProtectConfig"),
                "outputHash": last_probe.get("outputHash"),
                "outputPreview": last_probe.get("outputPreview"),
                "responseBodyHash": last_probe.get("responseBodyHash"),
                "responseBodyPreview": last_probe.get("responseBodyPreview"),
                "error": last_probe.get("error"),
                "errorDetails": last_probe.get("errorDetails"),
                "usage": last_probe.get("usage"),
                "failureKind": last_probe.get("failureKind"),
                "requestShape": last_probe.get("requestShape"),
                "action": last_probe.get("action"),
            })
    recent_accounts.sort(key=lambda item: item.get("lastProbeAt") or "")
    last_run = state.get("lastRun") if isinstance(state, dict) else None
    cronjob_spec = cronjob.get("spec") if isinstance(cronjob, dict) else {}
    secret_data = secret.get("data") if isinstance(secret, dict) else {}
    configmap_data = configmap.get("data") if isinstance(configmap, dict) else {}
    ok = cronjob is not None and secret is not None and configmap is not None
    return {
        "ok": ok,
        "desired": {
            "monitorEnabled": cfg.get("monitor", {}).get("enabled"),
            "actionsEnabled": cfg.get("actions", {}).get("enabled"),
            "schedule": cfg.get("schedule"),
            "cronJobName": cronjob_name,
            "configMapName": configmap_name,
            "credentialsSecretName": secret_name,
            "stateConfigMapName": state_name,
        },
        "cronJob": {
            "exists": cronjob is not None,
            "schedule": cronjob_spec.get("schedule") if isinstance(cronjob_spec, dict) else None,
            "suspend": cronjob_spec.get("suspend") if isinstance(cronjob_spec, dict) else None,
            "lastScheduleTime": (cronjob.get("status") or {}).get("lastScheduleTime") if isinstance(cronjob, dict) else None,
            "active": len((cronjob.get("status") or {}).get("active") or []) if isinstance(cronjob, dict) else None,
            "error": cronjob_error,
        },
        "secret": {
            "exists": secret is not None,
            "profileSecretPresent": isinstance(secret_data, dict) and "profiles.json" in secret_data,
            "valuesPrinted": False,
            "error": secret_error,
        },
        "configMap": {
            "exists": configmap is not None,
            "configPresent": isinstance(configmap_data, dict) and "config.json" in configmap_data,
            "runnerPresent": isinstance(configmap_data, dict) and "sentinel.py" in configmap_data,
            "error": configmap_error,
        },
        "state": {
            "exists": state_cm is not None,
            "accountCount": len(accounts),
            "quarantinedCount": len(quarantined),
            "quarantined": quarantined[-10:],
            "recentAccounts": recent_accounts[-12:],
            "lastRun": last_run,
            "error": state_error,
        },
        "valuesPrinted": False,
    }

def parse_epoch_z(value):
    if not isinstance(value, str) or not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except Exception:
        return None

def sentinel_state_object():
    if not TARGET_SENTINEL_ENABLED:
        return None, None
    state_name = SENTINEL_CONFIG.get("stateConfigMapName")
    if not state_name:
        return None, None
    obj, err = safe_kube_json(["-n", NAMESPACE, "get", "configmap", state_name], f"configmap/{state_name}")
    if not isinstance(obj, dict):
        return None, None
    raw_state = (obj.get("data") or {}).get("state.json")
    if not isinstance(raw_state, str) or not raw_state:
        return obj, None
    try:
        return obj, json.loads(raw_state)
    except Exception:
        return obj, None

def active_sentinel_quarantine_names():
    if not TARGET_SENTINEL_ENABLED:
        return set()
    _, state = sentinel_state_object()
    if not isinstance(state, dict):
        return set()
    accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {}
    names = set()
    for name, account_state in accounts_state.items():
        if not isinstance(name, str) or not isinstance(account_state, dict):
            continue
        quarantine = account_state.get("quarantine")
        if isinstance(quarantine, dict) and quarantine.get("active") is True and quarantine.get("applied") is True:
            names.add(name)
    return names

def default_sentinel_state():
    return {"version": 1, "accounts": {}, "ledger": {}, "history": []}

def clamp_sentinel_freezes_for_config(state, now):
    freeze_config = SENTINEL_CONFIG.get("freeze") if isinstance(SENTINEL_CONFIG.get("freeze"), dict) else {}
    try:
        max_interval = int(freeze_config.get("maxTtlMinutes") or 10)
    except Exception:
        max_interval = 10
    accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {}
    now_epoch = time.time()
    items = []
    for name, account_state in accounts_state.items():
        if not isinstance(name, str) or not isinstance(account_state, dict):
            continue
        quarantine = account_state.get("quarantine")
        if not isinstance(quarantine, dict) or quarantine.get("active") is not True or quarantine.get("applied") is not True:
            continue
        try:
            interval = int(quarantine.get("intervalMinutes") or 0)
        except Exception:
            interval = 0
        until_epoch = parse_epoch_z(quarantine.get("until"))
        old_until = quarantine.get("until")
        if interval <= max_interval and (until_epoch is None or until_epoch <= now_epoch + max_interval * 60):
            continue
        quarantine["previousIntervalMinutes"] = interval
        quarantine["intervalMinutes"] = max_interval
        quarantine["until"] = now
        quarantine["clampedAt"] = now
        quarantine["clampedBy"] = "sync-freeze-max-ttl"
        account_state["nextProbeAfter"] = now
        items.append({
            "accountName": name,
            "previousIntervalMinutes": interval,
            "maxIntervalMinutes": max_interval,
            "previousUntil": old_until,
            "nextProbeAfter": now,
        })
    return items

def parse_iso_epoch(value):
    if not isinstance(value, str) or not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except Exception:
        return None

def profile_success_max_interval(profile):
    cadence = SENTINEL_CONFIG.get("cadence") if isinstance(SENTINEL_CONFIG.get("cadence"), dict) else {}
    legacy = cadence.get("successMaxIntervalMinutes")
    if legacy is None:
        legacy = cadence.get("trustedSuccessMaxIntervalMinutes") or cadence.get("untrustedSuccessMaxIntervalMinutes") or 1
    if profile.get("trustUpstream") is True:
        value = cadence.get("trustedSuccessMaxIntervalMinutes") or legacy
    else:
        value = cadence.get("untrustedSuccessMaxIntervalMinutes") or legacy
    try:
        return int(value)
    except Exception:
        return int(legacy)

def clamp_sentinel_success_cadence_for_config(state, profiles, now):
    accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {}
    profile_map = {item.get("accountName"): item for item in profiles if isinstance(item, dict) and isinstance(item.get("accountName"), str)}
    now_epoch = time.time()
    items = []
    for name, profile in profile_map.items():
        account_state = accounts_state.get(name)
        if not isinstance(account_state, dict):
            continue
        quarantine = account_state.get("quarantine")
        if isinstance(quarantine, dict) and quarantine.get("active") is True:
            account_state["trustUpstream"] = profile.get("trustUpstream") is True
            account_state["sentinelProtectConfig"] = profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False}
            account_state["successMaxIntervalMinutes"] = profile_success_max_interval(profile)
            continue
        try:
            interval = int(account_state.get("successIntervalMinutes") or 0)
        except Exception:
            interval = 0
        next_epoch = parse_iso_epoch(account_state.get("nextProbeAfter"))
        max_interval = profile_success_max_interval(profile)
        account_state["trustUpstream"] = profile.get("trustUpstream") is True
        account_state["sentinelProtectConfig"] = profile.get("sentinelProtect") if isinstance(profile.get("sentinelProtect"), dict) else {"enabled": False}
        account_state["successMaxIntervalMinutes"] = max_interval
        if interval <= max_interval and (next_epoch is None or next_epoch <= now_epoch + max_interval * 60):
            continue
        old_next = account_state.get("nextProbeAfter")
        account_state["previousSuccessIntervalMinutes"] = interval
        account_state["successIntervalMinutes"] = min(interval, max_interval) if interval > 0 else interval
        account_state["nextProbeAfter"] = now
        account_state["cadenceClampedAt"] = now
        account_state["cadenceClampedBy"] = "sync-success-max-interval"
        items.append({
            "accountName": name,
            "trustUpstream": profile.get("trustUpstream") is True,
            "previousSuccessIntervalMinutes": interval,
            "maxIntervalMinutes": max_interval,
            "previousNextProbeAfter": old_next,
            "nextProbeAfter": now,
        })
    return items

def update_sentinel_state_configmap(obj, state):
    state_name = SENTINEL_CONFIG.get("stateConfigMapName")
    if not state_name:
        return {"ok": False, "reason": "state-configmap-missing"}
    state_json = json.dumps(state, ensure_ascii=False, indent=2)
    manifest = {
        "apiVersion": "v1",
        "kind": "ConfigMap",
        "metadata": {
            "name": state_name,
            "namespace": NAMESPACE,
            "labels": {
                "app.kubernetes.io/name": SERVICE_NAME,
                "app.kubernetes.io/component": "account-sentinel",
                "app.kubernetes.io/managed-by": "unidesk-platform-infra",
            },
        },
        "data": {"state.json": state_json},
    }
    proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], json.dumps(manifest))
    action = "applied" if isinstance(obj, dict) else "created"
    if proc.returncode != 0:
        return {"ok": False, "reason": f"{action}-failed", "stderrTail": text(proc.stderr, 2000)}
    return {"ok": True, "action": action}

def ensure_sentinel_state_for_sync(account_results, pending_only=False):
    if not sentinel_quality_gate_enabled():
        return {"ok": True, "skipped": True, "reason": "sentinel-quality-gate-disabled", "changedCount": 0, "items": [], "valuesPrinted": False}
    state_obj, state = sentinel_state_object()
    if not isinstance(state, dict):
        state = default_sentinel_state()
    state.setdefault("version", 1)
    accounts_state = state.setdefault("accounts", {})
    if not isinstance(accounts_state, dict):
        accounts_state = {}
        state["accounts"] = accounts_state
    now = utc_iso()
    items = []
    clamped_items = [] if pending_only else clamp_sentinel_freezes_for_config(state, now)
    cadence_clamped_items = [] if pending_only else clamp_sentinel_success_cadence_for_config(state, [item.get("profileConfig") for item in account_results if isinstance(item.get("profileConfig"), dict)], now)
    changed_count = 0
    fingerprint_only_count = 0
    for item in account_results:
        name = item.get("accountName")
        if not isinstance(name, str) or not name:
            continue
        account_state = accounts_state.setdefault(name, {})
        if not isinstance(account_state, dict):
            account_state = {}
            accounts_state[name] = account_state
        fingerprint_value = item.get("sentinelProbeConfigFingerprint")
        if isinstance(fingerprint_value, str) and fingerprint_value:
            account_state["probeConfigFingerprint"] = fingerprint_value
        if item.get("sentinelProbeRequired") is not True:
            fingerprint_only_count += 1
            continue
        changed_count += 1
        reasons = item.get("sentinelChangeReasons") if isinstance(item.get("sentinelChangeReasons"), list) else []
        quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else None
        if not (isinstance(quarantine, dict) and quarantine.get("active") is True):
            account_state["quarantine"] = {
                "active": False,
                "reason": "yaml-account-change-pending-sentinel-probe",
                "lastPendingAt": now,
                "changeReasons": reasons,
            }
        account_state["nextProbeAfter"] = now
        account_state["successStreak"] = 0
        account_state["successIntervalMinutes"] = 0
        profile_config = item.get("profileConfig") if isinstance(item.get("profileConfig"), dict) else {}
        account_state["trustUpstream"] = profile_config.get("trustUpstream") is True
        account_state["sentinelProtectConfig"] = profile_config.get("sentinelProtect") if isinstance(profile_config.get("sentinelProtect"), dict) else {"enabled": False}
        account_state["successMaxIntervalMinutes"] = profile_success_max_interval(profile_config)
        account_state["lastStatus"] = "pending-sentinel-quality-gate"
        account_state["qualityGate"] = {
            "pending": True,
            "reason": "yaml-account-change",
            "changeReasons": reasons,
            "markedAt": now,
            "pendingOnly": pending_only,
            "defaultFrozen": False,
        }
        items.append({"accountName": name, "changeReasons": reasons, "nextProbeAfter": now, "defaultFrozen": False, "defaultSchedulable": True, "pendingOnly": pending_only})
    if changed_count <= 0 and len(clamped_items) <= 0 and len(cadence_clamped_items) <= 0:
        return {"ok": True, "skipped": False, "reason": "no-new-or-changed-accounts", "changedCount": 0, "fingerprintOnlyCount": fingerprint_only_count, "clampedCount": 0, "cadenceClampedCount": 0, "items": [], "valuesPrinted": False}
    update = update_sentinel_state_configmap(state_obj, state)
    if pending_only and changed_count > 0:
        reason = "new-or-changed-accounts-pending-probe-prepared-default-schedulable"
    elif changed_count > 0 and (len(clamped_items) > 0 or len(cadence_clamped_items) > 0):
        reason = "new-or-changed-accounts-default-schedulable-and-sentinel-cadence-clamped"
    elif changed_count > 0:
        reason = "new-or-changed-accounts-default-schedulable"
    elif len(cadence_clamped_items) > 0:
        reason = "success-cadence-clamped-to-current-config"
    else:
        reason = "freeze-backoff-clamped-to-current-config"
    return {
        "ok": update.get("ok") is True,
        "skipped": False,
        "reason": reason,
        "changedCount": changed_count,
        "fingerprintOnlyCount": fingerprint_only_count,
        "clampedCount": len(clamped_items),
        "cadenceClampedCount": len(cadence_clamped_items),
        "pendingOnly": pending_only,
        "items": items,
        "clampedItems": clamped_items,
        "cadenceClampedItems": cadence_clamped_items,
        "update": update,
        "valuesPrinted": False,
    }

def sentinel_state_summary():
    _, state = sentinel_state_object()
    if not isinstance(state, dict):
        return {"exists": False, "valuesPrinted": False}
    accounts = state.get("accounts") if isinstance(state.get("accounts"), dict) else {}
    quarantined = []
    recent_accounts = []
    for name, account_state in accounts.items():
        if not isinstance(account_state, dict):
            continue
        quarantine = account_state.get("quarantine")
        if isinstance(quarantine, dict) and quarantine.get("active") is True:
            quarantined.append({
                "accountName": name,
                "until": quarantine.get("until"),
                "applied": quarantine.get("applied"),
                "reason": quarantine.get("reason"),
                "failureKind": quarantine.get("failureKind"),
                "errorDetails": quarantine.get("errorDetails"),
                "intervalMinutes": quarantine.get("intervalMinutes"),
                "sentinelProtect": quarantine.get("sentinelProtect"),
            })
        last_probe = account_state.get("lastProbe")
        if isinstance(last_probe, dict):
            recent_accounts.append({
                "accountName": name,
                "lastProbeAt": account_state.get("lastProbeAt"),
                "lastStatus": account_state.get("lastStatus"),
                "nextProbeAfter": account_state.get("nextProbeAfter"),
                "ok": last_probe.get("ok"),
                "purpose": last_probe.get("purpose"),
                "httpStatus": last_probe.get("httpStatus"),
                "durationMs": last_probe.get("durationMs"),
                "markerMatched": last_probe.get("markerMatched"),
                "sentinelProtect": last_probe.get("sentinelProtect") or account_state.get("sentinelProtectConfig"),
                "usage": last_probe.get("usage"),
                "responseBodyHash": last_probe.get("responseBodyHash"),
                "responseBodyPreview": last_probe.get("responseBodyPreview"),
                "error": last_probe.get("error"),
                "errorDetails": last_probe.get("errorDetails"),
                "failureKind": last_probe.get("failureKind"),
                "requestShape": last_probe.get("requestShape"),
                "action": last_probe.get("action"),
            })
    recent_accounts.sort(key=lambda item: item.get("lastProbeAt") or "")
    return {
        "exists": True,
        "accountCount": len(accounts),
        "quarantinedCount": len(quarantined),
        "quarantined": quarantined[-10:],
        "recentAccounts": recent_accounts[-12:],
        "lastRun": state.get("lastRun"),
        "valuesPrinted": False,
    }

def reassert_sentinel_freezes_after_sync(token):
    if not TARGET_SENTINEL_ENABLED:
        return {"ok": True, "skipped": True, "reason": "target-disabled", "items": [], "valuesPrinted": False}
    if (SENTINEL_CONFIG.get("actions") or {}).get("enabled") is not True:
        return {"ok": True, "skipped": True, "reason": "actions-disabled", "items": [], "valuesPrinted": False}
    _, state = sentinel_state_object()
    if not isinstance(state, dict):
        return {"ok": True, "skipped": True, "reason": "state-missing", "items": [], "valuesPrinted": False}
    accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {}
    accounts = list_accounts(token)
    by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)}
    items = []
    for name, account_state in accounts_state.items():
        if not isinstance(account_state, dict):
            continue
        quarantine = account_state.get("quarantine")
        if not isinstance(quarantine, dict) or quarantine.get("active") is not True or quarantine.get("applied") is not True:
            continue
        account = by_name.get(name)
        if not account or account.get("id") is None:
            items.append({"accountName": name, "ok": False, "reason": "account-not-found"})
            continue
        try:
            ensure_success(
                curl_api("POST", f"/api/v1/admin/accounts/{account['id']}/schedulable", bearer=token, payload={"schedulable": False}),
                f"reassert sentinel freeze for {name}",
            )
            items.append({"accountName": name, "accountId": account.get("id"), "ok": True, "until": quarantine.get("until")})
        except Exception as exc:
            items.append({"accountName": name, "accountId": account.get("id"), "ok": False, "error": str(exc)})
    return {
        "ok": all(item.get("ok") is True for item in items),
        "skipped": False,
        "items": items,
        "valuesPrinted": False,
    }

def list_user_keys(token):
    data = ensure_success(curl_api("GET", "/api/v1/keys?page=1&page_size=200", bearer=token), "list user keys")
    return extract_items(data)

def ensure_sub2api_api_key(token, api_key, group_id):
    keys = list_user_keys(token)
    existing = next((item for item in keys if item.get("key") == api_key), None)
    action = "kept-existing"
    if existing is None:
        existing = next((item for item in keys if item.get("name") == POOL_API_KEY_NAME), None)
    if existing is None or existing.get("key") != api_key:
        payload = {
            "name": POOL_API_KEY_NAME,
            "group_id": group_id,
            "custom_key": api_key,
            "quota": 0,
            "rate_limit_5h": 0,
            "rate_limit_1d": 0,
            "rate_limit_7d": 0,
        }
        created = ensure_success(curl_api("POST", "/api/v1/keys", bearer=token, payload=payload), "create pool API key")
        existing = created if isinstance(created, dict) else existing
        action = "created"
    elif existing.get("id") is not None and existing.get("group_id") != group_id:
        updated = ensure_success(curl_api("PUT", f"/api/v1/keys/{existing['id']}", bearer=token, payload={"name": POOL_API_KEY_NAME, "group_id": group_id}), "update pool API key group")
        existing = updated if isinstance(updated, dict) else existing
        action = "updated-group"
    return {
        "action": action,
        "id": existing.get("id") if isinstance(existing, dict) else None,
        "name": existing.get("name") if isinstance(existing, dict) else POOL_API_KEY_NAME,
        "groupId": existing.get("group_id") if isinstance(existing, dict) else group_id,
        "userId": existing.get("user_id") if isinstance(existing, dict) else None,
    }

def get_admin_user(token, user_id):
    data = ensure_success(curl_api("GET", f"/api/v1/admin/users/{user_id}", bearer=token), "get API key owner")
    if not isinstance(data, dict):
        raise RuntimeError("API key owner response is not an object")
    return data

def ensure_pool_owner_balance(token, user_id):
    user = get_admin_user(token, user_id)
    current = float(user.get("balance") or 0)
    if current >= MIN_OWNER_BALANCE_USD:
        return {
            "action": "kept-existing",
            "userId": user_id,
            "balanceBefore": current,
            "balanceAfter": current,
            "minimumBalanceUsd": MIN_OWNER_BALANCE_USD,
        }
    updated = ensure_success(curl_api("POST", f"/api/v1/admin/users/{user_id}/balance", bearer=token, payload={
        "balance": MIN_OWNER_BALANCE_USD,
        "operation": "set",
        "notes": "UniDesk Sub2API Codex pool internal API key bootstrap balance.",
    }), "set API key owner balance")
    after = float(updated.get("balance") or MIN_OWNER_BALANCE_USD) if isinstance(updated, dict) else MIN_OWNER_BALANCE_USD
    return {
        "action": "set",
        "userId": user_id,
        "balanceBefore": current,
        "balanceAfter": after,
        "minimumBalanceUsd": MIN_OWNER_BALANCE_USD,
    }

def ensure_pool_owner_concurrency(token, user_id):
    user = get_admin_user(token, user_id)
    try:
        current = int(user.get("concurrency") or 0)
    except Exception:
        current = 0
    if current >= MIN_OWNER_CONCURRENCY:
        return {
            "ok": True,
            "action": "kept-existing",
            "userId": user_id,
            "concurrencyBefore": current,
            "concurrencyAfter": current,
            "minimumConcurrency": MIN_OWNER_CONCURRENCY,
            "minimumConcurrencySource": MIN_OWNER_CONCURRENCY_SOURCE,
        }
    updated = ensure_success(curl_api("PUT", f"/api/v1/admin/users/{user_id}", bearer=token, payload={
        "concurrency": MIN_OWNER_CONCURRENCY,
    }), "set API key owner concurrency")
    try:
        after = int(updated.get("concurrency") or MIN_OWNER_CONCURRENCY) if isinstance(updated, dict) else MIN_OWNER_CONCURRENCY
    except Exception:
        after = MIN_OWNER_CONCURRENCY
    return {
        "ok": after >= MIN_OWNER_CONCURRENCY,
        "action": "set",
        "userId": user_id,
        "concurrencyBefore": current,
        "concurrencyAfter": after,
        "minimumConcurrency": MIN_OWNER_CONCURRENCY,
        "minimumConcurrencySource": MIN_OWNER_CONCURRENCY_SOURCE,
    }

def validate_gateway(api_key):
    resp = curl_api("GET", "/v1/models", bearer=api_key)
    parsed = resp.get("json")
    model_count = None
    if isinstance(parsed, dict) and isinstance(parsed.get("data"), list):
        model_count = len(parsed["data"])
    return {
        "ok": resp.get("ok"),
        "httpStatus": resp.get("httpStatus"),
        "transportExitCode": resp.get("transportExitCode"),
        "modelCount": model_count,
        "bodyPreview": text(resp.get("body", ""), 500) if not resp.get("ok") else "",
        "stderr": resp.get("stderr", ""),
        "method": "GET /v1/models",
        "serviceDns": SERVICE_DNS,
        "valuesPrinted": False,
    }

def response_output_preview(parsed):
    if not isinstance(parsed, dict):
        return ""
    if isinstance(parsed.get("output_text"), str):
        return parsed["output_text"][:240]
    output = parsed.get("output")
    if not isinstance(output, list):
        return ""
    parts = []
    for item in output:
        if not isinstance(item, dict):
            continue
        content = item.get("content")
        if not isinstance(content, list):
            continue
        for block in content:
            if not isinstance(block, dict):
                continue
            text_value = block.get("text")
            if isinstance(text_value, str) and text_value:
                parts.append(text_value)
    return "\\n".join(parts)[:240]

def request_log_evidence(request_id):
    proc = runtime_logs("5m", 800)
    stdout = proc.stdout.decode("utf-8", errors="replace")
    lines = [line for line in stdout.splitlines() if request_id in line]
    failovers = []
    final = None
    for line in lines:
        json_start = line.find("{")
        if json_start < 0:
            continue
        try:
            item = json.loads(line[json_start:])
        except Exception:
            continue
        if "upstream_failover_switching" in line:
            failovers.append({
                "accountId": item.get("account_id"),
                "upstreamStatus": item.get("upstream_status"),
                "switchCount": item.get("switch_count"),
                "maxSwitches": item.get("max_switches"),
            })
        if "http request completed" in line:
            final = {
                "accountId": item.get("account_id"),
                "statusCode": item.get("status_code"),
                "latencyMs": item.get("latency_ms"),
                "path": item.get("path"),
            }
    return {
        "requestId": request_id,
        "matchedLogLineCount": len(lines),
        "failovers": failovers,
        "final": final,
        "logsExitCode": proc.returncode,
        "logsStderr": text(proc.stderr, 1000),
    }

def recent_compact_gateway_evidence():
    proc = runtime_logs("6h", 2500)
    stdout = proc.stdout.decode("utf-8", errors="replace")
    failures = []
    successes = []
    failovers = []
    final_errors = []
    context_canceled = []
    for line in stdout.splitlines():
        if "/responses/compact" not in line and "remote_compact" not in line:
            continue
        json_start = line.find("{")
        if json_start < 0:
            continue
        try:
            item = json.loads(line[json_start:])
        except Exception:
            continue
        path = item.get("path")
        entry = {
            "requestId": item.get("request_id"),
            "clientRequestId": item.get("client_request_id"),
            "accountId": item.get("account_id"),
            "statusCode": item.get("status_code"),
            "upstreamStatus": item.get("upstream_status"),
            "latencyMs": item.get("latency_ms"),
            "path": path,
        }
        if "codex.remote_compact.failed" in line:
            failures.append(entry)
        elif "codex.remote_compact.succeeded" in line:
            successes.append(entry)
        elif "upstream_failover_switching" in line and path == "/responses/compact":
            failovers.append({
                **entry,
                "switchCount": item.get("switch_count"),
                "maxSwitches": item.get("max_switches"),
            })
        elif "http request completed" in line and path == "/responses/compact" and isinstance(item.get("status_code"), int) and item.get("status_code") >= 400:
            final_errors.append(entry)
        if "context canceled" in line and path == "/responses/compact":
            context_canceled.append(entry)
    return {
        "ok": True,
        "degraded": len(failures) > 0 or len(final_errors) > 0 or len(context_canceled) > 0,
        "window": "6h",
        "tailLines": 2500,
        "failureCount": len(failures),
        "successCount": len(successes),
        "failoverCount": len(failovers),
        "finalErrorCount": len(final_errors),
        "contextCanceledCount": len(context_canceled),
        "recentFailures": failures[-5:],
        "recentSuccesses": successes[-5:],
        "recentFailovers": failovers[-8:],
        "recentFinalErrors": final_errors[-5:],
        "recentContextCanceled": context_canceled[-5:],
        "logsExitCode": proc.returncode,
        "logsStderr": text(proc.stderr, 1000),
        "valuesPrinted": False,
    }

def is_ignored_probe_noise(entry):
    for value in (entry.get("requestId"), entry.get("clientRequestId")):
        if isinstance(value, str) and value.startswith("unidesk-400-model-probe-"):
            return True
    return False

def filter_ignored_probe_noise(items):
    return [item for item in items if not is_ignored_probe_noise(item)]

def ignored_probe_noise(items, section):
    result = []
    for item in items:
        if is_ignored_probe_noise(item):
            probe = dict(item)
            probe["section"] = section
            result.append(probe)
    return result

def group_by_request_id(items):
    grouped = {}
    for item in items:
        request_id = item.get("requestId")
        if not isinstance(request_id, str) or not request_id:
            continue
        grouped.setdefault(request_id, []).append(item)
    return grouped

def failover_budget_exhausted_evidence(failovers, final_errors):
    final_by_request = {}
    for item in final_errors:
        request_id = item.get("requestId")
        if isinstance(request_id, str) and request_id:
            final_by_request[request_id] = item
    exhausted = []
    for request_id, request_failovers in group_by_request_id(failovers).items():
        final = final_by_request.get(request_id)
        if not final:
            continue
        last = request_failovers[-1]
        switch_count = last.get("switchCount")
        max_switches = last.get("maxSwitches")
        final_status = final.get("statusCode")
        if (
            isinstance(switch_count, int)
            and isinstance(max_switches, int)
            and max_switches > 0
            and switch_count >= max_switches
            and isinstance(final_status, int)
            and final_status >= 500
        ):
            exhausted.append({
                "requestId": request_id,
                "clientRequestId": final.get("clientRequestId") or last.get("clientRequestId"),
                "path": final.get("path") or last.get("path"),
                "finalAccountId": final.get("accountId"),
                "finalStatusCode": final_status,
                "switchCount": switch_count,
                "maxSwitches": max_switches,
                "lastFailoverAccountId": last.get("accountId"),
                "lastUpstreamStatus": last.get("upstreamStatus"),
            })
    return exhausted

def recent_responses_gateway_evidence():
    proc = runtime_logs("6h", 2500)
    stdout = proc.stdout.decode("utf-8", errors="replace")
    failovers = []
    forward_failures = []
    final_errors = []
    context_canceled = []
    slow_final_errors = []
    for line in stdout.splitlines():
        if '"/responses"' not in line and '"/v1/responses"' not in line:
            continue
        json_start = line.find("{")
        if json_start < 0:
            continue
        try:
            item = json.loads(line[json_start:])
        except Exception:
            continue
        path = item.get("path")
        if path not in ("/responses", "/v1/responses"):
            continue
        entry = {
            "requestId": item.get("request_id"),
            "clientRequestId": item.get("client_request_id"),
            "accountId": item.get("account_id"),
            "statusCode": item.get("status_code"),
            "upstreamStatus": item.get("upstream_status"),
            "latencyMs": item.get("latency_ms"),
            "path": path,
        }
        if "upstream_failover_switching" in line:
            failovers.append({
                **entry,
                "switchCount": item.get("switch_count"),
                "maxSwitches": item.get("max_switches"),
            })
        elif "openai.forward_failed" in line:
            forward_failures.append({
                **entry,
                "errorPreview": text(str(item.get("error") or ""), 500),
                "fallbackErrorResponseWritten": item.get("fallback_error_response_written"),
                "upstreamErrorResponseAlreadyWritten": item.get("upstream_error_response_already_written"),
            })
        elif "http request completed" in line and isinstance(item.get("status_code"), int) and item.get("status_code") >= 400:
            final_errors.append(entry)
            latency_ms = item.get("latency_ms")
            if isinstance(latency_ms, int) and latency_ms >= 30000:
                slow_final_errors.append(entry)
        if "context canceled" in line:
            context_canceled.append(entry)
    visible_failovers = filter_ignored_probe_noise(failovers)
    visible_forward_failures = filter_ignored_probe_noise(forward_failures)
    visible_final_errors = filter_ignored_probe_noise(final_errors)
    visible_slow_final_errors = filter_ignored_probe_noise(slow_final_errors)
    visible_context_canceled = filter_ignored_probe_noise(context_canceled)
    failover_budget_exhausted = failover_budget_exhausted_evidence(visible_failovers, visible_final_errors)
    probe_noise = (
        ignored_probe_noise(failovers, "failovers")
        + ignored_probe_noise(forward_failures, "forwardFailures")
        + ignored_probe_noise(final_errors, "finalErrors")
        + ignored_probe_noise(slow_final_errors, "slowFinalErrors")
        + ignored_probe_noise(context_canceled, "contextCanceled")
    )
    return {
        "ok": True,
        "degraded": len(visible_forward_failures) > 0 or len(visible_final_errors) > 0 or len(visible_context_canceled) > 0,
        "window": "6h",
        "tailLines": 2500,
        "failoverCount": len(visible_failovers),
        "forwardFailureCount": len(visible_forward_failures),
        "finalErrorCount": len(visible_final_errors),
        "slowFinalErrorCount": len(visible_slow_final_errors),
        "contextCanceledCount": len(visible_context_canceled),
        "ignoredProbeNoiseCount": len(probe_noise),
        "failoverBudgetExhausted": failover_budget_exhausted[-8:],
        "rawCounts": {
            "failoverCount": len(failovers),
            "forwardFailureCount": len(forward_failures),
            "finalErrorCount": len(final_errors),
            "slowFinalErrorCount": len(slow_final_errors),
            "contextCanceledCount": len(context_canceled),
        },
        "recentFailovers": visible_failovers[-8:],
        "recentForwardFailures": visible_forward_failures[-8:],
        "recentFinalErrors": visible_final_errors[-8:],
        "recentSlowFinalErrors": visible_slow_final_errors[-5:],
        "recentContextCanceled": visible_context_canceled[-5:],
        "recentProbeNoise": probe_noise[-5:],
        "logsExitCode": proc.returncode,
        "logsStderr": text(proc.stderr, 1000),
        "valuesPrinted": False,
    }

def validate_gateway_responses(api_key):
    request_id = "unidesk-codex-pool-validate-" + str(int(time.time() * 1000))
    payload = {
        "model": RESPONSES_SMOKE_MODEL,
        "input": "Reply exactly: unidesk-sub2api-validate-ok",
        "stream": False,
        "store": False,
        "max_output_tokens": 32,
    }
    body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
    script = r'''
set -eu
token="$1"
request_id="$2"
url="$3"
tmp="$(mktemp)"
trap 'rm -f "$tmp"' EXIT
cat > "$tmp"
curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X POST \
  -H "Authorization: Bearer $token" \
  -H 'Content-Type: application/json' \
  -H "X-Request-ID: $request_id" \
  -H "OpenAI-Client-Request-ID: $request_id" \
  --data-binary @"$tmp" \
  "$url"
'''
    started = time.time()
    if RUNTIME_MODE == "host-docker":
        if not isinstance(HOST_DOCKER_APP_PORT, int):
            raise RuntimeError("host-docker app port missing")
        proc = run(["sh", "-c", script, "sh", api_key, request_id, f"http://127.0.0.1:{HOST_DOCKER_APP_PORT}/v1/responses"], body)
    else:
        proc = run([
            "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
            "--", "sh", "-c", script, "sh", api_key, request_id, "http://127.0.0.1:8080/v1/responses",
        ], body)
    resp = parse_curl_output(proc)
    evidence = request_log_evidence(request_id)
    parsed = resp.get("json")
    failover_count = len(evidence.get("failovers") or [])
    return {
        "ok": resp.get("ok"),
        "degraded": failover_count > 0,
        "outcome": "succeeded-with-failover" if resp.get("ok") and failover_count > 0 else ("succeeded" if resp.get("ok") else "failed"),
        "httpStatus": resp.get("httpStatus"),
        "transportExitCode": resp.get("transportExitCode"),
        "method": "POST /v1/responses",
        "model": RESPONSES_SMOKE_MODEL,
        "requestId": request_id,
        "durationMs": int((time.time() - started) * 1000),
        "outputTextPreview": response_output_preview(parsed),
        "bodyPreview": "" if resp.get("ok") else text(resp.get("body", ""), 800),
        "stderr": resp.get("stderr", ""),
        "evidence": evidence,
        "valuesPrinted": False,
    }

def bool_value(value):
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        if value.lower() == "true":
            return True
        if value.lower() == "false":
            return False
    return False

def normalize_temp_unschedulable_credentials(credentials):
    if not isinstance(credentials, dict):
        credentials = {}
    enabled = bool_value(credentials.get("temp_unschedulable_enabled"))
    raw_rules = credentials.get("temp_unschedulable_rules")
    if isinstance(raw_rules, str):
        try:
            raw_rules = json.loads(raw_rules)
        except json.JSONDecodeError:
            raw_rules = []
    rules = []
    if isinstance(raw_rules, list):
        for rule in raw_rules:
            if not isinstance(rule, dict):
                continue
            error_code = rule.get("error_code", rule.get("statusCode"))
            duration_minutes = rule.get("duration_minutes", rule.get("durationMinutes"))
            keywords = rule.get("keywords")
            if not isinstance(error_code, int) or not isinstance(duration_minutes, int) or not isinstance(keywords, list):
                continue
            clean_keywords = [item for item in keywords if isinstance(item, str) and item.strip()]
            if not clean_keywords:
                continue
            description = rule.get("description") if isinstance(rule.get("description"), str) else ""
            rules.append({
                "error_code": error_code,
                "keywords": clean_keywords,
                "duration_minutes": duration_minutes,
                "description": description,
            })
    return {
        "enabled": enabled,
        "rules": rules,
    }

def summarize_temp_unschedulable_rules(rules):
    return [{
        "errorCode": rule.get("error_code"),
        "durationMinutes": rule.get("duration_minutes"),
        "keywordCount": len(rule.get("keywords") or []),
        "keywords": rule.get("keywords") or [],
        "hasDescription": bool(rule.get("description")),
    } for rule in rules]

def success_body_reclassification_requirement():
    for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE):
        expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name])
        if expected["enabled"] is not True:
            continue
        for rule in expected["rules"]:
            error_code = rule.get("error_code")
            keywords = rule.get("keywords") or []
            if isinstance(error_code, int) and 200 <= error_code < 300 and keywords:
                return {
                    "required": True,
                    "sourceAccountName": name,
                    "statusCode": error_code,
                    "keywords": keywords,
                    "representativeKeyword": keywords[0],
                    "durationMinutes": rule.get("duration_minutes"),
                }
    return {
        "required": False,
        "sourceAccountName": None,
        "statusCode": None,
        "keywords": [],
        "representativeKeyword": None,
        "durationMinutes": None,
    }

def model_routing_400_failover_requirement():
    preferred = ["invalid_encrypted_content", "encrypted content", "could not be verified", "could not be decrypted", "暂不支持", "可用模型", "unsupported model", "model not supported", "does not support", "not supported", "model_not_found", "no available channel for model"]
    for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE):
        expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name])
        if expected["enabled"] is not True:
            continue
        for rule in expected["rules"]:
            error_code = rule.get("error_code")
            keywords = rule.get("keywords") or []
            if error_code != 400 or not keywords:
                continue
            representative_keyword = next((item for item in preferred if item in keywords), keywords[0])
            return {
                "required": True,
                "sourceAccountName": name,
                "statusCode": error_code,
                "keywords": keywords,
                "representativeKeyword": representative_keyword,
                "durationMinutes": rule.get("duration_minutes"),
            }
    return {
        "required": False,
        "sourceAccountName": None,
        "statusCode": None,
        "keywords": [],
        "representativeKeyword": None,
        "durationMinutes": None,
    }

def delete_probe_resource(token, method, path, label):
    if not path:
        return {"label": label, "ok": True, "skipped": True}
    resp = curl_api(method, path, bearer=token)
    ok = resp.get("ok") is True or resp.get("httpStatus") in (404, 410)
    return {
        "label": label,
        "ok": ok,
        "method": method,
        "path": path,
        "httpStatus": resp.get("httpStatus"),
        "transportExitCode": resp.get("transportExitCode"),
        "bodyPreview": "" if ok else text(resp.get("body", ""), 500),
        "valuesPrinted": False,
    }

def validate_runtime_capabilities(token):
    success_body = success_body_reclassification_requirement()
    model_routing_400 = model_routing_400_failover_requirement()
    return {
        "ok": success_body.get("required") is not True and model_routing_400.get("required") is True,
        "runtimeImage": app_pod_runtime_image(),
        "successBodyReclassification": {
            "ok": success_body.get("required") is not True,
            "required": success_body.get("required"),
            "outcome": "not-required-by-yaml" if success_body.get("required") is not True else "requires-source-review-and-real-traffic-evidence",
            "requirement": success_body,
            "valuesPrinted": False,
        },
        "modelRouting400Failover": {
            "ok": model_routing_400.get("required") is True,
            "required": model_routing_400.get("required"),
            "outcome": "yaml-rules-present-runtime-observed-by-real-traffic",
            "requirement": model_routing_400,
            "evidence": "Use Sub2API source review plus validation.gatewayResponsesRecent and Artificer/real request ids for runtime proof; default validate does not create mock upstreams or temporary failover accounts.",
            "valuesPrinted": False,
        },
        "valuesPrinted": False,
    }

def app_pod_runtime_image():
    if RUNTIME_MODE == "host-docker":
        proc = docker(["inspect", HOST_DOCKER_APP_CONTAINER])
        if proc.returncode != 0:
            return {
                "container": HOST_DOCKER_APP_CONTAINER,
                "error": text(proc.stderr, 1000) or text(proc.stdout, 1000),
            }
        try:
            data = json.loads(proc.stdout.decode("utf-8"))
            item = data[0] if isinstance(data, list) and data else {}
        except Exception as exc:
            return {"container": HOST_DOCKER_APP_CONTAINER, "error": str(exc)}
        state = item.get("State") if isinstance(item, dict) and isinstance(item.get("State"), dict) else {}
        health = state.get("Health") if isinstance(state.get("Health"), dict) else {}
        config = item.get("Config") if isinstance(item, dict) and isinstance(item.get("Config"), dict) else {}
        return {
            "container": HOST_DOCKER_APP_CONTAINER,
            "id": (item.get("Id") or "")[:12] if isinstance(item.get("Id"), str) else None,
            "image": config.get("Image"),
            "imageID": item.get("Image"),
            "ready": state.get("Running") is True and (not health or health.get("Status") in (None, "healthy")),
            "restartCount": item.get("RestartCount"),
            "startedAt": state.get("StartedAt"),
            "health": health.get("Status"),
        }
    try:
        pod = kube_json(["-n", NAMESPACE, "get", "pod", APP_POD], f"pod/{APP_POD}")
        spec_containers = ((pod.get("spec") or {}).get("containers") or []) if isinstance(pod, dict) else []
        status_containers = ((pod.get("status") or {}).get("containerStatuses") or []) if isinstance(pod, dict) else []
        spec = next((item for item in spec_containers if item.get("name") == "sub2api"), spec_containers[0] if spec_containers else {})
        status = next((item for item in status_containers if item.get("name") == "sub2api"), status_containers[0] if status_containers else {})
        return {
            "pod": APP_POD,
            "image": spec.get("image"),
            "imageID": status.get("imageID"),
            "ready": status.get("ready"),
            "restartCount": status.get("restartCount"),
        }
    except Exception as exc:
        return {"pod": APP_POD, "error": str(exc)}

def get_account_detail(token, account):
    account_id = account.get("id") if isinstance(account, dict) else None
    if account_id is None:
        return account
    data = ensure_success(curl_api("GET", f"/api/v1/admin/accounts/{account_id}", bearer=token), f"get account {account.get('name')}")
    return data if isinstance(data, dict) else account

def account_temp_unschedulable_status(token):
    accounts = list_accounts(token)
    by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)}
    items = []
    missing = []
    mismatched = []
    enabled_names = []
    for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE):
        expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name])
        account = by_name.get(name)
        if account is None:
            missing.append(name)
            items.append({
                "accountName": name,
                "accountId": None,
                "expectedEnabled": expected["enabled"],
                "runtimeEnabled": None,
                "expectedRuleCount": len(expected["rules"]),
                "runtimeRuleCount": None,
                "ok": False,
            })
            continue
        detail = get_account_detail(token, account)
        credentials = detail.get("credentials") if isinstance(detail.get("credentials"), dict) else {}
        runtime = normalize_temp_unschedulable_credentials(credentials)
        temp_until = detail.get("temp_unschedulable_until") or detail.get("tempUnschedulableUntil")
        temp_reason = detail.get("temp_unschedulable_reason") or detail.get("tempUnschedulableReason") or ""
        ok = runtime == expected
        if expected["enabled"]:
            enabled_names.append(name)
        if not ok:
            mismatched.append(name)
        items.append({
            "accountName": name,
            "accountId": account.get("id"),
            "expectedEnabled": expected["enabled"],
            "runtimeEnabled": runtime["enabled"],
            "expectedRuleCount": len(expected["rules"]),
            "runtimeRuleCount": len(runtime["rules"]),
            "expectedRules": summarize_temp_unschedulable_rules(expected["rules"]),
            "runtimeRules": summarize_temp_unschedulable_rules(runtime["rules"]),
            "status": account.get("status"),
            "schedulable": account.get("schedulable"),
            "tempUnschedulableUntil": temp_until,
            "tempUnschedulableReasonPreview": text(str(temp_reason), 500) if temp_reason else "",
            "tempUnschedulableSet": temp_until is not None or bool(temp_reason),
            "ok": ok,
        })
    return {
        "ok": len(missing) == 0 and len(mismatched) == 0,
        "desired": len(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE),
        "enabled": enabled_names,
        "missing": missing,
        "mismatched": mismatched,
        "items": items,
        "valuesPrinted": False,
    }

def account_capacity_status(token):
    accounts = list_accounts(token)
    by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)}
    items = []
    missing = []
    mismatched = []
    expected_capacity_total = 0
    runtime_concurrency_total = 0
    schedulable_runtime_concurrency_total = 0
    available_runtime_concurrency_total = 0
    temp_unschedulable_runtime_concurrency_total = 0
    for name in sorted(EXPECTED_ACCOUNT_CAPACITIES):
        expected = int(EXPECTED_ACCOUNT_CAPACITIES[name])
        expected_capacity_total += expected
        account = by_name.get(name)
        if account is None:
            missing.append(name)
            items.append({
                "accountName": name,
                "accountId": None,
                "expectedCapacity": expected,
                "runtimeConcurrency": None,
                "ok": False,
            })
            continue
        runtime = account.get("concurrency")
        runtime_int = runtime if isinstance(runtime, int) else 0
        runtime_concurrency_total += runtime_int
        if account.get("schedulable") is True:
            runtime_status = account.get("status")
            if runtime_status is None or runtime_status == "active":
                runtime_status_ok = True
            else:
                runtime_status_ok = False
            if runtime_status_ok:
                schedulable_runtime_concurrency_total += runtime_int
                temp_until = account.get("temp_unschedulable_until") or account.get("tempUnschedulableUntil")
                temp_reason = account.get("temp_unschedulable_reason") or account.get("tempUnschedulableReason")
                if temp_until is None and not bool(temp_reason):
                    available_runtime_concurrency_total += runtime_int
                else:
                    temp_unschedulable_runtime_concurrency_total += runtime_int
        ok = runtime == expected
        if not ok:
            mismatched.append(name)
        items.append({
            "accountName": name,
            "accountId": account.get("id"),
            "expectedCapacity": expected,
            "runtimeConcurrency": runtime,
            "priority": account.get("priority"),
            "status": account.get("status"),
            "schedulable": account.get("schedulable"),
            "ok": ok,
        })
    return {
        "ok": len(missing) == 0 and len(mismatched) == 0,
        "defaultAccountCapacity": POOL_DEFAULT_ACCOUNT_CAPACITY,
        "desired": len(EXPECTED_ACCOUNT_CAPACITIES),
        "totals": {
            "expectedCapacityTotal": expected_capacity_total,
            "runtimeConcurrencyTotal": runtime_concurrency_total,
            "schedulableRuntimeConcurrencyTotal": schedulable_runtime_concurrency_total,
            "availableRuntimeConcurrencyTotal": available_runtime_concurrency_total,
            "tempUnschedulableRuntimeConcurrencyTotal": temp_unschedulable_runtime_concurrency_total,
            "minOwnerConcurrency": MIN_OWNER_CONCURRENCY,
            "minOwnerConcurrencySource": MIN_OWNER_CONCURRENCY_SOURCE,
        },
        "missing": missing,
        "mismatched": mismatched,
        "items": items,
        "valuesPrinted": False,
    }

def account_load_factor_status(token):
    accounts = list_accounts(token)
    by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)}
    items = []
    missing = []
    mismatched = []
    for name in sorted(EXPECTED_ACCOUNT_LOAD_FACTORS):
        expected = int(EXPECTED_ACCOUNT_LOAD_FACTORS[name])
        account = by_name.get(name)
        if account is None:
            missing.append(name)
            items.append({
                "accountName": name,
                "accountId": None,
                "expectedLoadFactor": expected,
                "runtimeLoadFactor": None,
                "ok": False,
            })
            continue
        runtime_raw = account.get("load_factor")
        if runtime_raw is None:
            detail = get_account_detail(token, account)
            runtime_raw = detail.get("load_factor") if isinstance(detail, dict) else None
        try:
            runtime = int(runtime_raw)
        except Exception:
            runtime = runtime_raw
        ok = runtime == expected
        if not ok:
            mismatched.append(name)
        items.append({
            "accountName": name,
            "accountId": account.get("id"),
            "expectedLoadFactor": expected,
            "runtimeLoadFactor": runtime,
            "priority": account.get("priority"),
            "status": account.get("status"),
            "schedulable": account.get("schedulable"),
            "ok": ok,
        })
    return {
        "ok": len(missing) == 0 and len(mismatched) == 0,
        "defaultAccountLoadFactor": POOL_DEFAULT_ACCOUNT_LOAD_FACTOR,
        "desired": len(EXPECTED_ACCOUNT_LOAD_FACTORS),
        "missing": missing,
        "mismatched": mismatched,
        "items": items,
        "valuesPrinted": False,
    }

def account_ws_v2_status(token):
    accounts = list_accounts(token)
    by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)}
    items = []
    missing = []
    mismatched = []
    enabled_names = []
    unschedulable_enabled = []
    schedulable_enabled = []
    for name in sorted(EXPECTED_ACCOUNT_WS_MODES):
        expected_mode = EXPECTED_ACCOUNT_WS_MODES[name]
        expected_enabled = expected_mode not in (None, "", "off")
        account = by_name.get(name)
        if account is None:
            missing.append(name)
            items.append({
                "accountName": name,
                "accountId": None,
                "expectedMode": expected_mode,
                "expectedEnabled": expected_enabled,
                "runtimeMode": None,
                "runtimeEnabled": None,
                "ok": False,
            })
            continue
        extra = account.get("extra") if isinstance(account.get("extra"), dict) else {}
        runtime_mode = extra.get("openai_apikey_responses_websockets_v2_mode")
        runtime_enabled = extra.get("openai_apikey_responses_websockets_v2_enabled")
        schedulable = account.get("schedulable")
        if expected_mode == "off":
            ok = runtime_mode == "off" and runtime_enabled is False
        elif expected_mode is None:
            ok = runtime_mode in (None, "", "off") and runtime_enabled in (None, False)
        else:
            ok = runtime_mode == expected_mode and runtime_enabled is True
        if not ok:
            mismatched.append(name)
        if expected_enabled:
            enabled_names.append(name)
            if schedulable is True:
                schedulable_enabled.append(name)
            else:
                unschedulable_enabled.append(name)
        items.append({
            "accountName": name,
            "accountId": account.get("id"),
            "expectedMode": expected_mode,
            "expectedEnabled": expected_enabled,
            "runtimeMode": runtime_mode,
            "runtimeEnabled": runtime_enabled,
            "status": account.get("status"),
            "schedulable": schedulable,
            "ok": ok and ((not expected_enabled) or schedulable is True),
        })
    availability_ok = len(enabled_names) == 0 or len(schedulable_enabled) > 0
    return {
        "ok": len(missing) == 0 and len(mismatched) == 0 and availability_ok,
        "desired": len(EXPECTED_ACCOUNT_WS_MODES),
        "enabled": enabled_names,
        "schedulableEnabled": schedulable_enabled,
        "unschedulableEnabled": unschedulable_enabled,
        "missing": missing,
        "mismatched": mismatched,
        "items": items,
        "valuesPrinted": False,
    }

def api_key_preview(api_key):
    if len(api_key) <= 14:
        return "***"
    return api_key[:10] + "..." + api_key[-4:]

def secret_fingerprint(value):
    return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]

def run_sync():
    global MANUAL_ACCOUNT_PROTECTIONS
    payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8"))
    manual_accounts_payload = payload.get("manualAccounts") if isinstance(payload.get("manualAccounts"), dict) else {}
    resolved_manual_protections = manual_accounts_payload.get("protected")
    if not isinstance(resolved_manual_protections, list):
        raise RuntimeError("sync payload has no manualAccounts.protected binding plan")
    MANUAL_ACCOUNT_PROTECTIONS = resolved_manual_protections
    profiles = payload.get("profiles") or []
    prune_removed = bool(payload.get("pruneRemoved"))
    sentinel_payload = payload.get("sentinel") if isinstance(payload.get("sentinel"), dict) else {}
    if not profiles and not MANUAL_ACCOUNT_PROTECTIONS:
        raise RuntimeError("sync payload has no profiles and no manualAccounts.protected binding plan")
    admin_email, token, admin_compliance = login()
    group, group_action = ensure_group(token)
    group_id = group.get("id") if isinstance(group, dict) else None
    if group_id is None:
        raise RuntimeError("pool group id missing after ensure")
    existing_accounts = list_accounts(token)
    planned_account_results = planned_sentinel_account_results(profiles, existing_accounts)
    sentinel_quality_prepare = ensure_sentinel_state_for_sync(planned_account_results, True)
    if sentinel_quality_prepare.get("ok") is not True:
        raise RuntimeError("prepare sentinel pending probe failed: " + json.dumps(sentinel_quality_prepare, ensure_ascii=False))
    protected_frozen_names = active_sentinel_quarantine_names()
    account_results, pruned_account_results = ensure_accounts(token, profiles, group_id, prune_removed, protected_frozen_names, existing_accounts)
    manual_account_proxy_bindings = ensure_manual_account_proxy_bindings(token)
    manual_account_group_bindings = ensure_manual_account_group_bindings(token, group_id)
    manual_account_protections = manual_account_protection_status(token, group_id)
    capacity_status = account_capacity_status(token)
    load_factor_status = account_load_factor_status(token)
    ws_v2_status = account_ws_v2_status(token)
    temp_unschedulable_status = account_temp_unschedulable_status(token)
    api_key, secret_action, secret_apply_stdout = ensure_api_key_secret(group_id, token)
    api_key_result = ensure_sub2api_api_key(token, api_key, group_id)
    owner_balance = ensure_pool_owner_balance(token, api_key_result["userId"])
    owner_concurrency = ensure_pool_owner_concurrency(token, api_key_result["userId"])
    gateway = validate_gateway(api_key)
    responses_smoke = validate_gateway_responses(api_key)
    compact_evidence = recent_compact_gateway_evidence()
    responses_evidence = recent_responses_gateway_evidence()
    runtime_capabilities = validate_runtime_capabilities(token)
    sentinel = apply_sentinel_manifest(sentinel_payload.get("manifest"))
    sentinel_quality = ensure_sentinel_state_for_sync(account_results)
    sentinel_reassert = reassert_sentinel_freezes_after_sync(token)
    return {
        "ok": gateway["ok"] is True and responses_smoke["ok"] is True and owner_concurrency["ok"] is True and capacity_status["ok"] is True and load_factor_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True and manual_account_proxy_bindings.get("ok") is True and manual_account_group_bindings.get("ok") is True and manual_account_protections.get("ok") is True and sentinel.get("ok") is True and sentinel_quality_prepare.get("ok") is True and sentinel_quality.get("ok") is True and sentinel_reassert.get("ok") is True and runtime_capabilities.get("ok") is True,
        "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")) or bool(responses_evidence.get("degraded")) or runtime_capabilities.get("ok") is not True,
        "mode": "sync",
        "namespace": NAMESPACE,
        "serviceDns": SERVICE_DNS,
        "appPod": APP_POD,
        "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance},
        "pool": {"name": POOL_GROUP_NAME, "id": group_id, "action": group_action, "platform": group.get("platform") if isinstance(group, dict) else "openai"},
        "accounts": {
            "desired": len(profiles),
            "created": sum(1 for item in account_results if item["action"] == "created"),
            "updated": sum(1 for item in account_results if item["action"] == "updated"),
            "pruned": len(pruned_account_results),
            "pruneMode": "explicit" if prune_removed else "disabled-by-default",
            "items": account_results,
            "prunedItems": pruned_account_results,
            "processControl": {"schedulableRestore": "sentinel marker probe only; sync does not restore schedulable for existing accounts", "durableConfig": False},
            "valuesPrinted": False,
        },
        "manualAccounts": {**manual_account_protections, "proxySync": manual_account_proxy_bindings, "groupSync": manual_account_group_bindings},
        "capacity": capacity_status,
        "loadFactor": load_factor_status,
        "webSocketsV2": ws_v2_status,
        "tempUnschedulable": temp_unschedulable_status,
        "apiKey": {
            "name": POOL_API_KEY_NAME,
            "secret": pool_api_key_secret_location(),
            "secretAction": secret_action,
            "secretApply": secret_apply_stdout,
            "sub2apiAction": api_key_result["action"],
            "sub2apiId": api_key_result["id"],
            "groupId": api_key_result["groupId"],
            "userId": api_key_result["userId"],
            "apiKeyFingerprint": secret_fingerprint(api_key),
            "valuesPrinted": False,
        },
        "ownerBalance": owner_balance,
        "ownerConcurrency": owner_concurrency,
        "sentinel": {**sentinel, "qualityGatePrepare": sentinel_quality_prepare, "qualityGate": sentinel_quality, "freezeReassert": sentinel_reassert},
        "runtimeCapabilities": runtime_capabilities,
        "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayResponsesRecent": responses_evidence, "gatewayCompactRecent": compact_evidence},
    }

def run_validate():
    api_key = decode_secret_value(POOL_API_KEY_SECRET_NAME, POOL_API_KEY_SECRET_KEY)
    if not api_key:
        raise RuntimeError(f"{POOL_API_KEY_SECRET_NAME}.{POOL_API_KEY_SECRET_KEY} missing")
    admin_email, token, admin_compliance = login()
    key_item = next((item for item in list_user_keys(token) if item.get("key") == api_key), None)
    owner_balance = None
    owner_concurrency = None
    if key_item is not None and key_item.get("user_id") is not None:
        owner_balance = ensure_pool_owner_balance(token, key_item["user_id"])
        owner_concurrency = ensure_pool_owner_concurrency(token, key_item["user_id"])
    capacity_status = account_capacity_status(token)
    load_factor_status = account_load_factor_status(token)
    ws_v2_status = account_ws_v2_status(token)
    temp_unschedulable_status = account_temp_unschedulable_status(token)
    pool_group_id = key_item.get("group_id") if isinstance(key_item, dict) else None
    manual_account_protections = manual_account_protection_status(token, pool_group_id)
    gateway = validate_gateway(api_key)
    responses_smoke = validate_gateway_responses(api_key)
    compact_evidence = recent_compact_gateway_evidence()
    responses_evidence = recent_responses_gateway_evidence()
    runtime_capabilities = validate_runtime_capabilities(token)
    sentinel = sentinel_runtime_status()
    return {
        "ok": gateway["ok"] is True and responses_smoke["ok"] is True and (owner_concurrency is None or owner_concurrency["ok"] is True) and capacity_status["ok"] is True and load_factor_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True and manual_account_protections.get("ok") is True and sentinel.get("ok") is True and runtime_capabilities.get("ok") is True,
        "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")) or bool(responses_evidence.get("degraded")) or runtime_capabilities.get("ok") is not True,
        "mode": "validate",
        "namespace": NAMESPACE,
        "serviceDns": SERVICE_DNS,
        "appPod": APP_POD,
        "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance},
        "apiKey": {
            "secret": pool_api_key_secret_location(),
            "sub2apiId": key_item.get("id") if isinstance(key_item, dict) else None,
            "userId": key_item.get("user_id") if isinstance(key_item, dict) else None,
            "groupId": key_item.get("group_id") if isinstance(key_item, dict) else None,
            "apiKeyFingerprint": secret_fingerprint(api_key),
            "valuesPrinted": False,
        },
        "ownerBalance": owner_balance,
        "ownerConcurrency": owner_concurrency,
        "capacity": capacity_status,
        "loadFactor": load_factor_status,
        "webSocketsV2": ws_v2_status,
        "tempUnschedulable": temp_unschedulable_status,
        "manualAccounts": manual_account_protections,
        "sentinel": sentinel,
        "runtimeCapabilities": runtime_capabilities,
        "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayResponsesRecent": responses_evidence, "gatewayCompactRecent": compact_evidence},
    }

def parse_log_line(line):
    json_start = line.find("{")
    if json_start < 0:
        return None
    prefix = line[:json_start].rstrip()
    try:
        item = json.loads(line[json_start:])
    except Exception:
        return None
    if not isinstance(item, dict):
        return None
    at = None
    parts = prefix.split()
    if parts:
        at = parts[0]
    message = ""
    if len(parts) >= 4:
        message = " ".join(parts[3:])
    elif len(parts) >= 3:
        message = parts[2]
    elif len(parts) >= 1:
        message = parts[-1]
    item["_at"] = at
    item["_message"] = message
    item["_line"] = line
    return item

def log_time_epoch(item):
    at = item.get("_at") if isinstance(item, dict) else None
    if not isinstance(at, str) or not at:
        return None
    try:
        return datetime.strptime(at, "%Y-%m-%dT%H:%M:%S.%f%z").timestamp()
    except Exception:
        try:
            return datetime.fromisoformat(at.replace("Z", "+00:00")).timestamp()
        except Exception:
            return None

def event_base(item, account_names_by_id):
    account_id = item.get("account_id")
    if isinstance(account_id, str) and account_id.isdigit():
        account_id = int(account_id)
    account_name = account_names_by_id.get(account_id)
    return {
        "at": item.get("_at"),
        "message": item.get("_message"),
        "requestId": item.get("request_id"),
        "clientRequestId": item.get("client_request_id"),
        "path": item.get("path"),
        "method": item.get("method"),
        "model": item.get("model"),
        "accountId": account_id,
        "accountName": account_name,
        "statusCode": item.get("status_code"),
        "upstreamStatus": item.get("upstream_status"),
        "latencyMs": item.get("latency_ms"),
    }

def classify_trace_event(item, account_names_by_id):
    message = str(item.get("_message") or "")
    event = event_base(item, account_names_by_id)
    if "content_moderation.gateway_check_start" in message:
        event.update({
            "type": "request-start",
            "stream": item.get("stream"),
            "bodyBytes": item.get("body_bytes"),
            "groupId": item.get("group_id"),
            "groupName": item.get("group_name"),
            "apiKeyName": item.get("api_key_name"),
        })
    elif "content_moderation.gateway_check_done" in message:
        event.update({
            "type": "gateway-check",
            "allowed": item.get("allowed"),
            "blocked": item.get("blocked"),
            "action": item.get("action"),
        })
    elif "openai.upstream_failover_switching" in message:
        event.update({
            "type": "failover",
            "switchCount": item.get("switch_count"),
            "maxSwitches": item.get("max_switches"),
        })
    elif "openai.account_select_failed" in message:
        event.update({
            "type": "select-failed",
            "error": item.get("error"),
            "excludedAccountCount": item.get("excluded_account_count"),
        })
    elif "account_upstream_error" in message:
        event.update({
            "type": "upstream-error",
            "error": item.get("error"),
        })
    elif "account_temp_unschedulable" in message:
        event.update({
            "type": "temp-unschedulable",
            "until": item.get("until") or item.get("temp_unschedulable_until"),
            "ruleIndex": item.get("rule_index"),
            "matchedKeyword": item.get("matched_keyword"),
            "reason": item.get("reason") or item.get("error"),
        })
    elif "http request completed" in message:
        event.update({
            "type": "final",
            "clientIp": item.get("client_ip"),
            "protocol": item.get("protocol"),
            "platform": item.get("platform"),
            "completedAt": item.get("completed_at"),
        })
    elif "admin account schedulable updated" in message or "account schedulable updated" in message or "/schedulable" in str(item.get("path") or ""):
        event.update({
            "type": "admin-schedulable",
            "schedulable": item.get("schedulable"),
        })
    else:
        event.update({"type": "other"})
    return event

def with_trace_phase(event, first_epoch, last_epoch):
    epoch = None
    at = event.get("at") if isinstance(event, dict) else None
    if isinstance(at, str) and at:
        epoch = log_time_epoch({"_at": at})
    if epoch is None or first_epoch is None:
        phase = "unknown"
    elif epoch < first_epoch:
        phase = "before-request"
    elif last_epoch is not None and epoch > last_epoch:
        phase = "after-request"
    else:
        phase = "during-request"
    event["phase"] = phase
    return event

def account_snapshot_from_runtime(token):
    try:
        accounts = list_accounts(token)
    except Exception as exc:
        return [], {"error": str(exc)}
    rows = []
    for item in accounts:
        if not isinstance(item, dict):
            continue
        rows.append({
            "accountId": item.get("id"),
            "accountName": item.get("name"),
            "schedulable": item.get("schedulable"),
            "status": item.get("status"),
            "concurrency": item.get("concurrency"),
            "priority": item.get("priority"),
            "tempUnschedulableUntil": item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil"),
            "tempUnschedulableSet": (item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil")) is not None or bool(item.get("temp_unschedulable_reason") or item.get("tempUnschedulableReason")),
        })
    rows.sort(key=lambda row: (str(row.get("accountName") or ""), int(row.get("accountId") or 0)))
    return rows, None

def trace_reason(events, final_event):
    failovers = [item for item in events if item.get("type") == "failover"]
    select_failures = [item for item in events if item.get("type") == "select-failed"]
    upstream_errors = [item for item in events if item.get("type") == "upstream-error"]
    if failover_budget_exhausted(failovers, final_event):
        return "failover-budget-exhausted"
    if failovers and select_failures:
        return "failover-attempted-no-candidate"
    if failovers:
        return "failover-attempted"
    if select_failures:
        return "account-select-failed"
    if upstream_errors:
        return "upstream-error"
    if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") >= 400:
        return "final-http-error"
    if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int):
        return "completed"
    return "unknown"

def failover_budget_exhausted(failovers, final_event):
    if not failovers or not isinstance(final_event, dict):
        return False
    last = failovers[-1]
    switch_count = last.get("switchCount")
    max_switches = last.get("maxSwitches")
    final_status = final_event.get("statusCode")
    return (
        isinstance(switch_count, int)
        and isinstance(max_switches, int)
        and max_switches > 0
        and switch_count >= max_switches
        and isinstance(final_status, int)
        and final_status >= 500
    )

def trace_untried_schedulable_accounts(failovers, final_event, account_snapshot):
    if not failover_budget_exhausted(failovers, final_event):
        return []
    tried = set()
    for item in failovers:
        account_id = item.get("accountId")
        if isinstance(account_id, int):
            tried.add(account_id)
    final_account = final_event.get("accountId") if isinstance(final_event, dict) else None
    if isinstance(final_account, int):
        tried.add(final_account)
    result = []
    for item in account_snapshot:
        account_id = item.get("accountId")
        if not isinstance(account_id, int) or account_id in tried:
            continue
        if item.get("schedulable") is True and item.get("status") == "active" and item.get("tempUnschedulableSet") is not True:
            result.append({
                "accountId": account_id,
                "accountName": item.get("accountName"),
                "priority": item.get("priority"),
                "concurrency": item.get("concurrency"),
            })
    return result

def run_trace():
    payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {}
    request_id = payload.get("requestId")
    since = payload.get("since") or "24h"
    tail = int(payload.get("tail") or 20000)
    context_seconds = int(payload.get("contextSeconds") or 300)
    show_lines = bool(payload.get("showLines"))
    if not isinstance(request_id, str) or not request_id:
        raise RuntimeError("trace payload missing requestId")
    admin_email, token, admin_compliance = login()
    account_snapshot, account_snapshot_error = account_snapshot_from_runtime(token)
    account_names_by_id = {}
    for row in account_snapshot:
        account_id = row.get("accountId")
        if isinstance(account_id, str) and account_id.isdigit():
            account_id = int(account_id)
        if isinstance(account_id, int) and isinstance(row.get("accountName"), str):
            account_names_by_id[account_id] = row.get("accountName")
    proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", f"--since={since}", f"--tail={tail}"])
    stdout = proc.stdout.decode("utf-8", errors="replace")
    parsed_lines = []
    matched = []
    for line in stdout.splitlines():
        parsed = parse_log_line(line)
        if parsed is None:
            continue
        parsed_lines.append(parsed)
        if request_id in line:
            matched.append(parsed)
    first_epoch = None
    last_epoch = None
    for item in matched:
        epoch = log_time_epoch(item)
        if epoch is None:
            continue
        first_epoch = epoch if first_epoch is None else min(first_epoch, epoch)
        last_epoch = epoch if last_epoch is None else max(last_epoch, epoch)
    window_lines = []
    if first_epoch is not None:
        start_epoch = first_epoch - context_seconds
        end_epoch = (last_epoch if last_epoch is not None else first_epoch) + context_seconds
        for item in parsed_lines:
            epoch = log_time_epoch(item)
            if epoch is not None and start_epoch <= epoch <= end_epoch:
                window_lines.append(item)
    else:
        window_lines = matched
    events = [classify_trace_event(item, account_names_by_id) for item in matched]
    request_start = next((item for item in events if item.get("type") == "request-start"), None)
    final_event = next((item for item in reversed(events) if item.get("type") == "final"), None)
    failovers = [item for item in events if item.get("type") == "failover"]
    select_failures = [item for item in events if item.get("type") == "select-failed"]
    upstream_errors = [item for item in events if item.get("type") == "upstream-error"]
    temp_unsched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if "account_temp_unschedulable" in str(item.get("_message") or "")]
    admin_sched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if ("schedulable" in str(item.get("_message") or "") or "/schedulable" in str(item.get("path") or ""))]
    window_events = [classify_trace_event(item, account_names_by_id) for item in window_lines]
    final_errors = [item for item in window_events if item.get("type") == "final" and isinstance(item.get("statusCode"), int) and item.get("statusCode") >= 400]
    window_failovers = [item for item in window_events if item.get("type") == "failover"]
    window_select_failures = [item for item in window_events if item.get("type") == "select-failed"]
    untried_schedulable_accounts = trace_untried_schedulable_accounts(failovers, final_event or {}, account_snapshot)
    reason = trace_reason(events, final_event)
    if not matched:
        outcome = "not-found"
    elif isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") < 400:
        outcome = "succeeded"
    elif isinstance(final_event, dict):
        outcome = "failed"
    else:
        outcome = "incomplete"
    return {
        "ok": proc.returncode == 0 and len(matched) > 0,
        "mode": "trace",
        "namespace": NAMESPACE,
        "serviceDns": SERVICE_DNS,
        "appPod": APP_POD,
        "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance},
        "requestId": request_id,
        "summary": {
            "outcome": outcome,
            "reason": reason,
            "eventCount": len(events),
            "matchedLineCount": len(matched),
            "firstAt": events[0].get("at") if events else None,
            "lastAt": events[-1].get("at") if events else None,
        },
        "window": {
            "since": since,
            "tail": tail,
            "beforeSeconds": context_seconds,
            "afterSeconds": context_seconds,
            "lineCount": len(window_lines),
        },
        "request": request_start or {},
        "final": final_event or {},
        "events": events,
        "failovers": failovers,
        "selectFailures": select_failures,
        "upstreamErrors": upstream_errors,
        "tempUnschedulable": temp_unsched,
        "adminSchedulable": admin_sched[-20:],
        "windowStats": {
            "matchedLines": len(matched),
            "eventCount": len(window_events),
            "finalErrorCount": len(final_errors),
            "failoverCount": len(window_failovers),
            "selectFailedCount": len(window_select_failures),
            "tempUnschedulableCount": len(temp_unsched),
            "adminSchedulableCount": len(admin_sched),
        },
        "diagnostics": {
            "failoverBudgetExhausted": failover_budget_exhausted(failovers, final_event or {}),
            "untriedSchedulableAccounts": untried_schedulable_accounts,
        },
        "accountSnapshot": account_snapshot,
        "accountSnapshotError": account_snapshot_error,
        "rawLines": [{"line": item.get("_line")} for item in matched[-30:]] if show_lines else [],
        "showLines": show_lines,
        "logs": {
            "exitCode": proc.returncode,
            "stderrTail": text(proc.stderr, 1000),
            "stdoutLineCount": len(stdout.splitlines()),
        },
        "valuesPrinted": False,
    }

def parse_embedded_json(stdout):
    if not isinstance(stdout, str) or not stdout.strip():
        return None
    start = stdout.find("{")
    end = stdout.rfind("}")
    if start < 0 or end <= start:
        return None
    try:
        return json.loads(stdout[start:end + 1])
    except Exception:
        return None

def inject_env(template, name, value):
    spec = template.setdefault("spec", {})
    containers = spec.setdefault("containers", [])
    if not containers:
        raise RuntimeError("sentinel job template has no containers")
    container = containers[0]
    env = container.setdefault("env", [])
    env[:] = [item for item in env if not (isinstance(item, dict) and item.get("name") == name)]
    env.append({"name": name, "value": value})

def sentinel_probe_job_manifest(accounts):
    cronjob_name = SENTINEL_CONFIG.get("cronJobName")
    if not isinstance(cronjob_name, str) or not cronjob_name:
        raise RuntimeError("sentinel cronJobName missing from config")
    cronjob = kube_json(["-n", NAMESPACE, "get", "cronjob", cronjob_name], f"cronjob/{cronjob_name}")
    job_template = ((cronjob.get("spec") or {}).get("jobTemplate") or {}) if isinstance(cronjob, dict) else {}
    job_spec = copy_json(job_template.get("spec") or {})
    if not isinstance(job_spec, dict) or not job_spec:
        raise RuntimeError("sentinel CronJob jobTemplate.spec missing")
    template = job_spec.get("template")
    if not isinstance(template, dict):
        raise RuntimeError("sentinel CronJob jobTemplate.spec.template missing")
    inject_env(template, "SENTINEL_ACCOUNT_NAMES", ",".join(accounts))
    job_spec["ttlSecondsAfterFinished"] = int(job_spec.get("ttlSecondsAfterFinished") or 3600)
    suffix = str(int(time.time()))[-8:] + "-" + "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(5))
    job_name = ("sub2api-sentinel-probe-" + suffix)[:63]
    return job_name, {
        "apiVersion": "batch/v1",
        "kind": "Job",
        "metadata": {
            "name": job_name,
            "namespace": NAMESPACE,
            "labels": {
                "app.kubernetes.io/name": cronjob_name,
                "app.kubernetes.io/part-of": "platform-infra",
                "app.kubernetes.io/managed-by": "unidesk",
                "unidesk.ai/job-purpose": "sub2api-account-sentinel-manual-probe",
            },
        },
        "spec": job_spec,
    }

def copy_json(value):
    return json.loads(json.dumps(value))

def job_condition(job, cond_type):
    for item in ((job.get("status") or {}).get("conditions") or []):
        if item.get("type") == cond_type and item.get("status") == "True":
            return item
    return None

def wait_sentinel_probe_job(job_name, timeout_seconds):
    deadline = time.time() + timeout_seconds
    latest = None
    while time.time() < deadline:
        latest, err = safe_kube_json(["-n", NAMESPACE, "get", "job", job_name], f"job/{job_name}")
        if isinstance(latest, dict):
            complete = job_condition(latest, "Complete")
            failed = job_condition(latest, "Failed")
            if complete is not None:
                return "succeeded", latest, complete
            if failed is not None:
                return "failed", latest, failed
        time.sleep(2)
    return "timeout", latest, None

def job_logs(job_name):
    proc = kubectl(["-n", NAMESPACE, "logs", f"job/{job_name}", "--tail=4000"])
    return {
        "exitCode": proc.returncode,
        "stdout": proc.stdout.decode("utf-8", errors="replace"),
        "stderr": text(proc.stderr, 4000),
    }

def run_sentinel_probe():
    payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {}
    accounts = payload.get("accounts") if isinstance(payload, dict) else None
    if not isinstance(accounts, list) or not accounts or not all(isinstance(item, str) and item for item in accounts):
        raise RuntimeError("sentinel-probe payload requires non-empty accounts")
    job_name, manifest = sentinel_probe_job_manifest(accounts)
    proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], json.dumps(manifest))
    if proc.returncode != 0:
        raise RuntimeError(f"apply sentinel probe job failed: {text(proc.stderr, 2000)}")
    timeout_seconds = max(300, int(SENTINEL_CONFIG.get("probe", {}).get("timeoutSeconds") or 30) + 240)
    status, job, condition = wait_sentinel_probe_job(job_name, timeout_seconds)
    logs = job_logs(job_name)
    parsed = parse_embedded_json(logs.get("stdout") or "")
    results = parsed.get("results") if isinstance(parsed, dict) and isinstance(parsed.get("results"), list) else []
    state_summary = sentinel_state_summary()
    last_run = state_summary.get("lastRun") if isinstance(state_summary, dict) and isinstance(state_summary.get("lastRun"), dict) else {}
    if not results and isinstance(last_run.get("results"), list):
        results = last_run.get("results")
    requested = set(accounts)
    measured = {item.get("accountName") for item in results if isinstance(item, dict)}
    missing = sorted(name for name in requested if name not in measured)
    marker_ok = len(missing) == 0 and all(isinstance(item, dict) and item.get("accountName") in requested and item.get("markerMatched") is True for item in results if isinstance(item, dict) and item.get("accountName") in requested)
    if not results and isinstance(last_run, dict):
        selected = int(last_run.get("selected") or 0)
        ok_count = int(last_run.get("okCount") or 0)
        marker_mismatch_count = int(last_run.get("markerMismatchCount") or 0)
        transport_failure_count = int(last_run.get("transportFailureCount") or 0)
        if selected >= len(requested) and ok_count >= len(requested) and marker_mismatch_count == 0 and transport_failure_count == 0:
            missing = []
            marker_ok = True
    job_ok = status == "succeeded" and logs.get("exitCode") == 0
    return {
        "ok": job_ok and marker_ok,
        "jobExecutionOk": job_ok,
        "markerOk": marker_ok,
        "mode": "sentinel-probe",
        "namespace": NAMESPACE,
        "requestedAccounts": accounts,
        "missingAccounts": missing,
        "job": {
            "name": job_name,
            "status": status,
            "condition": condition,
            "timeoutSeconds": timeout_seconds,
            "applyStdoutTail": text(proc.stdout, 1200),
            "logsExitCode": logs.get("exitCode"),
            "logsStderrTail": logs.get("stderr"),
        },
        "probe": parsed,
        "summary": {
            "at": last_run.get("at"),
            "selected": last_run.get("selected"),
            "okCount": last_run.get("okCount"),
            "markerMismatchCount": last_run.get("markerMismatchCount"),
            "transportFailureCount": last_run.get("transportFailureCount"),
            "actionsTaken": last_run.get("actionsTaken"),
            "selection": last_run.get("selection"),
        },
        "results": results,
        "sentinelState": state_summary,
        "valuesPrinted": False,
    }

def run_cleanup_probes():
    admin_email, token, admin_compliance = login()
    cleanup = cleanup_probe_resources(token)
    return {
        "ok": cleanup.get("ok") is True,
        "mode": "cleanup-probes",
        "namespace": NAMESPACE,
        "serviceDns": SERVICE_DNS,
        "appPod": APP_POD,
        "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance},
        "cleanup": cleanup,
        "valuesPrinted": False,
    }

try:
    if MODE == "sync":
        result = run_sync()
    elif MODE == "trace":
        result = run_trace()
    elif MODE == "cleanup-probes":
        result = run_cleanup_probes()
    elif MODE == "sentinel-probe":
        result = run_sentinel_probe()
    else:
        result = run_validate()
except Exception as exc:
    result = {
        "ok": False,
        "mode": MODE,
        "namespace": NAMESPACE,
        "serviceDns": SERVICE_DNS,
        "appPod": globals().get("APP_POD"),
        "error": str(exc),
        "valuesPrinted": False,
    }

print(json.dumps(result, ensure_ascii=False, indent=2))
sys.exit(0 if result.get("ok") else 1)
PY
`;
}