Codex
435 lines
20 KiB
TypeScript
435 lines
20 KiB
TypeScript
// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. remote-scripts module for scripts/src/platform-infra-sub2api-codex.ts.
|
|
|
|
// Moved mechanically from scripts/src/platform-infra-sub2api-codex.ts:3800-4208 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, TraceOptions } from "./types";
|
|
import { shQuote } from "./remote";
|
|
import { remotePythonScript } from "./remote-python-sync";
|
|
import { sentinelImageDockerfilePath } from "./types";
|
|
|
|
export function syncScript(payload: unknown, pool: CodexPoolConfig, target: CodexPoolRuntimeTarget): string {
|
|
const encoded = Buffer.from(JSON.stringify(payload), "utf8").toString("base64");
|
|
return remotePythonScript("sync", encoded, pool, target);
|
|
}
|
|
|
|
export function validateScript(pool: CodexPoolConfig, target: CodexPoolRuntimeTarget): string {
|
|
return remotePythonScript("validate", "", pool, target);
|
|
}
|
|
|
|
export function traceScript(pool: CodexPoolConfig, options: TraceOptions, target: CodexPoolRuntimeTarget): string {
|
|
const encoded = Buffer.from(JSON.stringify({
|
|
requestId: options.requestId,
|
|
since: options.since,
|
|
tail: options.tail,
|
|
contextSeconds: options.contextSeconds,
|
|
showLines: options.showLines,
|
|
}), "utf8").toString("base64");
|
|
return remotePythonScript("trace", encoded, pool, target);
|
|
}
|
|
|
|
export function sentinelProbeScript(payload: unknown, pool: CodexPoolConfig, target: CodexPoolRuntimeTarget): string {
|
|
const encoded = Buffer.from(JSON.stringify(payload), "utf8").toString("base64");
|
|
return remotePythonScript("sentinel-probe", encoded, pool, target);
|
|
}
|
|
|
|
export function sentinelReportScript(pool: CodexPoolConfig, events: number, target: CodexPoolRuntimeTarget): string {
|
|
const stateName = pool.sentinel.stateConfigMapName;
|
|
const cronJobName = pool.sentinel.cronJobName;
|
|
return `
|
|
set -eu
|
|
python3 - <<'PY'
|
|
import json
|
|
import subprocess
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
NAMESPACE = ${JSON.stringify(target.namespace)}
|
|
STATE_NAME = ${JSON.stringify(stateName)}
|
|
CRONJOB_NAME = ${JSON.stringify(cronJobName)}
|
|
EVENT_LIMIT = ${JSON.stringify(events)}
|
|
|
|
def run(cmd):
|
|
return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
|
|
def text(data, limit=2000):
|
|
if isinstance(data, bytes):
|
|
data = data.decode("utf-8", errors="replace")
|
|
return data[-limit:]
|
|
|
|
def kube_json(args):
|
|
proc = run(["kubectl", *args, "-o", "json"])
|
|
if proc.returncode != 0:
|
|
return None, text(proc.stderr)
|
|
try:
|
|
return json.loads(proc.stdout.decode("utf-8")), None
|
|
except Exception as exc:
|
|
return None, str(exc)
|
|
|
|
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 minutes_between(a, b):
|
|
left = parse_iso(a)
|
|
right = parse_iso(b)
|
|
if left is None or right is None:
|
|
return None
|
|
return round((right - left).total_seconds() / 60, 1)
|
|
|
|
def day_ledgers(state):
|
|
ledger = {}
|
|
raw = state.get("ledger")
|
|
if isinstance(raw, dict):
|
|
for item in raw.values():
|
|
if not isinstance(item, dict):
|
|
continue
|
|
add_ledger(ledger, item)
|
|
return ledger
|
|
|
|
def add_ledger(target, source):
|
|
target["inputTokens"] = target.get("inputTokens", 0) + int(source.get("inputTokens") or 0)
|
|
target["outputTokens"] = target.get("outputTokens", 0) + int(source.get("outputTokens") or 0)
|
|
target["totalTokens"] = target.get("totalTokens", 0) + int(source.get("totalTokens") or 0)
|
|
target["estimatedCostUsd"] = target.get("estimatedCostUsd", 0.0) + float(source.get("estimatedCostUsd") or 0)
|
|
target["requestCount"] = target.get("requestCount", 0) + int(source.get("requestCount") or 0)
|
|
|
|
def account_ledger(account_state):
|
|
total = {}
|
|
raw = account_state.get("daily")
|
|
if isinstance(raw, dict):
|
|
for item in raw.values():
|
|
if isinstance(item, dict):
|
|
add_ledger(total, item)
|
|
return total
|
|
|
|
def action_type(probe):
|
|
action = probe.get("action") if isinstance(probe, dict) else None
|
|
if isinstance(action, dict):
|
|
value = action.get("type")
|
|
if value:
|
|
return value
|
|
return "taken" if action.get("taken") is True else ""
|
|
return ""
|
|
|
|
def error_code(probe):
|
|
details = probe.get("errorDetails") if isinstance(probe, dict) else None
|
|
if not isinstance(details, dict):
|
|
return ""
|
|
openai_error = details.get("openaiError")
|
|
if isinstance(openai_error, dict):
|
|
return openai_error.get("code") or openai_error.get("type") or ""
|
|
return details.get("kind") or ""
|
|
|
|
def sentinel_protect_summary(account_state, probe):
|
|
probe_protect = probe.get("sentinelProtect") if isinstance(probe.get("sentinelProtect"), dict) else {}
|
|
config_protect = account_state.get("sentinelProtectConfig") if isinstance(account_state.get("sentinelProtectConfig"), dict) else {}
|
|
source = probe_protect if probe_protect else config_protect
|
|
if not isinstance(source, dict) or source.get("enabled") is not True:
|
|
return {
|
|
"enabled": False,
|
|
"threshold": None,
|
|
"decision": None,
|
|
"failureCount": None,
|
|
}
|
|
return {
|
|
"enabled": True,
|
|
"threshold": source.get("threshold") or source.get("consecutiveFailures"),
|
|
"decision": probe_protect.get("decision"),
|
|
"failureCount": probe_protect.get("failureCount"),
|
|
}
|
|
|
|
def report():
|
|
cronjob, cron_error = kube_json(["-n", NAMESPACE, "get", "cronjob", CRONJOB_NAME])
|
|
state_cm, state_error = kube_json(["-n", NAMESPACE, "get", "configmap", STATE_NAME])
|
|
state = {}
|
|
parse_error = None
|
|
if isinstance(state_cm, dict):
|
|
raw_state = (state_cm.get("data") or {}).get("state.json")
|
|
if isinstance(raw_state, str) and raw_state.strip():
|
|
try:
|
|
state = json.loads(raw_state)
|
|
except Exception as exc:
|
|
parse_error = str(exc)
|
|
accounts = state.get("accounts") if isinstance(state.get("accounts"), dict) else {}
|
|
history = state.get("history") if isinstance(state.get("history"), list) else []
|
|
account_rows = []
|
|
for name, account_state in sorted(accounts.items()):
|
|
if not isinstance(account_state, dict):
|
|
continue
|
|
probe = account_state.get("lastProbe") if isinstance(account_state.get("lastProbe"), dict) else {}
|
|
quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else {}
|
|
gateway_failure = account_state.get("lastGatewayFailure") if isinstance(account_state.get("lastGatewayFailure"), dict) else {}
|
|
last_event_at = account_state.get("lastGatewayFailureAt") or account_state.get("lastProbeAt")
|
|
last_failure_kind = quarantine.get("failureKind") if quarantine.get("active") is True and quarantine.get("failureKind") else probe.get("failureKind")
|
|
last_action = action_type(probe)
|
|
if gateway_failure and (not account_state.get("lastProbeAt") or str(account_state.get("lastGatewayFailureAt") or "") >= str(account_state.get("lastProbeAt") or "")):
|
|
last_action = action_type(gateway_failure)
|
|
ledger = account_ledger(account_state)
|
|
protect = sentinel_protect_summary(account_state, probe)
|
|
account_rows.append({
|
|
"account": name,
|
|
"status": account_state.get("lastStatus"),
|
|
"quarantineActive": quarantine.get("active") is True,
|
|
"quarantineApplied": quarantine.get("applied") if isinstance(quarantine, dict) else None,
|
|
"freezeIntervalMin": quarantine.get("intervalMinutes") if isinstance(quarantine, dict) else None,
|
|
"freezeUntil": quarantine.get("until") if isinstance(quarantine, dict) else None,
|
|
"runtimeAccountId": account_state.get("runtimeAccountId"),
|
|
"runtimeStatus": account_state.get("runtimeStatus"),
|
|
"runtimeSchedulable": account_state.get("runtimeSchedulable"),
|
|
"runtimeMissing": account_state.get("runtimeMissing"),
|
|
"runtimeTempUnschedulableSet": account_state.get("runtimeTempUnschedulableSet"),
|
|
"runtimeTempUnschedulableUntil": account_state.get("runtimeTempUnschedulableUntil"),
|
|
"runtimeSyncedAt": account_state.get("runtimeSyncedAt"),
|
|
"runtimeSyncError": account_state.get("runtimeSyncError"),
|
|
"trustUpstream": account_state.get("trustUpstream") if account_state.get("trustUpstream") is not None else probe.get("trustUpstream"),
|
|
"successStreak": account_state.get("successStreak") or 0,
|
|
"successIntervalMin": account_state.get("successIntervalMinutes") or 0,
|
|
"successMaxIntervalMin": account_state.get("successMaxIntervalMinutes") or probe.get("successMaxIntervalMinutes"),
|
|
"probeCount": ledger.get("requestCount", 0),
|
|
"inputTokens": ledger.get("inputTokens", 0),
|
|
"outputTokens": ledger.get("outputTokens", 0),
|
|
"totalTokens": ledger.get("totalTokens", 0),
|
|
"estimatedCostUsd": round(float(ledger.get("estimatedCostUsd", 0)), 6),
|
|
"lastProbeAt": account_state.get("lastProbeAt"),
|
|
"lastEventAt": last_event_at,
|
|
"lastGatewayFailureAt": account_state.get("lastGatewayFailureAt"),
|
|
"lastPurpose": probe.get("purpose"),
|
|
"lastHttp": probe.get("httpStatus"),
|
|
"lastMarker": probe.get("markerMatched"),
|
|
"sentinelProtectEnabled": protect.get("enabled"),
|
|
"sentinelProtectDecision": protect.get("decision"),
|
|
"sentinelProtectFailureCount": protect.get("failureCount"),
|
|
"sentinelProtectThreshold": protect.get("threshold"),
|
|
"lastFailureKind": last_failure_kind,
|
|
"lastErrorCode": error_code(probe),
|
|
"lastAction": last_action,
|
|
"nextProbeAfter": account_state.get("nextProbeAfter"),
|
|
"observedLastToNextMin": minutes_between(last_event_at, account_state.get("nextProbeAfter")),
|
|
"requestShape": probe.get("requestShape"),
|
|
})
|
|
run_rows = []
|
|
for item in history:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
selection = item.get("selection") if isinstance(item.get("selection"), dict) else {}
|
|
gateway = item.get("gatewayFailureMonitor") if isinstance(item.get("gatewayFailureMonitor"), dict) else {}
|
|
run_rows.append({
|
|
"at": item.get("at"),
|
|
"selected": item.get("selected"),
|
|
"due": selection.get("due"),
|
|
"ok": item.get("okCount"),
|
|
"mismatch": item.get("mismatchCount") if item.get("mismatchCount") is not None else item.get("markerMismatchCount"),
|
|
"transportFailures": item.get("transportFailureCount"),
|
|
"actionsTaken": item.get("actionsTaken"),
|
|
"gatewayFailures": gateway.get("newFailures"),
|
|
"gatewayActions": gateway.get("actionsTaken"),
|
|
"reasserts": len(item.get("reconcile") or []),
|
|
})
|
|
quarantined = [item for item in account_rows if item.get("quarantineActive") is True]
|
|
runtime_schedulable = [item for item in account_rows if item.get("runtimeSchedulable") is True]
|
|
runtime_unschedulable = [item for item in account_rows if item.get("runtimeSchedulable") is False]
|
|
cron_spec = cronjob.get("spec") if isinstance(cronjob, dict) else {}
|
|
cron_status = cronjob.get("status") if isinstance(cronjob, dict) else {}
|
|
global_ledger = day_ledgers(state)
|
|
result = {
|
|
"ok": state_error is None and parse_error is None,
|
|
"metadata": {
|
|
"namespace": NAMESPACE,
|
|
"stateConfigMapName": STATE_NAME,
|
|
"cronJobName": CRONJOB_NAME,
|
|
"generatedAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"valuesPrinted": False,
|
|
},
|
|
"cronJob": {
|
|
"exists": isinstance(cronjob, dict),
|
|
"schedule": cron_spec.get("schedule") if isinstance(cron_spec, dict) else None,
|
|
"suspend": cron_spec.get("suspend") if isinstance(cron_spec, dict) else None,
|
|
"lastScheduleTime": cron_status.get("lastScheduleTime") if isinstance(cron_status, dict) else None,
|
|
"active": len(cron_status.get("active") or []) if isinstance(cron_status, dict) else None,
|
|
"error": cron_error,
|
|
},
|
|
"summary": {
|
|
"accountCount": len(account_rows),
|
|
"quarantinedCount": len(quarantined),
|
|
"runtimeSchedulableCount": len(runtime_schedulable),
|
|
"runtimeUnschedulableCount": len(runtime_unschedulable),
|
|
"historyCount": len(history),
|
|
"historyFrom": run_rows[0].get("at") if run_rows else None,
|
|
"historyTo": run_rows[-1].get("at") if run_rows else None,
|
|
"lastRun": state.get("lastRun"),
|
|
"runtimeSchedulable": state.get("runtimeSchedulable"),
|
|
},
|
|
"globalLedger": global_ledger,
|
|
"accounts": account_rows,
|
|
"runs": run_rows[-EVENT_LIMIT:],
|
|
"errors": {
|
|
"state": state_error,
|
|
"parse": parse_error,
|
|
},
|
|
"valuesPrinted": False,
|
|
}
|
|
return result
|
|
|
|
payload = report()
|
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
raise SystemExit(0 if payload.get("ok") else 1)
|
|
PY
|
|
`;
|
|
}
|
|
|
|
export function cleanupProbesScript(pool: CodexPoolConfig, target: CodexPoolRuntimeTarget): string {
|
|
return remotePythonScript("cleanup-probes", "", pool, target);
|
|
}
|
|
|
|
export function sentinelImageStatusScript(pool: CodexPoolConfig, targetRuntime: CodexPoolRuntimeTarget): string {
|
|
const target = codexPoolSentinelRuntimeImage(pool.sentinel);
|
|
return remoteSentinelImageScript("status", target, pool.sentinel, null, targetRuntime);
|
|
}
|
|
|
|
export function sentinelImageBuildScript(pool: CodexPoolConfig, targetRuntime: CodexPoolRuntimeTarget): string {
|
|
const target = codexPoolSentinelRuntimeImage(pool.sentinel);
|
|
const dockerfile = readFileSync(sentinelImageDockerfilePath, "utf8");
|
|
return remoteSentinelImageScript("build", target, pool.sentinel, dockerfile, targetRuntime);
|
|
}
|
|
|
|
export function remoteSentinelImageScript(mode: "status" | "build", target: ReturnType<typeof codexPoolSentinelRuntimeImage>, sentinel: CodexPoolSentinelConfig, dockerfile: string | null, targetRuntime: CodexPoolRuntimeTarget): string {
|
|
const dockerfileB64 = dockerfile === null ? "" : Buffer.from(dockerfile, "utf8").toString("base64");
|
|
return `
|
|
set -eu
|
|
mode=${shQuote(mode)}
|
|
image=${shQuote(target.runtimeImage)}
|
|
repo=${shQuote("platform-infra/sub2api-account-sentinel")}
|
|
tag=${shQuote(target.tag)}
|
|
base_image=${shQuote(target.baseImage)}
|
|
openai_version=${shQuote(sentinel.sdk.openaiPythonVersion)}
|
|
base_image_cache_policy=${shQuote(targetRuntime.sentinelImageBuild.baseImageCachePolicy)}
|
|
proxy_env_path=${shQuote(targetRuntime.sentinelImageBuild.proxyEnvPath ?? "")}
|
|
work=/tmp/unidesk-sub2api-sentinel-image
|
|
mkdir -p "$work"
|
|
dockerfile_path="$work/sentinel.Dockerfile"
|
|
registry_has_tag=false
|
|
if curl -fsS --max-time 10 "http://127.0.0.1:5000/v2/$repo/tags/list" 2>/dev/null | grep -F '"'"$tag"'"' >/dev/null 2>&1; then
|
|
registry_has_tag=true
|
|
fi
|
|
local_id="$(docker image inspect "$image" --format '{{.Id}}' 2>/dev/null || true)"
|
|
if [ "$mode" = "status" ]; then
|
|
if [ -n "$local_id" ]; then
|
|
python_version="$(docker run --rm "$image" python3 --version 2>/dev/null || true)"
|
|
openai_runtime_version="$(docker run --rm "$image" python3 -c 'import importlib.metadata; print(importlib.metadata.version("openai"))' 2>/dev/null || true)"
|
|
else
|
|
python_version=
|
|
openai_runtime_version=
|
|
fi
|
|
python3 - <<PY
|
|
import json
|
|
print(json.dumps({
|
|
"ok": bool("${"${registry_has_tag}"}" == "true" or "${"${local_id}"}"),
|
|
"mode": "status",
|
|
"image": "${target.runtimeImage}",
|
|
"baseImage": "${target.baseImage}",
|
|
"tag": "${target.tag}",
|
|
"localExists": bool("${"${local_id}"}"),
|
|
"localImageId": "${"${local_id}"}" or None,
|
|
"registryHasTag": "${"${registry_has_tag}"}" == "true",
|
|
"pythonVersion": "${"${python_version}"}" or None,
|
|
"openaiPythonVersion": "${"${openai_runtime_version}"}" or None,
|
|
}, ensure_ascii=False, indent=2))
|
|
PY
|
|
exit 0
|
|
fi
|
|
if [ "$registry_has_tag" = "true" ]; then
|
|
if [ -z "$local_id" ]; then
|
|
docker pull "$image" >/dev/null 2>&1 || true
|
|
local_id="$(docker image inspect "$image" --format '{{.Id}}' 2>/dev/null || true)"
|
|
fi
|
|
python3 - <<PY
|
|
import json
|
|
print(json.dumps({
|
|
"ok": True,
|
|
"mode": "reused-existing",
|
|
"image": "${target.runtimeImage}",
|
|
"baseImage": "${target.baseImage}",
|
|
"tag": "${target.tag}",
|
|
"registryHasTag": True,
|
|
"localImageId": "${"${local_id}"}" or None,
|
|
}, ensure_ascii=False, indent=2))
|
|
PY
|
|
exit 0
|
|
fi
|
|
base64 -d > "$dockerfile_path" <<'UNIDESK_SENTINEL_DOCKERFILE_B64'
|
|
${dockerfileB64}
|
|
UNIDESK_SENTINEL_DOCKERFILE_B64
|
|
if [ -n "$proxy_env_path" ] && [ -r "$proxy_env_path" ]; then
|
|
set -a
|
|
. "$proxy_env_path"
|
|
set +a
|
|
fi
|
|
HTTP_PROXY="$(printenv HTTP_PROXY 2>/dev/null || true)"
|
|
HTTPS_PROXY="$(printenv HTTPS_PROXY 2>/dev/null || true)"
|
|
ALL_PROXY="$(printenv ALL_PROXY 2>/dev/null || true)"
|
|
http_proxy_value="$(printenv http_proxy 2>/dev/null || true)"
|
|
https_proxy_value="$(printenv https_proxy 2>/dev/null || true)"
|
|
all_proxy_value="$(printenv all_proxy 2>/dev/null || true)"
|
|
if [ -z "$HTTP_PROXY" ]; then HTTP_PROXY="$http_proxy_value"; fi
|
|
if [ -z "$HTTPS_PROXY" ]; then HTTPS_PROXY="$https_proxy_value"; fi
|
|
if [ -z "$ALL_PROXY" ]; then ALL_PROXY="$all_proxy_value"; fi
|
|
if [ -z "$HTTPS_PROXY" ]; then HTTPS_PROXY="$HTTP_PROXY"; fi
|
|
if [ -z "$ALL_PROXY" ]; then ALL_PROXY="$HTTP_PROXY"; fi
|
|
export NO_PROXY=${shQuote(targetRuntime.sentinelImageBuild.noProxy)}
|
|
export no_proxy=$NO_PROXY
|
|
set -- --pull
|
|
base_image_source="registry"
|
|
if [ "$base_image_cache_policy" = "local-if-present" ] && docker image inspect "$base_image" >/dev/null 2>&1; then
|
|
set --
|
|
base_image_source="local-cache"
|
|
fi
|
|
docker build "$@" \\
|
|
--network host \\
|
|
--build-arg BASE_IMAGE="$base_image" \\
|
|
--build-arg OPENAI_PYTHON_VERSION="$openai_version" \\
|
|
--build-arg HTTP_PROXY="$HTTP_PROXY" --build-arg HTTPS_PROXY="$HTTPS_PROXY" --build-arg ALL_PROXY="$ALL_PROXY" \\
|
|
--build-arg http_proxy="$HTTP_PROXY" --build-arg https_proxy="$HTTPS_PROXY" --build-arg all_proxy="$ALL_PROXY" \\
|
|
--build-arg NO_PROXY --build-arg no_proxy \\
|
|
-f "$dockerfile_path" \\
|
|
-t "$image" \\
|
|
"$work"
|
|
docker run --rm "$image" python3 -c 'import importlib.metadata, sys; expected=sys.argv[1]; actual=importlib.metadata.version("openai"); assert actual == expected, (actual, expected); print("openai", actual)' "$openai_version"
|
|
docker push "$image"
|
|
digest="$(docker image inspect "$image" --format '{{index .RepoDigests 0}}' 2>/dev/null || true)"
|
|
python3 - <<PY
|
|
import json
|
|
print(json.dumps({
|
|
"ok": True,
|
|
"mode": "built-and-published",
|
|
"image": "${target.runtimeImage}",
|
|
"baseImage": "${target.baseImage}",
|
|
"tag": "${target.tag}",
|
|
"baseImageSource": "${"${base_image_source}"}",
|
|
"digest": "${"${digest}"}" or None,
|
|
}, ensure_ascii=False, indent=2))
|
|
PY
|
|
`;
|
|
}
|