Merge pull request #2002 from pikasTech/feat/sub2api-runtime-batch-reconcile

feat: Sub2API runtime 批量写后自动对账

config/platform-infra/sub2api-codex-pool.yaml

@@ -12,22 +12,38 @@ runtime:
   templates:
     - id: codex-upstream-failover
       kind: temp-unschedulable
-      description: Reusable Sub2API request-path cooling and failover policy for Codex-compatible API-key accounts.
+      description: Codex 兼容 API-key 账号的请求路径短时冷却与换号策略。
       spec:
         enabled: true
         rules:
+          - statusCode: 429
+            keywords: [concurrency limit exceeded, upstream rate limit exceeded]
+            durationMinutes: 1
+            description: 明确命中上游并发或速率限制时短时冷却当前账号。
+          - statusCode: 400
+            keywords: [input must be a list]
+            durationMinutes: 1
+            description: 明确命中 Responses input 列表兼容错误时短时冷却并切换账号。
+          - statusCode: 500
+            keywords: [failed to validate api key, do request failed]
+            durationMinutes: 1
+            description: 上游 API-key 查询依赖或请求传输内部失败时短时冷却当前账号。
           - statusCode: 502
-            keywords: [upstream service temporarily unavailable, upstream request failed, context window, stream usage incomplete, missing terminal event]
+            keywords: [upstream service temporarily unavailable, upstream request failed, service temporarily unavailable, overloaded, concurrency limit exceeded]
             durationMinutes: 1
-            description: Observed 502 upstream availability, context-window, and incomplete-stream failures briefly cool down the selected account.
+            description: 明确命中 502 上游不可用、过载或并发限制时短时冷却当前账号。
           - statusCode: 503
-            keywords: [upstream service temporarily unavailable, upstream request failed, temporarily unavailable]
+            keywords: [upstream service temporarily unavailable, upstream request failed, service temporarily unavailable, temporarily unavailable, overloaded, concurrency limit exceeded]
             durationMinutes: 1
-            description: Observed 503 temporary upstream availability failures briefly cool down the selected account.
+            description: 明确命中 503 上游暂时不可用、过载或并发限制时短时冷却当前账号。
+          - statusCode: 504
+            keywords: [concurrency limit exceeded, upstream service temporarily unavailable, input must be a list]
+            durationMinutes: 1
+            description: 明确命中 504 上游并发限制、暂时不可用或 Responses input 列表兼容错误时短时冷却当前账号。
           - statusCode: 524
-            keywords: [upstream request failed, missing_required_parameter, encrypted_content]
+            keywords: [upstream request failed, service temporarily unavailable, overloaded, concurrency limit exceeded]
             durationMinutes: 1
-            description: Observed 524 upstream request and encrypted-content failures briefly cool down the selected account.
+            description: 明确命中 524 上游不可用、过载或并发限制时短时冷却当前账号。
 
 pool:
   groupName: unidesk-codex-pool

scripts/src/platform-infra-sub2api-codex/runtime.ts

@@ -1,17 +1,20 @@
 import type { UniDeskConfig } from "../config";
+import type { RenderedCliResult } from "../output";
 
 import { desiredAccountNames } from "./accounts";
 import { readCodexPoolConfig } from "./config";
+import { isRecord, stringValue } from "./config-utils";
 import { protectedManualAccountsForTarget } from "./public-exposure";
 import { renderSub2ApiTempUnschedulableCredentials } from "./redaction";
 import { boolField, compactCapture, parseJsonOutput, runRemoteCodexPoolScript } from "./remote";
+import { renderTable } from "./render";
 import { codexPoolRuntimeTarget, defaultCodexPoolRuntimeTargetId } from "./runtime-target";
 
 type RuntimeAction = "list" | "get" | "errors" | "apply" | "delete";
 
 interface RuntimeOptions {
   action: RuntimeAction;
-  account: string | null;
+  accounts: string[];
   template: string | null;
   kind: "temp-unschedulable";
   confirm: boolean;
@@ -22,7 +25,7 @@ interface RuntimeOptions {
   targetId: string;
 }
 
-export async function codexPoolRuntime(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown>> {
+export async function codexPoolRuntime(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown> | RenderedCliResult> {
   const options = parseRuntimeOptions(args);
   const pool = readCodexPoolConfig();
   const target = codexPoolRuntimeTarget(options.targetId);
@@ -32,7 +35,8 @@ export async function codexPoolRuntime(config: UniDeskConfig, args: string[]): P
   }
   const payload = {
     action: options.action,
-    account: options.account,
+    account: options.accounts.length === 1 ? options.accounts[0] : null,
+    accounts: options.accounts,
     kind: options.kind,
     confirm: options.confirm,
     full: options.full,
@@ -57,6 +61,9 @@ export async function codexPoolRuntime(config: UniDeskConfig, args: string[]): P
   const result = await runRemoteCodexPoolScript(config, "runtime", runtimeScript(payload, target), target);
   const parsed = parseJsonOutput(result.stdout);
   const ok = result.exitCode === 0 && boolField(parsed, "ok", false);
+  if (!options.full && !options.raw && (options.action === "apply" || options.action === "delete")) {
+    return renderRuntimeMutation(ok, options, parsed, result.exitCode);
+  }
   if (options.raw) {
     return {
       ok,
@@ -78,7 +85,8 @@ export async function codexPoolRuntime(config: UniDeskConfig, args: string[]): P
     },
     request: {
       operation: options.action,
-      account: options.account,
+      account: options.accounts.length === 1 ? options.accounts[0] : undefined,
+      accounts: options.accounts.length > 1 ? options.accounts : undefined,
       template: options.template,
       kind: options.kind,
       confirm: options.confirm,
@@ -100,9 +108,10 @@ export async function codexPoolRuntime(config: UniDeskConfig, args: string[]): P
 export function parseRuntimeOptions(args: string[]): RuntimeOptions {
   const [actionRaw = "list", ...rest] = args;
   if (actionRaw !== "list" && actionRaw !== "get" && actionRaw !== "errors" && actionRaw !== "apply" && actionRaw !== "delete") {
-    throw new Error("runtime usage: list|get|errors|apply|delete [--account <name-or-id>] [--since 24h] [--tail 50000] [--template <id>] [--kind temp-unschedulable] [--confirm] [--target <id>] [--full|--raw]");
+    throw new Error("runtime usage: list|get|errors|apply|delete [--account <name-or-id>|--accounts <selector,...>] [--since 24h] [--tail 50000] [--template <id>] [--kind temp-unschedulable] [--confirm] [--target <id>] [--full|--raw]");
   }
   let account: string | null = null;
+  let accounts: string[] | null = null;
   let template: string | null = null;
   let kind = "temp-unschedulable" as const;
   let confirm = false;
@@ -124,6 +133,8 @@ export function parseRuntimeOptions(args: string[]): RuntimeOptions {
     else if (arg === "--raw") raw = true;
     else if (arg === "--account") account = readValue("--account");
     else if (arg.startsWith("--account=")) account = arg.slice("--account=".length).trim();
+    else if (arg === "--accounts") accounts = parseRuntimeAccounts(readValue("--accounts"));
+    else if (arg.startsWith("--accounts=")) accounts = parseRuntimeAccounts(arg.slice("--accounts=".length));
     else if (arg === "--template") template = readValue("--template");
     else if (arg.startsWith("--template=")) template = arg.slice("--template=".length).trim();
     else if (arg === "--kind") kind = parseRuntimeKind(readValue("--kind"));
@@ -136,17 +147,27 @@ export function parseRuntimeOptions(args: string[]): RuntimeOptions {
     else if (arg.startsWith("--tail=")) tail = parseRuntimeTail(arg.slice("--tail=".length));
     else throw new Error(`unsupported runtime option: ${arg}`);
   }
-  if ((actionRaw === "get" || actionRaw === "apply" || actionRaw === "delete") && !account) {
+  if (account !== null && accounts !== null) throw new Error("use only one of --account or --accounts");
+  const selectors = accounts ?? (account === null ? [] : [account]);
+  if ((actionRaw === "get" || actionRaw === "apply" || actionRaw === "delete") && selectors.length === 0) {
     throw new Error(`runtime ${actionRaw} requires --account <name-or-id>`);
   }
-  if (account !== null && (account.length > 256 || /[\r\n]/u.test(account))) throw new Error("--account has an unsupported format");
+  if (actionRaw !== "apply" && actionRaw !== "delete" && accounts !== null) throw new Error(`runtime ${actionRaw} does not accept --accounts`);
+  if (selectors.some((selector) => selector.length > 256 || /[\r\n]/u.test(selector))) throw new Error("account selector has an unsupported format");
   if (actionRaw === "apply" && !template) throw new Error("runtime apply requires --template <id>");
   if (actionRaw !== "apply" && template !== null) throw new Error(`runtime ${actionRaw} does not accept --template`);
   if (actionRaw !== "errors" && (since !== "24h" || tail !== 50_000)) throw new Error(`runtime ${actionRaw} does not accept --since or --tail`);
   if (!/^\d+[mhd]$/u.test(since)) throw new Error("--since must use <number>m, <number>h, or <number>d");
   if ((actionRaw === "list" || actionRaw === "get" || actionRaw === "errors") && confirm) throw new Error(`runtime ${actionRaw} does not accept --confirm`);
   if (full && raw) throw new Error("use only one of --full or --raw");
-  return { action: actionRaw, account, template, kind, confirm, full, raw, since, tail, targetId };
+  return { action: actionRaw, accounts: selectors, template, kind, confirm, full, raw, since, tail, targetId };
+}
+
+function parseRuntimeAccounts(value: string): string[] {
+  const accounts = value.split(",").map((item) => item.trim());
+  if (accounts.length === 0 || accounts.some((item) => item.length === 0)) throw new Error("--accounts requires comma-separated selectors");
+  if (new Set(accounts).size !== accounts.length) throw new Error("--accounts contains duplicate selectors");
+  return accounts;
 }
 
 function parseRuntimeKind(value: string): "temp-unschedulable" {
@@ -160,6 +181,60 @@ function parseRuntimeTail(value: string): number {
   return parsed;
 }
 
+function renderRuntimeMutation(ok: boolean, options: RuntimeOptions, parsed: Record<string, unknown> | null, exitCode: number): RenderedCliResult {
+  const summary = parsed !== null && isRecord(parsed.summary) ? parsed.summary : {};
+  const items = parsed !== null && Array.isArray(parsed.items) ? parsed.items.filter(isRecord) : [];
+  const lines = [[
+    "SUB2API RUNTIME",
+    `operation=${options.action}`,
+    `mode=${stringValue(parsed?.mode) ?? (options.confirm ? "confirmed" : "dry-run")}`,
+    `ok=${ok ? "true" : "false"}`,
+    `selected=${summary.selected ?? items.length}`,
+    `changed=${summary.changed ?? "?"}`,
+    `writeSucceeded=${summary.writeSucceeded ?? "?"}`,
+    `writeFailed=${summary.writeFailed ?? "?"}`,
+    `reconciled=${summary.reconciled ?? "?"}`,
+    `mismatched=${summary.mismatched ?? "?"}`,
+  ].join(" ")];
+  if (items.length > 0) {
+    lines.push(renderTable([
+      ["ACCOUNT", "ID", "CHANGE", "WRITE", "RECONCILED", "BEFORE", "DESIRED", "ACTUAL", "BEFORE_CODES", "DESIRED_CODES", "ACTUAL_CODES", "FAILURE"],
+      ...items.map((item) => {
+        const write = isRecord(item.write) ? item.write : {};
+        const before = isRecord(item.before) ? item.before : {};
+        const desired = isRecord(item.desired) ? item.desired : {};
+        const actual = isRecord(item.actual) ? item.actual : {};
+        const codes = (value: unknown): string => Array.isArray(value) ? value.join(",") || "-" : "-";
+        return [
+          stringValue(item.accountName) ?? "-",
+          String(item.accountId ?? "-"),
+          stringValue(item.change) ?? "-",
+          stringValue(write.status) ?? "-",
+          item.reconciled === true ? "true" : item.reconciled === false ? "false" : "-",
+          String(before.ruleCount ?? "-"),
+          String(desired.ruleCount ?? "-"),
+          String(actual.ruleCount ?? "-"),
+          codes(before.statusCodes),
+          codes(desired.statusCodes),
+          codes(actual.statusCodes),
+          stringValue(write.error) ?? stringValue(item.reconcileError) ?? "-",
+        ];
+      }),
+    ]));
+  } else if (parsed !== null && typeof parsed.error === "string") {
+    lines.push(`ERROR ${parsed.error}`);
+  } else if (parsed === null) {
+    lines.push(`ERROR runtime output unavailable remote_exit=${exitCode}`);
+  }
+  return {
+    ok,
+    command: "platform-infra sub2api codex-pool runtime",
+    renderedText: lines.join("\n"),
+    contentType: "text/plain",
+    projection: parsed ?? undefined,
+  };
+}
+
 function pyJson(value: unknown): string {
   return `json.loads(${JSON.stringify(JSON.stringify(value))})`;
 }
@@ -487,6 +562,78 @@ def observed_runtime_errors(accounts):
         "message": "observedStatusCodes comes only from account_upstream_error; forward failures are supplemental message evidence",
     }
 
+def selected_account_details(token, accounts):
+    selectors = PAYLOAD.get("accounts")
+    if not isinstance(selectors, list) or not selectors:
+        selector = PAYLOAD.get("account")
+        selectors = [selector] if isinstance(selector, str) and selector else []
+    details = []
+    selected_ids = set()
+    for raw_selector in selectors:
+        selector = str(raw_selector)
+        detail = find_account(token, accounts, selector)
+        account_id = detail.get("id")
+        if account_id in selected_ids:
+            raise RuntimeError("duplicate-account-selector: " + selector)
+        if detail.get("type") != "apikey":
+            raise RuntimeError("runtime temp-unschedulable policy requires an apikey account: " + selector)
+        selected_ids.add(account_id)
+        details.append(detail)
+    return details
+
+def mutation_plan(detail, action, include_rules):
+    credentials = dict(detail.get("credentials") or {})
+    before = normalize_policy(credentials)
+    if action == "apply":
+        template = PAYLOAD.get("selectedTemplate")
+        if not isinstance(template, dict):
+            raise RuntimeError("selected runtime template missing")
+        desired = dict(credentials)
+        desired.update(template.get("credentials") or {})
+        desired_policy = normalize_policy(desired)
+        exists = "temp_unschedulable_enabled" in credentials or "temp_unschedulable_rules" in credentials
+        change = "noop" if before == desired_policy else ("update" if exists else "create")
+    else:
+        desired = dict(credentials)
+        desired.pop("temp_unschedulable_enabled", None)
+        desired.pop("temp_unschedulable_rules", None)
+        desired_policy = normalize_policy(desired)
+        change = "delete" if before != desired_policy or "temp_unschedulable_enabled" in credentials or "temp_unschedulable_rules" in credentials else "noop"
+    return {
+        "accountName": detail.get("name"),
+        "accountId": detail.get("id"),
+        "change": change,
+        "before": policy_summary(before, include_rules),
+        "desired": policy_summary(desired_policy, include_rules),
+        "template": PAYLOAD.get("selectedTemplate", {}).get("id") if isinstance(PAYLOAD.get("selectedTemplate"), dict) else None,
+        "_desiredCredentials": desired,
+        "_desiredPolicy": desired_policy,
+    }
+
+def write_mutation_plan(token, plan):
+    if plan["change"] == "noop":
+        return {"attempted": False, "status": "noop", "error": None}
+    try:
+        data_of(curl_api("PUT", f"/api/v1/admin/accounts/{plan['accountId']}", bearer=token, payload={"credentials": plan["_desiredCredentials"]}), "update account runtime config")
+        return {"attempted": True, "status": "succeeded", "error": None}
+    except Exception as exc:
+        return {"attempted": True, "status": "failed", "error": compact_error(str(exc))}
+
+def reconcile_mutation_plan(token, plan, write, include_rules):
+    item = {key: value for key, value in plan.items() if not key.startswith("_")}
+    item["write"] = write
+    try:
+        actual_detail = account_detail(token, plan["accountId"])
+        actual_policy = normalize_policy(actual_detail.get("credentials"))
+        item["actual"] = policy_summary(actual_policy, include_rules)
+        item["reconciled"] = actual_policy == plan["_desiredPolicy"]
+        item["reconcileError"] = None
+    except Exception as exc:
+        item["actual"] = None
+        item["reconciled"] = False
+        item["reconcileError"] = compact_error(str(exc))
+    return item
+
 def runtime_result():
     token = login()
     group = next((item for item in list_groups(token) if item.get("name") == PAYLOAD["groupName"]), None)
@@ -506,43 +653,33 @@ def runtime_result():
     if action == "errors":
         errors = observed_runtime_errors(accounts)
         return {**base, "ok": True, "operation": "errors", "mutation": False, "errors": errors}
-    detail = find_account(token, accounts, str(PAYLOAD["account"]))
+    detail = find_account(token, accounts, str(PAYLOAD["account"])) if action == "get" else None
     if action == "get":
         return {**base, "ok": True, "operation": "get", "mutation": False, "account": account_summary(detail, True)}
-    if detail.get("type") != "apikey":
-        raise RuntimeError("runtime temp-unschedulable policy requires an apikey account")
-    credentials = dict(detail.get("credentials") or {})
-    before = normalize_policy(credentials)
-    if action == "apply":
-        template = PAYLOAD.get("selectedTemplate")
-        if not isinstance(template, dict):
-            raise RuntimeError("selected runtime template missing")
-        desired_fields = template.get("credentials") or {}
-        desired = dict(credentials)
-        desired.update(desired_fields)
-        after_plan = normalize_policy(desired)
-        exists = "temp_unschedulable_enabled" in credentials or "temp_unschedulable_rules" in credentials
-        change = "noop" if before == after_plan else ("update" if exists else "create")
-    else:
-        desired = dict(credentials)
-        desired.pop("temp_unschedulable_enabled", None)
-        desired.pop("temp_unschedulable_rules", None)
-        after_plan = normalize_policy(desired)
-        change = "delete" if before != after_plan or "temp_unschedulable_enabled" in credentials or "temp_unschedulable_rules" in credentials else "noop"
     include_plan_rules = PAYLOAD.get("full") is True
-    plan = {
-        "change": change,
-        "account": account_summary(detail, include_plan_rules),
-        "before": policy_summary(before, include_plan_rules),
-        "after": policy_summary(after_plan, include_plan_rules),
-        "template": PAYLOAD.get("selectedTemplate", {}).get("id") if isinstance(PAYLOAD.get("selectedTemplate"), dict) else None,
-        "credentialsPrinted": False,
-    }
-    if PAYLOAD.get("confirm") is not True or change == "noop":
-        return {**base, "ok": True, "operation": action, "mode": "dry-run" if PAYLOAD.get("confirm") is not True else "confirmed-noop", "mutation": False, "plan": plan}
-    updated = data_of(curl_api("PUT", f"/api/v1/admin/accounts/{detail['id']}", bearer=token, payload={"credentials": desired}), "update account runtime config")
-    after_detail = updated if isinstance(updated, dict) else account_detail(token, detail["id"])
-    return {**base, "ok": True, "operation": action, "mode": "confirmed", "mutation": True, "plan": plan, "account": account_summary(after_detail, include_plan_rules)}
+    details = selected_account_details(token, accounts)
+    plans = [mutation_plan(item, action, include_plan_rules) for item in details]
+    changed = sum(1 for plan in plans if plan["change"] != "noop")
+    if PAYLOAD.get("confirm") is not True:
+        items = []
+        for plan in plans:
+            item = {key: value for key, value in plan.items() if not key.startswith("_")}
+            item["write"] = {"attempted": False, "status": "dry-run", "error": None}
+            item["actual"] = item["before"]
+            item["reconciled"] = None
+            item["reconcileError"] = None
+            items.append(item)
+        summary = {"selected": len(items), "changed": changed, "writeSucceeded": 0, "writeFailed": 0, "reconciled": 0, "mismatched": 0}
+        return {**base, "ok": True, "operation": action, "mode": "dry-run", "mutation": False, "partialWrite": False, "summary": summary, "items": items, "credentialsPrinted": False}
+    writes = [write_mutation_plan(token, plan) for plan in plans]
+    items = [reconcile_mutation_plan(token, plan, write, include_plan_rules) for plan, write in zip(plans, writes)]
+    write_succeeded = sum(1 for item in items if item["write"]["status"] == "succeeded")
+    write_failed = sum(1 for item in items if item["write"]["status"] == "failed")
+    reconciled = sum(1 for item in items if item["reconciled"] is True)
+    mismatched = sum(1 for item in items if item["reconciled"] is False)
+    summary = {"selected": len(items), "changed": changed, "writeSucceeded": write_succeeded, "writeFailed": write_failed, "reconciled": reconciled, "mismatched": mismatched}
+    ok = write_failed == 0 and mismatched == 0
+    return {**base, "ok": ok, "operation": action, "mode": "confirmed", "mutation": write_succeeded > 0, "partialWrite": write_succeeded > 0 and write_failed > 0, "summary": summary, "items": items, "credentialsPrinted": False}
 
 try:
     result = runtime_result()

scripts/src/platform-infra-sub2api-codex/types.ts

@@ -362,7 +362,9 @@ export function codexPoolHelp(): unknown {
       "bun scripts/cli.ts platform-infra sub2api codex-pool runtime errors [--account <name-or-id>] [--since 24h] [--tail 50000] [--target PK01] [--full|--raw]",
       "bun scripts/cli.ts platform-infra sub2api codex-pool faults [--level P0|P1|P2] [--group <name-or-id>] [--account <name-or-id>] [--model <model>] [--stream sync|stream] [--endpoint <path>] [--request-id <id>] [--page-token <token>] [--target PK01] [--json]",
       "bun scripts/cli.ts platform-infra sub2api codex-pool runtime apply --account <name-or-id> --template <id> [--target PK01] [--confirm]",
+      "bun scripts/cli.ts platform-infra sub2api codex-pool runtime apply --accounts <name-or-id,...> --template <id> [--target PK01] [--confirm] [--full|--raw]",
       "bun scripts/cli.ts platform-infra sub2api codex-pool runtime delete --account <name-or-id> --kind temp-unschedulable [--target PK01] [--confirm]",
+      "bun scripts/cli.ts platform-infra sub2api codex-pool runtime delete --accounts <name-or-id,...> --kind temp-unschedulable [--target PK01] [--confirm] [--full|--raw]",
       "bun scripts/cli.ts platform-infra sub2api codex-pool trace [--target D601] --request-id <requestId> [--since 24h|--tail 20000|--context-seconds 300|--show-lines|--raw]",
       "bun scripts/cli.ts platform-infra sub2api codex-pool feedback --user <email-or-id> [--window 30m] [--page-token <token>|--request-id <client-or-internal-id>] [--target PK01] [--json]",
       "bun scripts/cli.ts platform-infra sub2api codex-pool sentinel-image status [--target D601]",