diff --git a/.agents/skills/unidesk-sub2api/SKILL.md b/.agents/skills/unidesk-sub2api/SKILL.md index 8eb98eb7..1b758545 100644 --- a/.agents/skills/unidesk-sub2api/SKILL.md +++ b/.agents/skills/unidesk-sub2api/SKILL.md @@ -69,9 +69,17 @@ bun scripts/cli.ts platform-infra sub2api codex-pool cleanup-probes --confirm - `profiles.entries[].tempUnschedulable`: 可选 per-account 临时下线规则覆盖；字段语义以 `docs/reference/platform-infra.md` 为权威。上游 Sub2API 不支持的成功体分类、调度策略或账号冷却行为不要在这里声明。 - `profiles.entries[].openaiResponsesWebSocketsV2Mode`: 需要 Responses WebSocket v2 的上游才设置，值为 `off`、`ctx_pool` 或 `passthrough`。 - `profiles.entries[].upstreamUserAgent`: 少数要求 Codex CLI User-Agent 的上游才设置，不能含换行。 +- `sentinel.monitor.enabled`: 账号级 HTTP 200 成功体哨兵监控开关；开启后 `codex-pool sync --confirm` 会在 `platform-infra` 创建/更新 k8s CronJob、ConfigMap、Secret、ServiceAccount、Role 和 RoleBinding。CronJob 直打 YAML-managed 上游账号的 OpenAI Responses `gpt-5.5`，用确定 marker 判定是否出现维护/故障/广告等非预期成功体，并在独立 state ConfigMap 中记录 token/cost 账本。 +- `sentinel.actions.enabled`: 账号级哨兵冻结/恢复动作开关；默认必须保持 `false`，先监控一段时间，确认 marker 判定正确率、token 消耗和误报率后再改为 `true`。动作关闭时只记录 `would-freeze`，不会调用 Sub2API admin API 改 `schedulable`。 +- `sentinel.probe.maxOutputTokens`: 哨兵 OpenAI Responses 请求的硬输出上限，必须保持小值；不要只靠 prompt 要求模型少输出。哨兵不限制并发和每轮账号数，所有到期账号会在同一轮并发探测。 +- `sentinel.cadence`: 成功信任指数退避配置。当前口径是从 1 分钟开始，连续成功后退避到最大 20 分钟；任意 marker mismatch 清零成功信任并进入冻结退避。 +- `sentinel.freeze`: 冻结 TTL 指数退避配置。当前口径是初始 10 分钟，失败后 `10m -> 20m -> 40m -> 80m -> 120m`，最大 2 小时；冻结到期后只做恢复 probe，通过才自动恢复，不能仅靠 TTL 到期解封。 +- `sentinel.pricing`: 直打上游时哨兵自己的 token/cost 估算价格。因为 direct upstream probe 不经过 Sub2API 普通用量账本，哨兵必须自己记录全局与 per-account token/cost；这些账本只用于观察，不作为跳过探测的预算门禁。 `sync --confirm` 会登录 Sub2API admin、创建/更新 group、创建/更新 YAML 中的 `unidesk-codex-*` accounts、创建/复用统一 API key Secret，并把 managed account 的 `schedulable=true` 恢复为过程控制基线；它默认不删除 YAML 中缺席的 managed account。只有明确退役上游时才使用 `sync --confirm --prune-removed` 删除缺席且 `extra.unidesk_managed=true` 的 `unidesk-codex-*` account。 +`sync --confirm` 同时会按 YAML 渲染账号级哨兵资源。哨兵默认使用 `sentinel.monitor.enabled=true` + `sentinel.actions.enabled=false` 的观察模式；如果后续要开启自动冻结/恢复，只改 `sentinel.actions.enabled=true` 后重新 `codex-pool sync --confirm`，不要手工 patch CronJob、Secret 或 Sub2API account。打开动作前必须先用 `codex-pool validate --full` 或 k8s state 证据确认近期 `lastRun` 的 marker mismatch 与实际上游行为一致，且 token/cost 账本可解释、成本可接受。 + `sync --confirm` 和 `validate` 可能超过单次 SSH/runtime 短连接窗口。必须继续使用 `bun scripts/cli.ts platform-infra sub2api codex-pool ...`，由 CLI 在 G14 远端提交作业并短轮询状态；不要改用裸 `trans G14:k3s script` 等一个长连接等待完整结果。若看到 `UNIDESK_SSH_RUNTIME_TIMEOUT`，先按 `docs/reference/platform-infra.md` 的规则处理为控制面可见性问题，修 CLI/job/poll 或重跑受控命令，不要手工 patch Sub2API credentials 或源码。 不要给 UniDesk-managed Codex accounts 开 Sub2API `pool_mode`。UniDesk 期望的 failover 是把失败账号临时标记为 unschedulable，让同组其他账号接手；`pool_mode` 会重试同一个 account path。 diff --git a/config/platform-infra/sub2api-codex-pool.yaml b/config/platform-infra/sub2api-codex-pool.yaml index 8c2469e7..f8452e24 100644 --- a/config/platform-infra/sub2api-codex-pool.yaml +++ b/config/platform-infra/sub2api-codex-pool.yaml @@ -144,3 +144,41 @@ localCodex: supportsWebSockets: false responsesWebSocketsV2: false responsesSmokeModel: gpt-5.5 +sentinel: + monitor: + enabled: true + actions: + enabled: false + freezeOnMarkerMismatch: true + freezeOnTransportError: false + schedule: "*/1 * * * *" + image: python:3.12-alpine + serviceAccountName: sub2api-account-sentinel + configMapName: sub2api-account-sentinel-config + credentialsSecretName: sub2api-account-sentinel-profiles + stateConfigMapName: sub2api-account-sentinel-state + cronJobName: sub2api-account-sentinel + model: gpt-5.5 + endpoint: responses + marker: + prefix: UDSG_OK + exact: true + probe: + timeoutSeconds: 30 + maxResponseBytes: 8192 + maxOutputTokens: 12 + transportRetryMinutes: 5 + cadence: + successInitialIntervalMinutes: 1 + successMaxIntervalMinutes: 20 + successBackoffMultiplier: 2 + jitterPercent: 10 + freeze: + initialTtlMinutes: 10 + maxTtlMinutes: 120 + backoffMultiplier: 2 + jitterPercent: 10 + pricing: + usdPer1MInputTokens: 1.25 + usdPer1MOutputTokens: 10 + historyLimit: 200 diff --git a/scripts/platform-infra-sub2api-codex-sentinel-contract-test.ts b/scripts/platform-infra-sub2api-codex-sentinel-contract-test.ts new file mode 100644 index 00000000..6a2da782 --- /dev/null +++ b/scripts/platform-infra-sub2api-codex-sentinel-contract-test.ts @@ -0,0 +1,107 @@ +import { readFileSync } from "node:fs"; +import { rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { rootPath } from "./src/config"; +import { + defaultCodexPoolSentinelConfig, + readCodexPoolSentinelConfig, + renderCodexPoolSentinelManifest, + sentinelRunnerPython, +} from "./src/platform-infra-sub2api-codex-sentinel"; + +function assertCondition(condition: unknown, message: string, detail: unknown = {}): void { + if (!condition) throw new Error(`${message}: ${JSON.stringify(detail)}`); +} + +const configPath = rootPath("config", "platform-infra", "sub2api-codex-pool.yaml"); +const parsed = Bun.YAML.parse(readFileSync(configPath, "utf8")) as { sentinel?: unknown }; +const sentinel = readCodexPoolSentinelConfig(parsed.sentinel, defaultCodexPoolSentinelConfig(), configPath); + +assertCondition(sentinel.monitor.enabled === true, "sentinel monitor must be enabled for observation-first rollout", sentinel); +assertCondition(sentinel.actions.enabled === false, "sentinel actions must default off until monitoring quality is reviewed", sentinel); +assertCondition(sentinel.actions.freezeOnMarkerMismatch === true, "marker mismatch must be configured as freeze-worthy when actions are enabled", sentinel.actions); +assertCondition(sentinel.actions.freezeOnTransportError === false, "transport errors must not freeze by default", sentinel.actions); +assertCondition(sentinel.endpoint === "responses", "v1 sentinel must target OpenAI Responses only", sentinel); +assertCondition(sentinel.model === "gpt-5.5", "v1 sentinel must use GPT-5.5", sentinel); +assertCondition(sentinel.probe.maxOutputTokens > 0 && sentinel.probe.maxOutputTokens <= 16, "sentinel maxOutputTokens must be tightly capped", sentinel.probe); +assertCondition(!("concurrency" in sentinel.probe), "sentinel must not cap probe concurrency; all due accounts are probed concurrently", sentinel.probe); +assertCondition(!("maxAccountsPerRun" in sentinel.probe), "sentinel must not cap accounts per run; all due accounts are eligible", sentinel.probe); +assertCondition(sentinel.cadence.successInitialIntervalMinutes === 1, "success trust backoff must start at 1 minute", sentinel.cadence); +assertCondition(sentinel.cadence.successMaxIntervalMinutes === 20, "success trust backoff must cap at 20 minutes", sentinel.cadence); +assertCondition(sentinel.freeze.initialTtlMinutes === 10, "freeze backoff must start at 10 minutes", sentinel.freeze); +assertCondition(sentinel.freeze.maxTtlMinutes === 120, "freeze backoff must cap at 2 hours", sentinel.freeze); +assertCondition(!("budget" in sentinel), "sentinel must not use token budgets as a probe gate; usage is recorded only", sentinel); + +const manifest = renderCodexPoolSentinelManifest(sentinel, [ + { + accountName: "unidesk-codex-example", + profile: "example", + baseUrl: "https://example.invalid/v1", + apiKey: "sk-test-secret", + upstreamUserAgent: null, + }, +], { + namespace: "platform-infra", + serviceName: "sub2api", + serviceDns: "sub2api.platform-infra.svc.cluster.local:8080", + appSecretName: "sub2api-secrets", +}); + +assertCondition(manifest.includes("kind: CronJob"), "sentinel manifest must render a CronJob", manifest.slice(0, 1000)); +assertCondition(manifest.includes("concurrencyPolicy: Forbid"), "sentinel CronJob must forbid overlapping runs", manifest); +assertCondition(manifest.includes("suspend: false"), "monitor.enabled=true must unsuspend the CronJob", manifest); +assertCondition(manifest.includes("kind: ServiceAccount") && manifest.includes("kind: Role") && manifest.includes("kind: RoleBinding"), "sentinel manifest must include minimal RBAC", manifest); +assertCondition(manifest.includes("sub2api-account-sentinel-state"), "sentinel manifest must reference the state ConfigMap", manifest); +assertCondition(manifest.includes("\"enabled\": false"), "sentinel manifest must preserve actions.enabled=false in config.json", manifest); +assertCondition(!manifest.includes("sk-test-secret"), "sentinel manifest must not expose upstream credentials as plaintext", manifest); +assertCondition(manifest.includes("profiles.json:"), "sentinel credentials Secret must include the profiles payload as Secret data", manifest); +assertCondition(manifest.includes("\"budgetMode\": \"record-only\""), "sentinel runner must expose record-only budget/accounting mode", manifest); +assertCondition(manifest.includes("max_workers=max(1, len(due))"), "sentinel runner must probe all due accounts concurrently", manifest); + +const disabledMonitor = { + ...sentinel, + monitor: { enabled: false }, + actions: { ...sentinel.actions, enabled: false }, +}; +const suspendedManifest = renderCodexPoolSentinelManifest(disabledMonitor, [], { + namespace: "platform-infra", + serviceName: "sub2api", + serviceDns: "sub2api.platform-infra.svc.cluster.local:8080", + appSecretName: "sub2api-secrets", +}); +assertCondition(suspendedManifest.includes("suspend: true"), "monitor.enabled=false must suspend the CronJob", suspendedManifest); + +const pythonPath = join(tmpdir(), `sub2api-account-sentinel-${process.pid}.py`); +writeFileSync(pythonPath, sentinelRunnerPython(), "utf8"); +try { + const pyCompile = Bun.spawnSync(["python3", "-m", "py_compile", pythonPath], { + stdout: "pipe", + stderr: "pipe", + }); + assertCondition(pyCompile.exitCode === 0, "sentinel runner python must compile", { + exitCode: pyCompile.exitCode, + stdout: pyCompile.stdout.toString(), + stderr: pyCompile.stderr.toString(), + }); +} finally { + rmSync(pythonPath, { force: true }); +} + +console.log(JSON.stringify({ + ok: true, + checks: [ + "sentinel has independent monitor/actions YAML switches", + "observation-first rollout keeps actions disabled", + "v1 scope is OpenAI Responses + GPT-5.5", + "probe max_output_tokens is tightly capped", + "probe concurrency is not artificially capped", + "budget is record-only and does not gate probes", + "success trust backoff is 1m to 20m", + "freeze backoff is 10m to 120m", + "CronJob is k8s-native with Forbid concurrency and minimal RBAC", + "monitor switch controls CronJob suspend state", + "rendered Secret avoids plaintext upstream credentials", + "embedded Python runner compiles", + ], +})); diff --git a/scripts/src/check.ts b/scripts/src/check.ts index 72adf862..8b4eb6d2 100644 --- a/scripts/src/check.ts +++ b/scripts/src/check.ts @@ -537,6 +537,7 @@ export async function runChecks(config: UniDeskConfig, options: CheckOptions = d items.push(await commandItem("playwright:cli-wrapper-contract", ["bun", "scripts/playwright-cli-contract-test.ts"], 30_000, process.env, options.checkHeartbeatMs)); items.push(await commandItem("platform-infra:sub2api-codex-local-config-contract", ["bun", "scripts/platform-infra-sub2api-codex-local-config-contract-test.ts"], 30_000, process.env, options.checkHeartbeatMs)); items.push(await commandItem("platform-infra:sub2api-codex-routing-contract", ["bun", "scripts/platform-infra-sub2api-codex-routing-contract-test.ts"], 30_000, process.env, options.checkHeartbeatMs)); + items.push(await commandItem("platform-infra:sub2api-codex-sentinel-contract", ["bun", "scripts/platform-infra-sub2api-codex-sentinel-contract-test.ts"], 30_000, process.env, options.checkHeartbeatMs)); items.push(await commandItem("platform-infra:sub2api-codex-temp-unsched-contract", ["bun", "scripts/platform-infra-sub2api-codex-temp-unsched-contract-test.ts"], 30_000, process.env, options.checkHeartbeatMs)); items.push(await commandItem("platform-infra:sub2api-http-upstream-contract", ["bun", "scripts/platform-infra-sub2api-http-upstream-contract-test.ts"], 30_000, process.env, options.checkHeartbeatMs)); items.push(await commandItem("auth-broker:p0-contract", ["bun", "scripts/auth-broker-contract-test.ts"], 30_000, process.env, options.checkHeartbeatMs)); @@ -579,6 +580,7 @@ export async function runChecks(config: UniDeskConfig, options: CheckOptions = d items.push(skippedItem("server:cleanup-plan-contract", "Server cleanup dry-run contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("check:gh-contract-scope-contract", "Check option scope contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("playwright:cli-wrapper-contract", "Playwright wrapper/headless/session contract is opt-in with script checks", "--scripts-typecheck or --full")); + items.push(skippedItem("platform-infra:sub2api-codex-sentinel-contract", "Sub2API Codex account sentinel contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("auth-broker:p0-contract", "Auth Broker P0 skeleton and CLI adapter contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("d601:recovery-guardrails-contract", "D601 recovery guardrails fixture contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("hwlab:cd-wrapper-contract", "HWLAB DEV CD wrapper contract is opt-in with script checks", "--scripts-typecheck or --full")); diff --git a/scripts/src/platform-infra-sub2api-codex-sentinel.ts b/scripts/src/platform-infra-sub2api-codex-sentinel.ts new file mode 100644 index 00000000..3845fdae --- /dev/null +++ b/scripts/src/platform-infra-sub2api-codex-sentinel.ts @@ -0,0 +1,1049 @@ +import { Buffer } from "node:buffer"; + +export interface CodexPoolSentinelConfig { + monitor: { + enabled: boolean; + }; + actions: { + enabled: boolean; + freezeOnMarkerMismatch: boolean; + freezeOnTransportError: boolean; + }; + schedule: string; + image: string; + serviceAccountName: string; + configMapName: string; + credentialsSecretName: string; + stateConfigMapName: string; + cronJobName: string; + model: string; + endpoint: "responses"; + marker: { + prefix: string; + exact: boolean; + }; + probe: { + timeoutSeconds: number; + maxResponseBytes: number; + maxOutputTokens: number; + transportRetryMinutes: number; + }; + cadence: { + successInitialIntervalMinutes: number; + successMaxIntervalMinutes: number; + successBackoffMultiplier: number; + jitterPercent: number; + }; + freeze: { + initialTtlMinutes: number; + maxTtlMinutes: number; + backoffMultiplier: number; + jitterPercent: number; + }; + pricing: { + usdPer1MInputTokens: number; + usdPer1MOutputTokens: number; + }; + historyLimit: number; +} + +export interface CodexPoolSentinelProfileSecret { + accountName: string; + profile: string; + baseUrl: string; + apiKey: string; + upstreamUserAgent: string | null; +} + +export interface CodexPoolSentinelManifestOptions { + namespace: string; + serviceName: string; + serviceDns: string; + appSecretName: string; +} + +export function defaultCodexPoolSentinelConfig(): CodexPoolSentinelConfig { + return { + monitor: { + enabled: true, + }, + actions: { + enabled: false, + freezeOnMarkerMismatch: true, + freezeOnTransportError: false, + }, + schedule: "*/1 * * * *", + image: "python:3.12-alpine", + serviceAccountName: "sub2api-account-sentinel", + configMapName: "sub2api-account-sentinel-config", + credentialsSecretName: "sub2api-account-sentinel-profiles", + stateConfigMapName: "sub2api-account-sentinel-state", + cronJobName: "sub2api-account-sentinel", + model: "gpt-5.5", + endpoint: "responses", + marker: { + prefix: "UDSG_OK", + exact: true, + }, + probe: { + timeoutSeconds: 30, + maxResponseBytes: 8192, + maxOutputTokens: 12, + transportRetryMinutes: 5, + }, + cadence: { + successInitialIntervalMinutes: 1, + successMaxIntervalMinutes: 20, + successBackoffMultiplier: 2, + jitterPercent: 10, + }, + freeze: { + initialTtlMinutes: 10, + maxTtlMinutes: 120, + backoffMultiplier: 2, + jitterPercent: 10, + }, + pricing: { + usdPer1MInputTokens: 1.25, + usdPer1MOutputTokens: 10, + }, + historyLimit: 200, + }; +} + +export function readCodexPoolSentinelConfig(value: unknown, defaults: CodexPoolSentinelConfig, sourcePath: string): CodexPoolSentinelConfig { + if (!isRecord(value)) return defaults; + const monitor = isRecord(value.monitor) ? value.monitor : {}; + const actions = isRecord(value.actions) ? value.actions : {}; + const marker = isRecord(value.marker) ? value.marker : {}; + const probe = isRecord(value.probe) ? value.probe : {}; + const cadence = isRecord(value.cadence) ? value.cadence : {}; + const freeze = isRecord(value.freeze) ? value.freeze : {}; + const pricing = isRecord(value.pricing) ? value.pricing : {}; + const config: CodexPoolSentinelConfig = { + monitor: { + enabled: readBoolean(valueAt(monitor, "enabled"), `${sourcePath}.sentinel.monitor.enabled`, defaults.monitor.enabled), + }, + actions: { + enabled: readBoolean(valueAt(actions, "enabled"), `${sourcePath}.sentinel.actions.enabled`, defaults.actions.enabled), + freezeOnMarkerMismatch: readBoolean(valueAt(actions, "freezeOnMarkerMismatch"), `${sourcePath}.sentinel.actions.freezeOnMarkerMismatch`, defaults.actions.freezeOnMarkerMismatch), + freezeOnTransportError: readBoolean(valueAt(actions, "freezeOnTransportError"), `${sourcePath}.sentinel.actions.freezeOnTransportError`, defaults.actions.freezeOnTransportError), + }, + schedule: readString(valueAt(value, "schedule"), `${sourcePath}.sentinel.schedule`, defaults.schedule), + image: readString(valueAt(value, "image"), `${sourcePath}.sentinel.image`, defaults.image), + serviceAccountName: readDnsName(valueAt(value, "serviceAccountName"), `${sourcePath}.sentinel.serviceAccountName`, defaults.serviceAccountName), + configMapName: readDnsName(valueAt(value, "configMapName"), `${sourcePath}.sentinel.configMapName`, defaults.configMapName), + credentialsSecretName: readDnsName(valueAt(value, "credentialsSecretName"), `${sourcePath}.sentinel.credentialsSecretName`, defaults.credentialsSecretName), + stateConfigMapName: readDnsName(valueAt(value, "stateConfigMapName"), `${sourcePath}.sentinel.stateConfigMapName`, defaults.stateConfigMapName), + cronJobName: readDnsName(valueAt(value, "cronJobName"), `${sourcePath}.sentinel.cronJobName`, defaults.cronJobName), + model: readModelName(valueAt(value, "model"), `${sourcePath}.sentinel.model`, defaults.model), + endpoint: "responses", + marker: { + prefix: readMarkerPrefix(valueAt(marker, "prefix"), `${sourcePath}.sentinel.marker.prefix`, defaults.marker.prefix), + exact: readBoolean(valueAt(marker, "exact"), `${sourcePath}.sentinel.marker.exact`, defaults.marker.exact), + }, + probe: { + timeoutSeconds: readInt(valueAt(probe, "timeoutSeconds"), `${sourcePath}.sentinel.probe.timeoutSeconds`, defaults.probe.timeoutSeconds, 3, 300), + maxResponseBytes: readInt(valueAt(probe, "maxResponseBytes"), `${sourcePath}.sentinel.probe.maxResponseBytes`, defaults.probe.maxResponseBytes, 1024, 1048576), + maxOutputTokens: readInt(valueAt(probe, "maxOutputTokens"), `${sourcePath}.sentinel.probe.maxOutputTokens`, defaults.probe.maxOutputTokens, 1, 128), + transportRetryMinutes: readInt(valueAt(probe, "transportRetryMinutes"), `${sourcePath}.sentinel.probe.transportRetryMinutes`, defaults.probe.transportRetryMinutes, 1, 120), + }, + cadence: { + successInitialIntervalMinutes: readInt(valueAt(cadence, "successInitialIntervalMinutes"), `${sourcePath}.sentinel.cadence.successInitialIntervalMinutes`, defaults.cadence.successInitialIntervalMinutes, 1, 1440), + successMaxIntervalMinutes: readInt(valueAt(cadence, "successMaxIntervalMinutes"), `${sourcePath}.sentinel.cadence.successMaxIntervalMinutes`, defaults.cadence.successMaxIntervalMinutes, 1, 1440), + successBackoffMultiplier: readInt(valueAt(cadence, "successBackoffMultiplier"), `${sourcePath}.sentinel.cadence.successBackoffMultiplier`, defaults.cadence.successBackoffMultiplier, 1, 10), + jitterPercent: readInt(valueAt(cadence, "jitterPercent"), `${sourcePath}.sentinel.cadence.jitterPercent`, defaults.cadence.jitterPercent, 0, 50), + }, + freeze: { + initialTtlMinutes: readInt(valueAt(freeze, "initialTtlMinutes"), `${sourcePath}.sentinel.freeze.initialTtlMinutes`, defaults.freeze.initialTtlMinutes, 1, 1440), + maxTtlMinutes: readInt(valueAt(freeze, "maxTtlMinutes"), `${sourcePath}.sentinel.freeze.maxTtlMinutes`, defaults.freeze.maxTtlMinutes, 1, 1440), + backoffMultiplier: readInt(valueAt(freeze, "backoffMultiplier"), `${sourcePath}.sentinel.freeze.backoffMultiplier`, defaults.freeze.backoffMultiplier, 1, 10), + jitterPercent: readInt(valueAt(freeze, "jitterPercent"), `${sourcePath}.sentinel.freeze.jitterPercent`, defaults.freeze.jitterPercent, 0, 50), + }, + pricing: { + usdPer1MInputTokens: readNumber(valueAt(pricing, "usdPer1MInputTokens"), `${sourcePath}.sentinel.pricing.usdPer1MInputTokens`, defaults.pricing.usdPer1MInputTokens, 0, 100000), + usdPer1MOutputTokens: readNumber(valueAt(pricing, "usdPer1MOutputTokens"), `${sourcePath}.sentinel.pricing.usdPer1MOutputTokens`, defaults.pricing.usdPer1MOutputTokens, 0, 100000), + }, + historyLimit: readInt(valueAt(value, "historyLimit"), `${sourcePath}.sentinel.historyLimit`, defaults.historyLimit, 1, 2000), + }; + if (config.actions.enabled && !config.monitor.enabled) { + throw new Error(`${sourcePath}.sentinel.actions.enabled requires sentinel.monitor.enabled=true`); + } + if (config.cadence.successMaxIntervalMinutes < config.cadence.successInitialIntervalMinutes) { + throw new Error(`${sourcePath}.sentinel.cadence.successMaxIntervalMinutes must be >= successInitialIntervalMinutes`); + } + if (config.freeze.maxTtlMinutes < config.freeze.initialTtlMinutes) { + throw new Error(`${sourcePath}.sentinel.freeze.maxTtlMinutes must be >= initialTtlMinutes`); + } + if (!/^[-0-9A-Za-z_/*,\s]+$/u.test(config.schedule)) { + throw new Error(`${sourcePath}.sentinel.schedule has an unsupported cron format`); + } + if (!/^[A-Za-z0-9._:/@-]+$/u.test(config.image)) { + throw new Error(`${sourcePath}.sentinel.image has an unsupported image format`); + } + return config; +} + +export function codexPoolSentinelSummary(config: CodexPoolSentinelConfig): Record { + return { + monitorEnabled: config.monitor.enabled, + actionsEnabled: config.actions.enabled, + schedule: config.schedule, + cronJobName: config.cronJobName, + configMapName: config.configMapName, + credentialsSecretName: config.credentialsSecretName, + stateConfigMapName: config.stateConfigMapName, + image: config.image, + model: config.model, + endpoint: config.endpoint, + probe: config.probe, + cadence: config.cadence, + freeze: config.freeze, + accounting: { + mode: "record-only", + pricing: config.pricing, + }, + marker: { + prefix: config.marker.prefix, + exact: config.marker.exact, + }, + valuesPrinted: false, + }; +} + +export function renderCodexPoolSentinelManifest( + config: CodexPoolSentinelConfig, + profiles: CodexPoolSentinelProfileSecret[], + options: CodexPoolSentinelManifestOptions, +): string { + const profilesJson = JSON.stringify({ profiles }, null, 2); + const runnerConfig = { + monitor: config.monitor, + actions: config.actions, + service: { + baseUrl: `http://${options.serviceDns}`, + adminEmailDefault: "admin@sub2api.platform-infra.local", + }, + model: config.model, + endpoint: config.endpoint, + marker: config.marker, + probe: config.probe, + cadence: config.cadence, + freeze: config.freeze, + pricing: config.pricing, + state: { + configMapName: config.stateConfigMapName, + historyLimit: config.historyLimit, + }, + }; + const suspend = config.monitor.enabled ? "false" : "true"; + const activeDeadlineSeconds = Math.max(120, Math.min(3600, config.probe.timeoutSeconds + 120)); + return `apiVersion: v1 +kind: Secret +metadata: + name: ${config.credentialsSecretName} + namespace: ${options.namespace} + labels: + app.kubernetes.io/name: ${config.cronJobName} + app.kubernetes.io/part-of: platform-infra + app.kubernetes.io/managed-by: unidesk + unidesk.ai/secret-purpose: sub2api-account-sentinel-profiles +type: Opaque +data: + profiles.json: ${Buffer.from(profilesJson, "utf8").toString("base64")} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: ${config.configMapName} + namespace: ${options.namespace} + labels: + app.kubernetes.io/name: ${config.cronJobName} + app.kubernetes.io/part-of: platform-infra + app.kubernetes.io/managed-by: unidesk +data: + config.json: | +${indentBlock(JSON.stringify(runnerConfig, null, 2), 4)} + sentinel.py: | +${indentBlock(sentinelRunnerPython(), 4)} +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: ${config.serviceAccountName} + namespace: ${options.namespace} + labels: + app.kubernetes.io/name: ${config.cronJobName} + app.kubernetes.io/part-of: platform-infra + app.kubernetes.io/managed-by: unidesk +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: ${config.serviceAccountName} + namespace: ${options.namespace} + labels: + app.kubernetes.io/name: ${config.cronJobName} + app.kubernetes.io/part-of: platform-infra + app.kubernetes.io/managed-by: unidesk +rules: + - apiGroups: [""] + resources: ["configmaps"] + verbs: ["get", "create", "update", "patch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: ${config.serviceAccountName} + namespace: ${options.namespace} + labels: + app.kubernetes.io/name: ${config.cronJobName} + app.kubernetes.io/part-of: platform-infra + app.kubernetes.io/managed-by: unidesk +subjects: + - kind: ServiceAccount + name: ${config.serviceAccountName} + namespace: ${options.namespace} +roleRef: + kind: Role + name: ${config.serviceAccountName} + apiGroup: rbac.authorization.k8s.io +--- +apiVersion: batch/v1 +kind: CronJob +metadata: + name: ${config.cronJobName} + namespace: ${options.namespace} + labels: + app.kubernetes.io/name: ${config.cronJobName} + app.kubernetes.io/part-of: platform-infra + app.kubernetes.io/managed-by: unidesk +spec: + schedule: "${config.schedule}" + concurrencyPolicy: Forbid + suspend: ${suspend} + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 5 + jobTemplate: + spec: + ttlSecondsAfterFinished: 3600 + activeDeadlineSeconds: ${activeDeadlineSeconds} + template: + metadata: + labels: + app.kubernetes.io/name: ${config.cronJobName} + app.kubernetes.io/part-of: platform-infra + app.kubernetes.io/managed-by: unidesk + spec: + serviceAccountName: ${config.serviceAccountName} + restartPolicy: Never + containers: + - name: sentinel + image: ${config.image} + imagePullPolicy: IfNotPresent + command: ["python3", "/opt/sentinel/sentinel.py"] + env: + - name: ADMIN_EMAIL + valueFrom: + configMapKeyRef: + name: sub2api-config + key: ADMIN_EMAIL + optional: true + - name: ADMIN_PASSWORD + valueFrom: + secretKeyRef: + name: ${options.appSecretName} + key: ADMIN_PASSWORD + optional: true + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + volumeMounts: + - name: sentinel-config + mountPath: /opt/sentinel + readOnly: true + - name: sentinel-profiles + mountPath: /opt/sentinel-secrets + readOnly: true + volumes: + - name: sentinel-config + configMap: + name: ${config.configMapName} + defaultMode: 0555 + - name: sentinel-profiles + secret: + secretName: ${config.credentialsSecretName} +`; +} + +export function sentinelRunnerPython(): string { + return String.raw`#!/usr/bin/env python3 +import base64 +import hashlib +import json +import math +import os +import random +import ssl +import time +import traceback +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone, timedelta +from urllib import error, request + +CONFIG_PATH = "/opt/sentinel/config.json" +PROFILES_PATH = "/opt/sentinel-secrets/profiles.json" +STATE_KEY = "state.json" + +def utc_now(): + return datetime.now(timezone.utc) + +def iso(dt): + return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + +def parse_iso(value): + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except Exception: + return None + +def load_json(path): + with open(path, encoding="utf-8") as handle: + return json.load(handle) + +def sha(value): + return hashlib.sha256(str(value).encode("utf-8", errors="replace")).hexdigest()[:16] + +def preview(value, limit=160): + if not isinstance(value, str): + value = str(value) + value = " ".join(value.replace("\r", " ").replace("\n", " ").split()) + return value[:limit] + +def day_key(now): + return now.strftime("%Y-%m-%d") + +def add_minutes(now, minutes, jitter_percent=0): + minutes = float(minutes) + if jitter_percent: + factor = 1 + random.uniform(-jitter_percent, jitter_percent) / 100 + minutes = max(1, minutes * factor) + return now + timedelta(minutes=minutes) + +def estimate_tokens(text): + if not isinstance(text, str) or not text: + return 0 + return max(1, math.ceil(len(text) / 4)) + +class KubeClient: + def __init__(self, namespace): + self.namespace = namespace + with open("/var/run/secrets/kubernetes.io/serviceaccount/token", encoding="utf-8") as handle: + self.token = handle.read().strip() + self.base = "https://kubernetes.default.svc" + self.context = ssl.create_default_context(cafile="/var/run/secrets/kubernetes.io/serviceaccount/ca.crt") + + def api(self, method, path, payload=None): + body = None if payload is None else json.dumps(payload).encode("utf-8") + req = request.Request( + self.base + path, + data=body, + method=method, + headers={ + "Authorization": "Bearer " + self.token, + "Accept": "application/json", + "Content-Type": "application/json", + }, + ) + try: + with request.urlopen(req, timeout=15, context=self.context) as resp: + raw = resp.read() + return resp.status, json.loads(raw.decode("utf-8")) if raw else None + except error.HTTPError as exc: + raw = exc.read() + try: + parsed = json.loads(raw.decode("utf-8")) if raw else None + except Exception: + parsed = {"message": raw.decode("utf-8", errors="replace")} + return exc.code, parsed + + def get_configmap(self, name): + status, data = self.api("GET", f"/api/v1/namespaces/{self.namespace}/configmaps/{name}") + if status == 404: + return None + if status >= 300: + raise RuntimeError(f"get configmap {name} failed: {status} {data}") + return data + + def create_configmap(self, name, state): + payload = { + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": { + "name": name, + "namespace": self.namespace, + "labels": { + "app.kubernetes.io/part-of": "platform-infra", + "app.kubernetes.io/managed-by": "unidesk", + "unidesk.ai/state-purpose": "sub2api-account-sentinel", + }, + }, + "data": {STATE_KEY: json.dumps(state, ensure_ascii=False, indent=2)}, + } + status, data = self.api("POST", f"/api/v1/namespaces/{self.namespace}/configmaps", payload) + if status >= 300 and status != 409: + raise RuntimeError(f"create state configmap {name} failed: {status} {data}") + return self.get_configmap(name) + + def update_configmap_state(self, obj, state): + obj.setdefault("data", {})[STATE_KEY] = json.dumps(state, ensure_ascii=False, indent=2) + name = obj["metadata"]["name"] + status, data = self.api("PUT", f"/api/v1/namespaces/{self.namespace}/configmaps/{name}", obj) + if status >= 300: + raise RuntimeError(f"update state configmap {name} failed: {status} {data}") + return data + +def default_state(): + return { + "version": 1, + "accounts": {}, + "ledger": {}, + "history": [], + } + +def load_state(kube, config): + name = config["state"]["configMapName"] + obj = kube.get_configmap(name) + if obj is None: + obj = kube.create_configmap(name, default_state()) + raw = (obj.get("data") or {}).get(STATE_KEY) + try: + state = json.loads(raw) if raw else default_state() + except Exception: + state = default_state() + state["stateLoadWarning"] = "invalid-state-json-reset" + state.setdefault("version", 1) + state.setdefault("accounts", {}) + state.setdefault("ledger", {}) + state.setdefault("history", []) + return obj, state + +def http_json(method, url, headers=None, payload=None, timeout=30, max_bytes=65536): + body = None if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8") + req = request.Request(url, data=body, method=method, headers=headers or {}) + started = time.time() + try: + with request.urlopen(req, timeout=timeout) as resp: + raw = resp.read(max_bytes + 1) + too_large = len(raw) > max_bytes + if too_large: + raw = raw[:max_bytes] + text = raw.decode("utf-8", errors="replace") + parsed = None + try: + parsed = json.loads(text) if text.strip() else None + except Exception: + parsed = None + return { + "ok": 200 <= resp.status < 300 and not too_large, + "status": resp.status, + "json": parsed, + "text": text, + "tooLarge": too_large, + "durationMs": int((time.time() - started) * 1000), + } + except error.HTTPError as exc: + raw = exc.read(max_bytes + 1) + text = raw.decode("utf-8", errors="replace") + parsed = None + try: + parsed = json.loads(text) if text.strip() else None + except Exception: + parsed = None + return { + "ok": False, + "status": exc.code, + "json": parsed, + "text": text, + "tooLarge": len(raw) > max_bytes, + "durationMs": int((time.time() - started) * 1000), + "error": str(exc), + } + except Exception as exc: + return { + "ok": False, + "status": 0, + "json": None, + "text": "", + "tooLarge": False, + "durationMs": int((time.time() - started) * 1000), + "error": str(exc), + } + +def find_token(value): + if isinstance(value, dict): + for key in ("access_token", "token"): + if isinstance(value.get(key), str) and value[key]: + return value[key] + for item in value.values(): + found = find_token(item) + if found: + return found + if isinstance(value, list): + for item in value: + found = find_token(item) + if found: + return found + return None + +class Sub2ApiAdmin: + def __init__(self, config): + self.base = config["service"]["baseUrl"].rstrip("/") + self.email = os.environ.get("ADMIN_EMAIL") or config["service"]["adminEmailDefault"] + self.password = os.environ.get("ADMIN_PASSWORD") or "" + self.token = None + self.accounts_by_name = None + + def login(self): + if self.token: + return self.token + if not self.password: + raise RuntimeError("ADMIN_PASSWORD is missing") + resp = http_json("POST", self.base + "/api/v1/auth/login", {"Content-Type": "application/json"}, {"email": self.email, "password": self.password}, timeout=15) + token = find_token(resp.get("json")) + if not resp["ok"] or not token: + raise RuntimeError("admin login failed") + self.token = token + return token + + def request(self, method, path, payload=None): + token = self.login() + headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"} + resp = http_json(method, self.base + path, headers, payload, timeout=20) + if not resp["ok"]: + raise RuntimeError(f"admin {method} {path} failed: {resp.get('status')} {preview(resp.get('text', ''), 300)}") + parsed = resp.get("json") + if isinstance(parsed, dict) and parsed.get("code") not in (None, 0): + raise RuntimeError(f"admin {method} {path} failed: code={parsed.get('code')} message={parsed.get('message')}") + if isinstance(parsed, dict) and "data" in parsed: + return parsed["data"] + return parsed + + def accounts(self): + if self.accounts_by_name is not None: + return self.accounts_by_name + data = self.request("GET", "/api/v1/admin/accounts?page=1&page_size=500&platform=openai&type=apikey&search=unidesk-codex-") + items = [] + if isinstance(data, list): + items = data + elif isinstance(data, dict): + for key in ("items", "accounts"): + if isinstance(data.get(key), list): + items = data[key] + break + self.accounts_by_name = {item.get("name"): item for item in items if isinstance(item, dict) and isinstance(item.get("name"), str)} + return self.accounts_by_name + + def set_schedulable(self, account_name, schedulable): + account = self.accounts().get(account_name) + if not account or account.get("id") is None: + raise RuntimeError(f"account {account_name} not found") + self.request("POST", f"/api/v1/admin/accounts/{account['id']}/schedulable", {"schedulable": bool(schedulable)}) + return {"accountId": account.get("id"), "schedulable": bool(schedulable)} + + def recover_state(self, account_name): + account = self.accounts().get(account_name) + if not account or account.get("id") is None: + return {"skipped": True, "reason": "account-not-found"} + try: + self.request("POST", f"/api/v1/admin/accounts/{account['id']}/recover-state", {}) + return {"ok": True, "accountId": account.get("id")} + except Exception as exc: + return {"ok": False, "accountId": account.get("id"), "error": str(exc)} + +def upstream_responses_url(base_url): + base = str(base_url).rstrip("/") + if base.endswith("/v1"): + return base + "/responses" + return base + "/v1/responses" + +def output_text(parsed): + if isinstance(parsed, dict) and isinstance(parsed.get("output_text"), str): + return parsed["output_text"] + parts = [] + output = parsed.get("output") if isinstance(parsed, dict) else None + if isinstance(output, list): + for item in output: + if not isinstance(item, dict): + continue + content = item.get("content") + if not isinstance(content, list): + continue + for block in content: + if isinstance(block, dict) and isinstance(block.get("text"), str): + parts.append(block["text"]) + return "\n".join(parts) + +def usage_from(parsed, prompt, out, config): + usage = parsed.get("usage") if isinstance(parsed, dict) and isinstance(parsed.get("usage"), dict) else {} + input_tokens = usage.get("input_tokens") + output_tokens = usage.get("output_tokens") + estimated = False + if not isinstance(input_tokens, int): + input_tokens = estimate_tokens(prompt) + estimated = True + if not isinstance(output_tokens, int): + output_tokens = estimate_tokens(out) + estimated = True + total = usage.get("total_tokens") + if not isinstance(total, int): + total = input_tokens + output_tokens + estimated = True + cost = ( + input_tokens * float(config["pricing"]["usdPer1MInputTokens"]) + + output_tokens * float(config["pricing"]["usdPer1MOutputTokens"]) + ) / 1000000 + return { + "inputTokens": input_tokens, + "outputTokens": output_tokens, + "totalTokens": total, + "estimated": estimated, + "estimatedCostUsd": cost, + } + +def probe_account(profile, config, purpose): + marker = config["marker"]["prefix"] + "_" + hashlib.sha256((profile["accountName"] + str(time.time()) + str(random.random())).encode()).hexdigest()[:10] + prompt = "Return exactly this marker and no other text: " + marker + payload = { + "model": config["model"], + "input": prompt, + "stream": False, + "store": False, + "max_output_tokens": int(config["probe"]["maxOutputTokens"]), + } + headers = { + "Authorization": "Bearer " + profile["apiKey"], + "Content-Type": "application/json", + "X-Request-ID": "unidesk-account-sentinel-" + hashlib.sha256(marker.encode()).hexdigest()[:16], + } + if profile.get("upstreamUserAgent"): + headers["User-Agent"] = profile["upstreamUserAgent"] + resp = http_json( + "POST", + upstream_responses_url(profile["baseUrl"]), + headers, + payload, + timeout=int(config["probe"]["timeoutSeconds"]), + max_bytes=int(config["probe"]["maxResponseBytes"]), + ) + parsed = resp.get("json") + out = output_text(parsed) + trimmed = out.strip() + marker_matched = trimmed == marker if config["marker"].get("exact", True) else marker in trimmed + usage = usage_from(parsed if isinstance(parsed, dict) else {}, prompt, out or resp.get("text", ""), config) + http_success = isinstance(resp.get("status"), int) and 200 <= resp.get("status") < 300 + ok = resp["ok"] and marker_matched + mismatch = http_success and not marker_matched + return { + "accountName": profile["accountName"], + "profile": profile.get("profile"), + "purpose": purpose, + "ok": ok, + "markerMatched": marker_matched, + "markerHash": sha(marker), + "httpStatus": resp.get("status"), + "transportOk": resp["ok"], + "tooLarge": resp.get("tooLarge"), + "durationMs": resp.get("durationMs"), + "outputHash": sha(out), + "outputPreview": "" if marker_matched else preview(out or resp.get("text", ""), 160), + "bodyPreviewHash": sha(resp.get("text", "")), + "error": resp.get("error"), + "usage": usage, + "mismatch": mismatch, + "transportFailure": not resp["ok"], + } + +def ledger_for(state, now): + day = day_key(now) + ledger = state.setdefault("ledger", {}).setdefault(day, {"inputTokens": 0, "outputTokens": 0, "totalTokens": 0, "estimatedCostUsd": 0, "requestCount": 0}) + return day, ledger + +def account_day(account_state, day): + return account_state.setdefault("daily", {}).setdefault(day, {"inputTokens": 0, "outputTokens": 0, "totalTokens": 0, "estimatedCostUsd": 0, "requestCount": 0}) + +def add_usage(state, account_state, now, usage): + day, ledger = ledger_for(state, now) + daily = account_day(account_state, day) + for target in (ledger, daily): + target["inputTokens"] = int(target.get("inputTokens") or 0) + int(usage.get("inputTokens") or 0) + target["outputTokens"] = int(target.get("outputTokens") or 0) + int(usage.get("outputTokens") or 0) + target["totalTokens"] = int(target.get("totalTokens") or 0) + int(usage.get("totalTokens") or 0) + target["estimatedCostUsd"] = round(float(target.get("estimatedCostUsd") or 0) + float(usage.get("estimatedCostUsd") or 0), 8) + target["requestCount"] = int(target.get("requestCount") or 0) + 1 + +def due_time(account_state): + quarantine = account_state.get("quarantine") + if isinstance(quarantine, dict) and quarantine.get("active") is True: + return parse_iso(quarantine.get("until")) + return parse_iso(account_state.get("nextProbeAfter")) + +def choose_due_profiles(profiles, state, config, now): + day, ledger = ledger_for(state, now) + due = [] + accounts = state.setdefault("accounts", {}) + for profile in profiles: + name = profile["accountName"] + account_state = accounts.setdefault(name, {}) + when = due_time(account_state) + if when is None or when <= now: + quarantine = account_state.get("quarantine") + purpose = "recovery" if isinstance(quarantine, dict) and quarantine.get("active") is True else "health" + due.append({"profile": profile, "purpose": purpose, "dueAt": iso(when) if when else None}) + due.sort(key=lambda item: item["dueAt"] or "") + return due, {"selected": len(due), "due": len(due), "limit": "all-due", "budgetMode": "record-only", "ledger": ledger} + +def next_success_interval(account_state, config): + streak = int(account_state.get("successStreak") or 0) + previous = int(account_state.get("successIntervalMinutes") or 0) + initial = int(config["cadence"]["successInitialIntervalMinutes"]) + maximum = int(config["cadence"]["successMaxIntervalMinutes"]) + multiplier = int(config["cadence"]["successBackoffMultiplier"]) + return initial if streak <= 0 or previous <= 0 else min(maximum, max(initial, previous * multiplier)) + +def next_freeze_interval(account_state, config, was_recovery): + quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else {} + previous = int(quarantine.get("intervalMinutes") or 0) + initial = int(config["freeze"]["initialTtlMinutes"]) + maximum = int(config["freeze"]["maxTtlMinutes"]) + multiplier = int(config["freeze"]["backoffMultiplier"]) + if was_recovery and previous > 0: + return min(maximum, max(initial, previous * multiplier)) + return initial + +def apply_result(result, state, config, now, admin): + name = result["accountName"] + account_state = state.setdefault("accounts", {}).setdefault(name, {}) + add_usage(state, account_state, now, result.get("usage") or {}) + actions_enabled = bool(config["actions"]["enabled"]) + quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else None + was_recovery = bool(quarantine and quarantine.get("active") is True) + action = {"taken": False, "type": None} + if result.get("ok") is True: + if was_recovery: + if actions_enabled and quarantine.get("applied") is True: + try: + action = {"taken": True, "type": "restore", "result": admin.set_schedulable(name, True), "recoverState": admin.recover_state(name)} + except Exception as exc: + action = {"taken": False, "type": "restore-failed", "error": str(exc)} + account_state["quarantine"] = {"active": False, "clearedAt": iso(now), "lastApplied": quarantine.get("applied") is True} + account_state["successStreak"] = 0 + account_state["successIntervalMinutes"] = 0 + interval = next_success_interval(account_state, config) + account_state["successStreak"] = int(account_state.get("successStreak") or 0) + 1 + account_state["successIntervalMinutes"] = interval + account_state["nextProbeAfter"] = iso(add_minutes(now, interval, int(config["cadence"]["jitterPercent"]))) + account_state["lastOkAt"] = iso(now) + account_state["lastStatus"] = "ok" + else: + should_freeze = bool(result.get("mismatch")) and bool(config["actions"]["freezeOnMarkerMismatch"]) + if result.get("transportFailure") and bool(config["actions"].get("freezeOnTransportError")): + should_freeze = True + if should_freeze: + interval = next_freeze_interval(account_state, config, was_recovery) + until = add_minutes(now, interval, int(config["freeze"]["jitterPercent"])) + applied = False + if actions_enabled: + try: + action = {"taken": True, "type": "freeze", "result": admin.set_schedulable(name, False)} + applied = True + except Exception as exc: + action = {"taken": False, "type": "freeze-failed", "error": str(exc)} + else: + action = {"taken": False, "type": "would-freeze"} + account_state["quarantine"] = { + "active": True, + "applied": applied, + "until": iso(until), + "intervalMinutes": interval, + "reason": "marker-mismatch" if result.get("mismatch") else "transport-failure", + "markerHash": result.get("markerHash"), + "outputHash": result.get("outputHash"), + "lastBadAt": iso(now), + } + account_state["nextProbeAfter"] = iso(until) + account_state["successStreak"] = 0 + account_state["successIntervalMinutes"] = 0 + account_state["lastStatus"] = "quarantined" + else: + retry = int(config["probe"]["transportRetryMinutes"]) + account_state["nextProbeAfter"] = iso(add_minutes(now, retry, int(config["cadence"]["jitterPercent"]))) + account_state["lastStatus"] = "transport-failed-no-freeze" + account_state["lastFailureAt"] = iso(now) + account_state["lastProbeAt"] = iso(now) + account_state["lastProbe"] = { + "ok": result.get("ok"), + "purpose": result.get("purpose"), + "httpStatus": result.get("httpStatus"), + "durationMs": result.get("durationMs"), + "markerMatched": result.get("markerMatched"), + "outputHash": result.get("outputHash"), + "outputPreview": result.get("outputPreview"), + "usage": result.get("usage"), + "action": action, + } + return action + +def reconcile_active_quarantines(state, config, now, admin): + actions = [] + if not config["actions"]["enabled"]: + return actions + for name, account_state in state.setdefault("accounts", {}).items(): + quarantine = account_state.get("quarantine") + if not isinstance(quarantine, dict) or quarantine.get("active") is not True: + continue + until = parse_iso(quarantine.get("until")) + if until is not None and until <= now: + continue + if quarantine.get("applied") is not True: + actions.append({"accountName": name, "type": "virtual-freeze-not-applied", "ok": True}) + continue + try: + admin.set_schedulable(name, False) + actions.append({"accountName": name, "type": "reassert-freeze", "ok": True}) + except Exception as exc: + actions.append({"accountName": name, "type": "reassert-freeze", "ok": False, "error": str(exc)}) + return actions + +def main(): + now = utc_now() + config = load_json(CONFIG_PATH) + profiles = load_json(PROFILES_PATH).get("profiles") or [] + namespace = os.environ.get("POD_NAMESPACE") or "platform-infra" + kube = KubeClient(namespace) + state_obj, state = load_state(kube, config) + admin = Sub2ApiAdmin(config) + reconcile = reconcile_active_quarantines(state, config, now, admin) + due, selection = choose_due_profiles(profiles, state, config, now) + results = [] + actions = [] + if config["monitor"]["enabled"] and due: + with ThreadPoolExecutor(max_workers=max(1, len(due))) as executor: + futures = [executor.submit(probe_account, item["profile"], config, item["purpose"]) for item in due] + for future in as_completed(futures): + result = future.result() + results.append(result) + actions.append({"accountName": result["accountName"], **apply_result(result, state, config, now, admin)}) + history = state.setdefault("history", []) + run_summary = { + "at": iso(now), + "monitorEnabled": bool(config["monitor"]["enabled"]), + "actionsEnabled": bool(config["actions"]["enabled"]), + "profileCount": len(profiles), + "selected": len(due), + "okCount": sum(1 for item in results if item.get("ok") is True), + "mismatchCount": sum(1 for item in results if item.get("mismatch") is True), + "transportFailureCount": sum(1 for item in results if item.get("transportFailure") is True), + "actionsTaken": sum(1 for item in actions if item.get("taken") is True), + "selection": selection, + "reconcile": reconcile[-20:], + } + history.append(run_summary) + del history[:-int(config["state"]["historyLimit"])] + state["lastRun"] = run_summary + kube.update_configmap_state(state_obj, state) + print(json.dumps({ + "ok": True, + "summary": run_summary, + "results": [{ + "accountName": item.get("accountName"), + "purpose": item.get("purpose"), + "ok": item.get("ok"), + "markerMatched": item.get("markerMatched"), + "httpStatus": item.get("httpStatus"), + "durationMs": item.get("durationMs"), + "usage": item.get("usage"), + "outputHash": item.get("outputHash"), + "outputPreview": item.get("outputPreview"), + "error": item.get("error"), + } for item in results], + "actions": actions, + "valuesPrinted": False, + }, ensure_ascii=False)) + +if __name__ == "__main__": + try: + main() + except Exception as exc: + print(json.dumps({ + "ok": False, + "error": str(exc), + "traceback": traceback.format_exc()[-4000:], + "valuesPrinted": False, + }, ensure_ascii=False)) + raise +`; +} + +function indentBlock(value: string, spaces: number): string { + const prefix = " ".repeat(spaces); + return value.split("\n").map((line) => `${prefix}${line}`).join("\n"); +} + +function valueAt(value: Record, key: string): unknown { + return Object.prototype.hasOwnProperty.call(value, key) ? value[key] : undefined; +} + +function readBoolean(value: unknown, key: string, fallback: boolean): boolean { + if (value === undefined || value === null) return fallback; + if (typeof value === "boolean") return value; + if (typeof value === "string" && value.trim() === "true") return true; + if (typeof value === "string" && value.trim() === "false") return false; + throw new Error(`${key} must be a boolean`); +} + +function readString(value: unknown, key: string, fallback: string): string { + if (value === undefined || value === null) return fallback; + if (typeof value !== "string" || value.trim().length === 0) throw new Error(`${key} must be a non-empty string`); + if (/[\r\n]/u.test(value)) throw new Error(`${key} must not contain newlines`); + return value.trim(); +} + +function readDnsName(value: unknown, key: string, fallback: string): string { + const text = readString(value, key, fallback); + if (!/^[a-z0-9]([-a-z0-9]*[a-z0-9])?$/u.test(text)) throw new Error(`${key} must be a Kubernetes DNS label`); + return text; +} + +function readModelName(value: unknown, key: string, fallback: string): string { + const text = readString(value, key, fallback); + if (!/^[A-Za-z0-9._:-]+$/u.test(text)) throw new Error(`${key} has an unsupported model name`); + return text; +} + +function readMarkerPrefix(value: unknown, key: string, fallback: string): string { + const text = readString(value, key, fallback); + if (!/^[A-Za-z0-9_-]{2,32}$/u.test(text)) throw new Error(`${key} must be 2-32 chars of letters, digits, _ or -`); + return text; +} + +function readInt(value: unknown, key: string, fallback: number, min: number, max: number): number { + if (value === undefined || value === null) return fallback; + const parsed = typeof value === "number" ? value : typeof value === "string" && value.trim() ? Number(value) : Number.NaN; + if (!Number.isInteger(parsed) || parsed < min || parsed > max) throw new Error(`${key} must be an integer from ${min} to ${max}`); + return parsed; +} + +function readNumber(value: unknown, key: string, fallback: number, min: number, max: number): number { + if (value === undefined || value === null) return fallback; + const parsed = typeof value === "number" ? value : typeof value === "string" && value.trim() ? Number(value) : Number.NaN; + if (!Number.isFinite(parsed) || parsed < min || parsed > max) throw new Error(`${key} must be a number from ${min} to ${max}`); + return parsed; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/scripts/src/platform-infra-sub2api-codex.ts b/scripts/src/platform-infra-sub2api-codex.ts index a0a821b0..7f71a3e4 100644 --- a/scripts/src/platform-infra-sub2api-codex.ts +++ b/scripts/src/platform-infra-sub2api-codex.ts @@ -4,6 +4,14 @@ import { homedir } from "node:os"; import { join } from "node:path"; import type { UniDeskConfig } from "./config"; import { rootPath } from "./config"; +import { + codexPoolSentinelSummary, + defaultCodexPoolSentinelConfig, + readCodexPoolSentinelConfig, + renderCodexPoolSentinelManifest, + type CodexPoolSentinelConfig, + type CodexPoolSentinelProfileSecret, +} from "./platform-infra-sub2api-codex-sentinel"; import { runSshCommandCapture, type SshCaptureResult } from "./ssh"; const g14K3sRoute = "G14:k3s"; @@ -89,6 +97,7 @@ interface CodexPoolConfig { profiles: CodexPoolProfileConfig[]; publicExposure: CodexPoolPublicExposureConfig; localCodex: CodexPoolLocalCodexConfig; + sentinel: CodexPoolSentinelConfig; } interface CodexPoolProfileConfig { @@ -174,6 +183,8 @@ export function codexPoolHelp(): unknown { poolApiKeySecretKey: pool.apiKeySecretKey, publicBaseUrl: pool.publicExposure.enabled ? pool.publicExposure.publicBaseUrl : null, masterBaseUrl: pool.publicExposure.enabled ? pool.publicExposure.masterBaseUrl : null, + sentinelMonitorEnabled: pool.sentinel.monitor.enabled, + sentinelActionsEnabled: pool.sentinel.actions.enabled, secretValuesPrinted: false, }, }; @@ -243,6 +254,9 @@ function codexPoolPlan(options: DisclosureOptions = { full: false, raw: false }) accountType: "openai/apikey", grouping: `All discovered Codex profiles are bound to one Sub2API group named ${pool.groupName}.`, unifiedApiKey: `The client-facing API_KEY is controlled by k3s Secret ${namespace}/${pool.apiKeySecretName}.${pool.apiKeySecretKey}.`, + sentinel: pool.sentinel.monitor.enabled + ? `Account sentinel is enabled as k8s CronJob ${namespace}/${pool.sentinel.cronJobName}; actions.enabled=${pool.sentinel.actions.enabled}.` + : "Account sentinel monitoring is disabled by YAML.", publicExposure: pool.publicExposure.enabled ? `Default Codex consumers use ${codexConsumerBaseUrl(pool)}; bounded master-local probes may use ${pool.publicExposure.masterBaseUrl}. FRP proxy ${pool.publicExposure.proxyName} maps public ${pool.publicExposure.publicBaseUrl} to ${pool.publicExposure.localIP}:${pool.publicExposure.localPort}.` : "Public FRP exposure is disabled by YAML.", @@ -272,6 +286,15 @@ async function codexPoolSync(config: UniDeskConfig, options: SyncOptions): Promi const payload = { pruneRemoved: options.pruneRemoved, + sentinel: { + manifest: renderCodexPoolSentinelManifest(pool.sentinel, sentinelProfileSecrets(profiles), { + namespace, + serviceName, + serviceDns, + appSecretName, + }), + summary: codexPoolSentinelSummary(pool.sentinel), + }, pool: { groupName: pool.groupName, apiKeyName: pool.apiKeyName, @@ -656,6 +679,7 @@ function readCodexPoolConfig(): CodexPoolConfig { profiles, publicExposure: readPublicExposureConfig(parsed.publicExposure, defaults.publicExposure), localCodex: readLocalCodexConfig(parsed.localCodex, defaults.localCodex), + sentinel: readCodexPoolSentinelConfig(parsed.sentinel, defaults.sentinel, codexPoolConfigPath), }; validateKubernetesName(config.groupName, "pool.groupName", false); validateKubernetesName(config.apiKeySecretName, "pool.apiKeySecretName", true); @@ -717,6 +741,7 @@ function defaultCodexPoolConfig(): CodexPoolConfig { responsesWebSocketsV2: true, responsesSmokeModel: "gpt-5.5", }, + sentinel: defaultCodexPoolSentinelConfig(), }; } @@ -1190,6 +1215,7 @@ function codexPoolConfigSummary(pool: CodexPoolConfig): Record profileCount: pool.profiles.length, publicExposure: publicExposureSummary(pool), localCodex: pool.localCodex, + sentinel: codexPoolSentinelSummary(pool.sentinel), disclosure: { full: "bun scripts/cli.ts platform-infra sub2api codex-pool plan --full", }, @@ -1432,6 +1458,67 @@ function compactGatewayCompactRecent(block: unknown): unknown { }; } +function compactSentinelStatus(block: unknown): unknown { + if (!isRecord(block)) return block; + const runtime = isRecord(block.runtime) ? block.runtime : block; + const desired = isRecord(runtime.desired) ? runtime.desired : {}; + const cronJob = isRecord(runtime.cronJob) ? runtime.cronJob : {}; + const secret = isRecord(runtime.secret) ? runtime.secret : {}; + const configMap = isRecord(runtime.configMap) ? runtime.configMap : {}; + const state = isRecord(runtime.state) ? runtime.state : {}; + const freezeReassert = isRecord(block.freezeReassert) ? block.freezeReassert : {}; + return { + ok: block.ok, + action: block.action, + desired: { + monitorEnabled: desired.monitorEnabled, + actionsEnabled: desired.actionsEnabled, + schedule: desired.schedule, + cronJobName: desired.cronJobName, + configMapName: desired.configMapName, + credentialsSecretName: desired.credentialsSecretName, + stateConfigMapName: desired.stateConfigMapName, + }, + cronJob: { + exists: cronJob.exists, + schedule: cronJob.schedule, + suspend: cronJob.suspend, + lastScheduleTime: cronJob.lastScheduleTime, + active: cronJob.active, + error: cronJob.error, + }, + secret: { + exists: secret.exists, + profileSecretPresent: secret.profileSecretPresent, + valuesPrinted: false, + error: secret.error, + }, + configMap: { + exists: configMap.exists, + configPresent: configMap.configPresent, + runnerPresent: configMap.runnerPresent, + error: configMap.error, + }, + state: { + exists: state.exists, + accountCount: state.accountCount, + quarantinedCount: state.quarantinedCount, + quarantined: state.quarantined, + recentAccounts: state.recentAccounts, + lastRun: state.lastRun, + error: state.error, + }, + freezeReassert: Object.keys(freezeReassert).length > 0 ? { + ok: freezeReassert.ok, + skipped: freezeReassert.skipped, + reason: freezeReassert.reason, + itemCount: freezeReassert.itemCount, + attentionItems: freezeReassert.attentionItems, + } : undefined, + valuesPrinted: false, + }; +} + function codexPoolValidationSummary(parsed: Record | null): Record | null { if (parsed === null) return null; const validation = isRecord(parsed.validation) ? parsed.validation : {}; @@ -1451,6 +1538,7 @@ function codexPoolValidationSummary(parsed: Record | null): Rec loadFactor: compactStatusBlock(parsed.loadFactor, ["accountName", "accountId", "expectedLoadFactor", "runtimeLoadFactor", "priority", "status", "schedulable", "ok"]), webSocketsV2: compactStatusBlock(parsed.webSocketsV2, ["accountName", "accountId", "expectedMode", "runtimeMode", "runtimeEnabled", "status", "schedulable", "ok"]), tempUnschedulable: compactTempUnschedulableStatus(parsed.tempUnschedulable), + sentinel: compactSentinelStatus(parsed.sentinel), runtimeCapabilities: { ok: runtimeCapabilities.ok, runtimeImage: runtimeCapabilities.runtimeImage, @@ -1528,10 +1616,28 @@ function poolTarget(pool = readCodexPoolConfig()): Record { defaultAccountPriority: pool.defaultAccountPriority, defaultAccountCapacity: pool.defaultAccountCapacity, defaultAccountLoadFactor: pool.defaultAccountLoadFactor, + sentinel: { + monitorEnabled: pool.sentinel.monitor.enabled, + actionsEnabled: pool.sentinel.actions.enabled, + cronJobName: pool.sentinel.cronJobName, + stateConfigMapName: pool.sentinel.stateConfigMapName, + }, valuesPrinted: false, }; } +function sentinelProfileSecrets(profiles: CodexProfile[]): CodexPoolSentinelProfileSecret[] { + return profiles + .filter((profile) => profile.ok && profile.apiKey !== null && profile.apiKey.length > 0) + .map((profile) => ({ + accountName: profile.accountName, + profile: profile.profile, + baseUrl: profile.baseUrl, + apiKey: profile.apiKey ?? "", + upstreamUserAgent: profile.upstreamUserAgent, + })); +} + function publicExposureSummary(pool: CodexPoolConfig): Record { return { enabled: pool.publicExposure.enabled, @@ -2262,6 +2368,7 @@ import string import subprocess import sys import time +from datetime import datetime from urllib.parse import quote NAMESPACE = "${namespace}" @@ -2284,6 +2391,7 @@ EXPECTED_ACCOUNT_CAPACITIES = ${JSON.stringify(desiredAccountCapacityMap(pool))} EXPECTED_ACCOUNT_LOAD_FACTORS = ${JSON.stringify(desiredAccountLoadFactorMap(pool))} EXPECTED_ACCOUNT_WS_MODES = json.loads(${JSON.stringify(JSON.stringify(desiredAccountWebSocketsV2ModeMap(pool)))}) EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE = json.loads(${JSON.stringify(JSON.stringify(desiredAccountTempUnschedulableMap(pool)))}) +SENTINEL_CONFIG = json.loads(${JSON.stringify(JSON.stringify(pool.sentinel))}) MODE = "${mode}" PAYLOAD_B64 = "${encodedPayload}" @@ -2726,6 +2834,201 @@ def ensure_api_key_secret(group_id): raise RuntimeError(f"apply API key secret failed: {text(proc.stderr, 1000)}") return api_key, secret_action, text(proc.stdout, 1000) +def apply_sentinel_manifest(manifest): + if not isinstance(manifest, str) or not manifest.strip(): + return { + "ok": False, + "action": "missing-manifest", + "valuesPrinted": False, + } + proc = kubectl(["apply", "--server-side", "--force-conflicts", f"--field-manager={FIELD_MANAGER}", "-f", "-"], manifest) + if proc.returncode != 0: + return { + "ok": False, + "action": "apply-failed", + "stdoutTail": text(proc.stdout, 2000), + "stderrTail": text(proc.stderr, 4000), + "valuesPrinted": False, + } + status = sentinel_runtime_status() + return { + "ok": status.get("ok") is True, + "action": "applied", + "stdoutTail": text(proc.stdout, 2000), + "runtime": status, + "valuesPrinted": False, + } + +def safe_kube_json(args, label): + proc = kubectl([*args, "-o", "json"]) + if proc.returncode != 0: + return None, {"label": label, "exitCode": proc.returncode, "stderrTail": text(proc.stderr, 1000)} + try: + return json.loads(proc.stdout.decode("utf-8")), None + except Exception as exc: + return None, {"label": label, "exitCode": proc.returncode, "error": str(exc), "stdoutTail": text(proc.stdout, 1000)} + +def sentinel_runtime_status(): + cfg = SENTINEL_CONFIG + cronjob_name = cfg.get("cronJobName") + secret_name = cfg.get("credentialsSecretName") + configmap_name = cfg.get("configMapName") + state_name = cfg.get("stateConfigMapName") + cronjob, cronjob_error = safe_kube_json(["-n", NAMESPACE, "get", "cronjob", cronjob_name], f"cronjob/{cronjob_name}") + secret, secret_error = safe_kube_json(["-n", NAMESPACE, "get", "secret", secret_name], f"secret/{secret_name}") + configmap, configmap_error = safe_kube_json(["-n", NAMESPACE, "get", "configmap", configmap_name], f"configmap/{configmap_name}") + state_cm, state_error = safe_kube_json(["-n", NAMESPACE, "get", "configmap", state_name], f"configmap/{state_name}") + state = None + if isinstance(state_cm, dict): + raw_state = (state_cm.get("data") or {}).get("state.json") + if isinstance(raw_state, str) and raw_state: + try: + state = json.loads(raw_state) + except Exception as exc: + state = {"parseError": str(exc)} + accounts = (state.get("accounts") or {}) if isinstance(state, dict) else {} + quarantined = [] + recent_accounts = [] + for name, account_state in accounts.items(): + if not isinstance(account_state, dict): + continue + quarantine = account_state.get("quarantine") + if isinstance(quarantine, dict) and quarantine.get("active") is True: + quarantined.append({ + "accountName": name, + "until": quarantine.get("until"), + "applied": quarantine.get("applied"), + "reason": quarantine.get("reason"), + "intervalMinutes": quarantine.get("intervalMinutes"), + }) + 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"), + "outputHash": last_probe.get("outputHash"), + "outputPreview": last_probe.get("outputPreview"), + "usage": last_probe.get("usage"), + "action": last_probe.get("action"), + }) + recent_accounts.sort(key=lambda item: item.get("lastProbeAt") or "") + last_run = state.get("lastRun") if isinstance(state, dict) else None + cronjob_spec = cronjob.get("spec") if isinstance(cronjob, dict) else {} + secret_data = secret.get("data") if isinstance(secret, dict) else {} + configmap_data = configmap.get("data") if isinstance(configmap, dict) else {} + ok = cronjob is not None and secret is not None and configmap is not None + return { + "ok": ok, + "desired": { + "monitorEnabled": cfg.get("monitor", {}).get("enabled"), + "actionsEnabled": cfg.get("actions", {}).get("enabled"), + "schedule": cfg.get("schedule"), + "cronJobName": cronjob_name, + "configMapName": configmap_name, + "credentialsSecretName": secret_name, + "stateConfigMapName": state_name, + }, + "cronJob": { + "exists": cronjob is not None, + "schedule": cronjob_spec.get("schedule") if isinstance(cronjob_spec, dict) else None, + "suspend": cronjob_spec.get("suspend") if isinstance(cronjob_spec, dict) else None, + "lastScheduleTime": (cronjob.get("status") or {}).get("lastScheduleTime") if isinstance(cronjob, dict) else None, + "active": len((cronjob.get("status") or {}).get("active") or []) if isinstance(cronjob, dict) else None, + "error": cronjob_error, + }, + "secret": { + "exists": secret is not None, + "profileSecretPresent": isinstance(secret_data, dict) and "profiles.json" in secret_data, + "valuesPrinted": False, + "error": secret_error, + }, + "configMap": { + "exists": configmap is not None, + "configPresent": isinstance(configmap_data, dict) and "config.json" in configmap_data, + "runnerPresent": isinstance(configmap_data, dict) and "sentinel.py" in configmap_data, + "error": configmap_error, + }, + "state": { + "exists": state_cm is not None, + "accountCount": len(accounts), + "quarantinedCount": len(quarantined), + "quarantined": quarantined[-10:], + "recentAccounts": recent_accounts[-12:], + "lastRun": last_run, + "error": state_error, + }, + "valuesPrinted": False, + } + +def parse_epoch_z(value): + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() + except Exception: + return None + +def sentinel_state_object(): + state_name = SENTINEL_CONFIG.get("stateConfigMapName") + if not state_name: + return None + obj, err = safe_kube_json(["-n", NAMESPACE, "get", "configmap", state_name], f"configmap/{state_name}") + if not isinstance(obj, dict): + return None + raw_state = (obj.get("data") or {}).get("state.json") + if not isinstance(raw_state, str) or not raw_state: + return None + try: + return json.loads(raw_state) + except Exception: + return None + +def reassert_sentinel_freezes_after_sync(token): + if (SENTINEL_CONFIG.get("actions") or {}).get("enabled") is not True: + return {"ok": True, "skipped": True, "reason": "actions-disabled", "items": [], "valuesPrinted": False} + state = sentinel_state_object() + if not isinstance(state, dict): + return {"ok": True, "skipped": True, "reason": "state-missing", "items": [], "valuesPrinted": False} + accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {} + accounts = list_accounts(token) + by_name = {item.get("name"): item for item in accounts if isinstance(item.get("name"), str)} + now_epoch = time.time() + 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 + until_epoch = parse_epoch_z(quarantine.get("until")) + if until_epoch is not None and until_epoch <= now_epoch: + 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) @@ -3818,6 +4121,7 @@ def run_sync(): payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) profiles = payload.get("profiles") or [] prune_removed = bool(payload.get("pruneRemoved")) + sentinel_payload = payload.get("sentinel") if isinstance(payload.get("sentinel"), dict) else {} if not profiles: raise RuntimeError("sync payload has no profiles") admin_email, token, admin_compliance = login() @@ -3839,8 +4143,10 @@ def run_sync(): 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_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, + "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 sentinel.get("ok") is True and sentinel_reassert.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, @@ -3877,6 +4183,7 @@ def run_sync(): }, "ownerBalance": owner_balance, "ownerConcurrency": owner_concurrency, + "sentinel": {**sentinel, "freezeReassert": sentinel_reassert}, "runtimeCapabilities": runtime_capabilities, "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayResponsesRecent": responses_evidence, "gatewayCompactRecent": compact_evidence}, } @@ -3901,8 +4208,9 @@ def run_validate(): 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, + "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 sentinel.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, @@ -3922,6 +4230,7 @@ def run_validate(): "loadFactor": load_factor_status, "webSocketsV2": ws_v2_status, "tempUnschedulable": temp_unschedulable_status, + "sentinel": sentinel, "runtimeCapabilities": runtime_capabilities, "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayResponsesRecent": responses_evidence, "gatewayCompactRecent": compact_evidence}, }