feat: add sub2api sentinel trust cadence

2026-06-11 13:55:25 +00:00
parent ea92eed148
commit d475cf2a9e
4 changed files with 159 additions and 15 deletions
@@ -76,6 +76,7 @@ interface CodexProfile {
  apiKeySource: "auth-json" | "env" | null;
  openaiResponsesWebSocketsV2Mode: OpenAIResponsesWebSocketsV2Mode | null;
  upstreamUserAgent: string | null;
+  trustUpstream: boolean;
  priority: number;
  capacity: number;
  loadFactor: number;
@@ -126,6 +127,7 @@ interface CodexPoolProfileConfig {
  fallbackAuthFile: string | null;
  openaiResponsesWebSocketsV2Mode: OpenAIResponsesWebSocketsV2Mode | null;
  upstreamUserAgent: string | null;
+  trustUpstream: boolean;
  priority: number;
  capacity: number | null;
  loadFactor: number | null;
@@ -480,9 +482,11 @@ async function codexPoolSync(config: UniDeskConfig, options: SyncOptions): Promi
        apiKeyFingerprint: fingerprint(profile.apiKey ?? ""),
        upstreamUserAgent: profile.upstreamUserAgent,
        openaiResponsesWebSocketsV2Mode: profile.openaiResponsesWebSocketsV2Mode,
+        trustUpstream: profile.trustUpstream,
      }),
      openaiResponsesWebSocketsV2Mode: profile.openaiResponsesWebSocketsV2Mode,
      upstreamUserAgent: profile.upstreamUserAgent,
+      trustUpstream: profile.trustUpstream,
      priority: profile.priority,
      capacity: profile.capacity,
      loadFactor: profile.loadFactor,
@@ -830,6 +834,7 @@ function collectCodexProfiles(): CodexProfile[] {
      apiKeySource: null,
      openaiResponsesWebSocketsV2Mode: entry.openaiResponsesWebSocketsV2Mode,
      upstreamUserAgent: entry.upstreamUserAgent,
+      trustUpstream: entry.trustUpstream,
      priority: entry.priority,
      capacity: entry.capacity ?? pool.defaultAccountCapacity,
      loadFactor: entry.loadFactor ?? pool.defaultAccountLoadFactor,
@@ -902,6 +907,7 @@ function discoverCodexProfileConfigs(
        fallbackAuthFile: null,
        openaiResponsesWebSocketsV2Mode: null,
        upstreamUserAgent: null,
+        trustUpstream: false,
        priority: defaultPriority,
        capacity: null,
        loadFactor: null,
@@ -1135,6 +1141,7 @@ function readProfileConfig(
    if (fallbackAuthFile !== null) validateCodexFileName(fallbackAuthFile, `profiles.entries[${index}].fallbackAuthFile`);
    const openaiResponsesWebSocketsV2Mode = readOpenAIResponsesWebSocketsV2Mode(entry.openaiResponsesWebSocketsV2Mode, `profiles.entries[${index}].openaiResponsesWebSocketsV2Mode`);
    const upstreamUserAgent = readUpstreamUserAgent(entry.upstreamUserAgent, `profiles.entries[${index}].upstreamUserAgent`);
+    const trustUpstream = readTrustUpstream(entry.trustUpstream, `profiles.entries[${index}].trustUpstream`);
    const priority = readAccountPriority(entry.priority, `profiles.entries[${index}].priority`, defaultPriority);
    const capacity = entry.capacity === undefined || entry.capacity === null ? null : readAccountCapacity(entry.capacity, `profiles.entries[${index}].capacity`);
    const loadFactor = entry.loadFactor === undefined || entry.loadFactor === null ? null : readAccountLoadFactor(entry.loadFactor, `profiles.entries[${index}].loadFactor`);
@@ -1148,6 +1155,7 @@ function readProfileConfig(
      fallbackAuthFile,
      openaiResponsesWebSocketsV2Mode,
      upstreamUserAgent,
+      trustUpstream,
      priority,
      capacity,
      loadFactor,
@@ -1183,6 +1191,13 @@ function readUpstreamUserAgent(value: unknown, key: string): string | null {
  return text;
 }

+function readTrustUpstream(value: unknown, key: string): boolean {
+  if (value === undefined || value === null) return false;
+  const parsed = booleanValue(value);
+  if (parsed === null) throw new Error(`${codexPoolConfigPath}.${key} must be a boolean`);
+  return parsed;
+}
+
 function readAccountPriority(value: unknown, key: string, fallback = defaultAccountPriority): number {
  if (value === undefined || value === null) return fallback;
  const priority = numberValue(value);
@@ -1502,6 +1517,7 @@ function redactProfile(profile: CodexProfile): Record<string, unknown> {
    apiKeySource: profile.apiKeySource,
    openaiResponsesWebSocketsV2Mode: profile.openaiResponsesWebSocketsV2Mode,
    upstreamUserAgent: profile.upstreamUserAgent,
+    trustUpstream: profile.trustUpstream,
    priority: profile.priority,
    capacity: profile.capacity,
    loadFactor: profile.loadFactor,
@@ -1523,6 +1539,7 @@ function compactProfile(profile: CodexProfile): Record<string, unknown> {
    provider: profile.provider || null,
    model: profile.model,
    priority: profile.priority,
+    trustUpstream: profile.trustUpstream,
    capacity: profile.capacity,
    loadFactor: profile.loadFactor,
    tempUnschedulableEnabled: profile.tempUnschedulable.enabled && profile.tempUnschedulable.rules.length > 0,
@@ -1996,13 +2013,15 @@ function renderSentinelReport(
  lines.push("");
  lines.push("ACCOUNTS");
  lines.push(renderTable([
-    ["ACCOUNT", "STATE", "Q", "F_MIN", "S_MIN", "PROBES", "LAST", "HTTP", "M", "KIND", "ACTION", "NEXT", "OBS_MIN"],
+    ["ACCOUNT", "STATE", "Q", "T", "F_MIN", "S_MIN", "S_MAX", "PROBES", "LAST", "HTTP", "M", "KIND", "ACTION", "NEXT", "OBS_MIN"],
    ...accounts.map((account) => [
      stringValue(account.account) ?? "-",
      stringValue(account.status) ?? "-",
      account.quarantineActive === true ? "Y" : "-",
+      account.trustUpstream === true ? "Y" : account.trustUpstream === false ? "N" : "-",
      textValue(account.freezeIntervalMin),
      textValue(account.successIntervalMin),
+      textValue(account.successMaxIntervalMin),
      textValue(account.probeCount),
      shortIso(account.lastProbeAt),
      textValue(account.lastHttp),
@@ -2031,7 +2050,7 @@ function renderSentinelReport(
    ]));
  }
  lines.push("");
-  lines.push("LEGEND Q=quarantined M=marker matched F_MIN=freeze interval S_MIN=success interval OBS_MIN=last probe to next probe minutes TF=transport failures");
+  lines.push("LEGEND Q=quarantined T=trusted upstream M=marker matched F_MIN=freeze interval S_MIN=success interval S_MAX=success max interval OBS_MIN=last probe to next probe minutes TF=transport failures");
  lines.push("Raw: bun scripts/cli.ts platform-infra sub2api codex-pool sentinel-report --raw");
  return lines.join("\n");
 }
@@ -2209,6 +2228,7 @@ function sentinelProfileSecrets(profiles: CodexProfile[]): CodexPoolSentinelProf
      baseUrl: profile.baseUrl,
      apiKey: profile.apiKey ?? "",
      upstreamUserAgent: profile.upstreamUserAgent,
+      trustUpstream: profile.trustUpstream,
    }));
 }

@@ -2883,6 +2903,7 @@ export function codexPoolSentinelProbeConfigFingerprint(input: {
  apiKeyFingerprint: string | null;
  upstreamUserAgent: string | null;
  openaiResponsesWebSocketsV2Mode: string | null;
+  trustUpstream: boolean;
 }): string {
  return fingerprint(JSON.stringify({
    accountName: input.accountName,
@@ -2891,6 +2912,7 @@ export function codexPoolSentinelProbeConfigFingerprint(input: {
    apiKeyFingerprint: input.apiKeyFingerprint,
    upstreamUserAgent: input.upstreamUserAgent,
    openaiResponsesWebSocketsV2Mode: input.openaiResponsesWebSocketsV2Mode,
+    trustUpstream: input.trustUpstream,
  }));
 }

@@ -3027,8 +3049,10 @@ def report():
            "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,
+            "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),
@@ -3445,6 +3469,8 @@ def sentinel_probe_change_reasons(current, profile):
    runtime_user_agent = empty_to_none(credentials.get("user_agent"))
    expected_ws_mode = empty_to_none(profile.get("openaiResponsesWebSocketsV2Mode"))
    runtime_ws_mode = empty_to_none(extra.get("openai_apikey_responses_websockets_v2_mode"))
+    expected_trust_upstream = profile.get("trustUpstream") is True
+    runtime_trust_upstream = extra.get("unidesk_trust_upstream") is True
    reasons = []
    if empty_to_none(extra.get("unidesk_codex_profile")) != profile.get("profile"):
        reasons.append("profile")
@@ -3456,6 +3482,8 @@ def sentinel_probe_change_reasons(current, profile):
        reasons.append("upstream-user-agent")
    if runtime_ws_mode != expected_ws_mode:
        reasons.append("responses-websockets-v2-mode")
+    if runtime_trust_upstream != expected_trust_upstream:
+        reasons.append("trust-upstream")
    return reasons

 def curl_api(method, path, bearer=None, payload=None):
@@ -3669,6 +3697,7 @@ def account_payload(profile, group_id):
        "openai_responses_mode": "force_responses",
        "unidesk_codex_profile": profile["profile"],
        "unidesk_managed": True,
+        "unidesk_trust_upstream": profile.get("trustUpstream") is True,
    }
    ws_mode = profile.get("openaiResponsesWebSocketsV2Mode")
    if ws_mode:
@@ -3715,6 +3744,10 @@ def planned_sentinel_account_results(profiles, existing_accounts):
        results.append({
            "profile": profile["profile"],
            "accountName": profile["accountName"],
+            "profileConfig": {
+                "accountName": profile["accountName"],
+                "trustUpstream": profile.get("trustUpstream") is True,
+            },
            "sentinelProbeConfigFingerprint": profile.get("sentinelProbeConfigFingerprint"),
            "sentinelProbeRequired": quality_gate_required,
            "sentinelChangeReasons": change_reasons if quality_gate_required else [],
@@ -3754,6 +3787,10 @@ def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_fr
        results.append({
            "profile": profile["profile"],
            "accountName": profile["accountName"],
+            "profileConfig": {
+                "accountName": profile["accountName"],
+                "trustUpstream": profile.get("trustUpstream") is True,
+            },
            "accountId": data.get("id") if isinstance(data, dict) else None,
            "action": action,
            "baseUrl": profile["baseUrl"],
@@ -3765,6 +3802,7 @@ def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_fr
            "sentinelDefaultFrozen": quality_gate_required,
            "sentinelFreezeProtected": keep_frozen,
            "openaiResponsesWebSocketsV2Mode": profile.get("openaiResponsesWebSocketsV2Mode"),
+            "trustUpstream": profile.get("trustUpstream") is True,
            "priority": int(profile.get("priority", POOL_DEFAULT_ACCOUNT_PRIORITY) or POOL_DEFAULT_ACCOUNT_PRIORITY),
            "capacity": int(profile.get("capacity", 5) or 5),
            "loadFactor": int(profile.get("loadFactor", POOL_DEFAULT_ACCOUNT_LOAD_FACTOR) or POOL_DEFAULT_ACCOUNT_LOAD_FACTOR),
@@ -4061,6 +4099,68 @@ def clamp_sentinel_freezes_for_config(state, now):
        })
    return items

+def parse_iso_epoch(value):
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
+    except Exception:
+        return None
+
+def profile_success_max_interval(profile):
+    cadence = SENTINEL_CONFIG.get("cadence") if isinstance(SENTINEL_CONFIG.get("cadence"), dict) else {}
+    legacy = cadence.get("successMaxIntervalMinutes")
+    if legacy is None:
+        legacy = cadence.get("trustedSuccessMaxIntervalMinutes") or cadence.get("untrustedSuccessMaxIntervalMinutes") or 1
+    if profile.get("trustUpstream") is True:
+        value = cadence.get("trustedSuccessMaxIntervalMinutes") or legacy
+    else:
+        value = cadence.get("untrustedSuccessMaxIntervalMinutes") or legacy
+    try:
+        return int(value)
+    except Exception:
+        return int(legacy)
+
+def clamp_sentinel_success_cadence_for_config(state, profiles, now):
+    accounts_state = state.get("accounts") if isinstance(state.get("accounts"), dict) else {}
+    profile_map = {item.get("accountName"): item for item in profiles if isinstance(item, dict) and isinstance(item.get("accountName"), str)}
+    now_epoch = time.time()
+    items = []
+    for name, profile in profile_map.items():
+        account_state = accounts_state.get(name)
+        if not isinstance(account_state, dict):
+            continue
+        quarantine = account_state.get("quarantine")
+        if isinstance(quarantine, dict) and quarantine.get("active") is True:
+            account_state["trustUpstream"] = profile.get("trustUpstream") is True
+            account_state["successMaxIntervalMinutes"] = profile_success_max_interval(profile)
+            continue
+        try:
+            interval = int(account_state.get("successIntervalMinutes") or 0)
+        except Exception:
+            interval = 0
+        next_epoch = parse_iso_epoch(account_state.get("nextProbeAfter"))
+        max_interval = profile_success_max_interval(profile)
+        account_state["trustUpstream"] = profile.get("trustUpstream") is True
+        account_state["successMaxIntervalMinutes"] = max_interval
+        if interval <= max_interval and (next_epoch is None or next_epoch <= now_epoch + max_interval * 60):
+            continue
+        old_next = account_state.get("nextProbeAfter")
+        account_state["previousSuccessIntervalMinutes"] = interval
+        account_state["successIntervalMinutes"] = min(interval, max_interval) if interval > 0 else interval
+        account_state["nextProbeAfter"] = now
+        account_state["cadenceClampedAt"] = now
+        account_state["cadenceClampedBy"] = "sync-success-max-interval"
+        items.append({
+            "accountName": name,
+            "trustUpstream": profile.get("trustUpstream") is True,
+            "previousSuccessIntervalMinutes": interval,
+            "maxIntervalMinutes": max_interval,
+            "previousNextProbeAfter": old_next,
+            "nextProbeAfter": now,
+        })
+    return items
+
 def update_sentinel_state_configmap(obj, state):
    state_name = SENTINEL_CONFIG.get("stateConfigMapName")
    if not state_name:
@@ -4101,6 +4201,7 @@ def ensure_sentinel_state_for_sync(account_results, pending_only=False):
    pending_until = utc_iso(3600)
    items = []
    clamped_items = [] if pending_only else clamp_sentinel_freezes_for_config(state, now)
+    cadence_clamped_items = [] if pending_only else clamp_sentinel_success_cadence_for_config(state, [item.get("profileConfig") for item in account_results if isinstance(item.get("profileConfig"), dict)], now)
    changed_count = 0
    fingerprint_only_count = 0
    for item in account_results:
@@ -4133,6 +4234,9 @@ def ensure_sentinel_state_for_sync(account_results, pending_only=False):
        account_state["nextProbeAfter"] = pending_until if pending_only else now
        account_state["successStreak"] = 0
        account_state["successIntervalMinutes"] = 0
+        profile_config = item.get("profileConfig") if isinstance(item.get("profileConfig"), dict) else {}
+        account_state["trustUpstream"] = profile_config.get("trustUpstream") is True
+        account_state["successMaxIntervalMinutes"] = profile_success_max_interval(profile_config)
        account_state["lastStatus"] = "pending-sentinel-quality-gate"
        account_state["qualityGate"] = {
            "pending": True,
@@ -4142,15 +4246,17 @@ def ensure_sentinel_state_for_sync(account_results, pending_only=False):
            "pendingOnly": pending_only,
        }
        items.append({"accountName": name, "changeReasons": reasons, "nextProbeAfter": pending_until if pending_only else now, "defaultFrozen": True, "pendingOnly": pending_only})
-    if changed_count <= 0 and len(clamped_items) <= 0:
-        return {"ok": True, "skipped": False, "reason": "no-new-or-changed-accounts", "changedCount": 0, "fingerprintOnlyCount": fingerprint_only_count, "clampedCount": 0, "items": [], "valuesPrinted": False}
+    if changed_count <= 0 and len(clamped_items) <= 0 and len(cadence_clamped_items) <= 0:
+        return {"ok": True, "skipped": False, "reason": "no-new-or-changed-accounts", "changedCount": 0, "fingerprintOnlyCount": fingerprint_only_count, "clampedCount": 0, "cadenceClampedCount": 0, "items": [], "valuesPrinted": False}
    update = update_sentinel_state_configmap(state_obj, state)
    if pending_only and changed_count > 0:
        reason = "new-or-changed-accounts-pending-quality-gate-prepared"
-    elif changed_count > 0 and len(clamped_items) > 0:
-        reason = "new-or-changed-accounts-default-frozen-and-freeze-backoff-clamped"
+    elif changed_count > 0 and (len(clamped_items) > 0 or len(cadence_clamped_items) > 0):
+        reason = "new-or-changed-accounts-default-frozen-and-sentinel-cadence-clamped"
    elif changed_count > 0:
        reason = "new-or-changed-accounts-default-frozen"
+    elif len(cadence_clamped_items) > 0:
+        reason = "success-cadence-clamped-to-current-config"
    else:
        reason = "freeze-backoff-clamped-to-current-config"
    return {
@@ -4160,9 +4266,11 @@ def ensure_sentinel_state_for_sync(account_results, pending_only=False):
        "changedCount": changed_count,
        "fingerprintOnlyCount": fingerprint_only_count,
        "clampedCount": len(clamped_items),
+        "cadenceClampedCount": len(cadence_clamped_items),
        "pendingOnly": pending_only,
        "items": items,
        "clampedItems": clamped_items,
+        "cadenceClampedItems": cadence_clamped_items,
        "update": update,
        "valuesPrinted": False,
    }