fix: restore sub2api accounts from sentinel runtime state

2026-06-13 08:42:23 +00:00
parent 23d5daf388
commit ad5b6264d8
2 changed files with 139 additions and 22 deletions
@@ -834,8 +834,12 @@ class Sub2ApiAdmin:
        account = self.account(account_name)
        if not account or account.get("id") is None:
            raise RuntimeError(f"account {account_name} not found")
+        previous = account.get("schedulable")
        self.request("POST", f"/api/v1/admin/accounts/{account['id']}/schedulable", {"schedulable": bool(schedulable)})
-        return {"accountId": account.get("id"), "schedulable": bool(schedulable)}
+        account["schedulable"] = bool(schedulable)
+        if self.accounts_by_name is not None:
+            self.accounts_by_name[account_name] = account
+        return {"accountId": account.get("id"), "previousSchedulable": previous, "schedulable": bool(schedulable)}

 def upstream_base_url(base_url):
    base = str(base_url).rstrip("/")
@@ -1244,6 +1248,99 @@ def ledger_for(state, now):
 def account_day(account_state, day):
    return account_state.setdefault("daily", {}).setdefault(day, {"inputTokens": 0, "outputTokens": 0, "totalTokens": 0, "estimatedCostUsd": 0, "requestCount": 0})

+def runtime_temp_unschedulable_until(account):
+    if not isinstance(account, dict):
+        return None
+    return account.get("temp_unschedulable_until") or account.get("tempUnschedulableUntil")
+
+def runtime_temp_unschedulable_reason(account):
+    if not isinstance(account, dict):
+        return None
+    return account.get("temp_unschedulable_reason") or account.get("tempUnschedulableReason")
+
+def runtime_recovery_due(account_state):
+    return account_state.get("runtimeSchedulable") is False
+
+def sync_runtime_schedulable_state(state, profiles, now, admin):
+    accounts_state = state.setdefault("accounts", {})
+    synced_at = iso(now)
+    summary = {
+        "ok": False,
+        "syncedAt": synced_at,
+        "managedAccountCount": len(profiles),
+        "schedulableAccounts": [],
+        "unschedulableAccounts": [],
+        "missingAccounts": [],
+    }
+    try:
+        runtime_accounts = admin.accounts()
+    except Exception as exc:
+        error_text = str(exc)
+        summary["error"] = error_text
+        state["runtimeSchedulable"] = summary
+        for profile in profiles:
+            name = profile.get("accountName") if isinstance(profile, dict) else None
+            if isinstance(name, str) and name:
+                account_state = accounts_state.setdefault(name, {})
+                account_state["runtimeSchedulable"] = None
+                account_state["runtimeSyncedAt"] = synced_at
+                account_state["runtimeSyncError"] = error_text
+        return [{"type": "runtime-sync", "ok": False, "error": error_text}]
+    for profile in profiles:
+        name = profile.get("accountName") if isinstance(profile, dict) else None
+        if not isinstance(name, str) or not name:
+            continue
+        account_state = accounts_state.setdefault(name, {})
+        account = runtime_accounts.get(name)
+        if not isinstance(account, dict):
+            account_state["runtimeMissing"] = True
+            account_state["runtimeAccountId"] = None
+            account_state["runtimeStatus"] = None
+            account_state["runtimeSchedulable"] = None
+            account_state["runtimeTempUnschedulableUntil"] = None
+            account_state["runtimeTempUnschedulableSet"] = None
+            account_state["runtimeSyncedAt"] = synced_at
+            account_state.pop("runtimeSyncError", None)
+            summary["missingAccounts"].append(name)
+            continue
+        temp_until = runtime_temp_unschedulable_until(account)
+        temp_reason = runtime_temp_unschedulable_reason(account)
+        temp_set = temp_until is not None or bool(temp_reason)
+        schedulable = account.get("schedulable")
+        if not isinstance(schedulable, bool):
+            schedulable = None
+        account_state["runtimeMissing"] = False
+        account_state["runtimeAccountId"] = account.get("id")
+        account_state["runtimeStatus"] = account.get("status")
+        account_state["runtimeSchedulable"] = schedulable
+        account_state["runtimeTempUnschedulableUntil"] = temp_until
+        account_state["runtimeTempUnschedulableSet"] = temp_set
+        account_state["runtimeSyncedAt"] = synced_at
+        account_state.pop("runtimeSyncError", None)
+        if temp_reason:
+            account_state["runtimeTempUnschedulableReasonHash"] = sha(temp_reason)
+            account_state["runtimeTempUnschedulableReasonPreview"] = preview(temp_reason, 240)
+        else:
+            account_state.pop("runtimeTempUnschedulableReasonHash", None)
+            account_state.pop("runtimeTempUnschedulableReasonPreview", None)
+        if schedulable is True:
+            summary["schedulableAccounts"].append(name)
+        elif schedulable is False:
+            summary["unschedulableAccounts"].append({
+                "accountName": name,
+                "status": account.get("status"),
+                "tempUnschedulableSet": temp_set,
+            })
+    summary["ok"] = True
+    state["runtimeSchedulable"] = summary
+    return [{
+        "type": "runtime-sync",
+        "ok": True,
+        "schedulableCount": len(summary["schedulableAccounts"]),
+        "unschedulableCount": len(summary["unschedulableAccounts"]),
+        "missingCount": len(summary["missingAccounts"]),
+    }]
+
 def add_usage(state, account_state, now, usage):
    day, ledger = ledger_for(state, now)
    daily = account_day(account_state, day)
@@ -1258,6 +1355,8 @@ 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"))
+    if runtime_recovery_due(account_state):
+        return None
    return parse_iso(account_state.get("nextProbeAfter"))

 def choose_due_profiles(profiles, state, config, now):
@@ -1270,7 +1369,8 @@ def choose_due_profiles(profiles, state, config, now):
        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"
+            active_quarantine = isinstance(quarantine, dict) and quarantine.get("active") is True
+            purpose = "recovery" if active_quarantine else "runtime-recovery" if runtime_recovery_due(account_state) 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}
@@ -1291,7 +1391,8 @@ def choose_forced_profiles(profiles, state, config, now, names):
            continue
        account_state = accounts.setdefault(name, {})
        quarantine = account_state.get("quarantine")
-        purpose = "manual-recovery" if isinstance(quarantine, dict) and quarantine.get("active") is True else "manual-health"
+        active_quarantine = isinstance(quarantine, dict) and quarantine.get("active") is True
+        purpose = "manual-recovery" if active_quarantine else "manual-runtime-recovery" if runtime_recovery_due(account_state) else "manual-health"
        due.append({"profile": profile, "purpose": purpose, "dueAt": "forced"})
        found.append(name)
    missing = sorted(name for name in names if name not in set(found))
@@ -1328,16 +1429,22 @@ def apply_result(result, state, config, now, admin, profile):
    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)
+    was_runtime_recovery = runtime_recovery_due(account_state)
    action = {"taken": False, "type": None}
    if result.get("ok") is True:
        quality_gate = account_state.get("qualityGate") if isinstance(account_state.get("qualityGate"), dict) else None
-        if was_recovery:
-            if actions_enabled and quarantine.get("applied") is True:
+        if was_recovery or was_runtime_recovery:
+            if actions_enabled:
                try:
-                    action = {"taken": True, "type": "restore", "result": admin.set_schedulable(name, True)}
+                    action_type = "restore" if was_recovery else "restore-runtime-unschedulable"
+                    action = {"taken": True, "type": action_type, "result": admin.set_schedulable(name, True)}
+                    account_state["runtimeSchedulable"] = True
+                    account_state["runtimeSyncedAt"] = iso(now)
+                    account_state.pop("runtimeSyncError", None)
                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}
+            if was_recovery:
+                account_state["quarantine"] = {"active": False, "clearedAt": iso(now), "lastApplied": quarantine.get("applied") is True}
            account_state["successStreak"] = 0
            account_state["successIntervalMinutes"] = 0
        elif isinstance(quarantine, dict) and quarantine.get("active") is not True:
@@ -1361,6 +1468,8 @@ def apply_result(result, state, config, now, admin, profile):
                try:
                    action = {"taken": True, "type": "freeze", "result": admin.set_schedulable(name, False)}
                    applied = True
+                    account_state["runtimeSchedulable"] = False
+                    account_state["runtimeSyncedAt"] = iso(now)
                except Exception as exc:
                    action = {"taken": False, "type": "freeze-failed", "error": str(exc)}
            else:
@@ -1863,7 +1972,8 @@ def main():
    kube = KubeClient(namespace)
    state_obj, state = load_state(kube, config)
    admin = Sub2ApiAdmin(config)
-    reconcile = reconcile_active_quarantines(state, config, now, admin)
+    runtime_sync = sync_runtime_schedulable_state(state, profiles, now, admin)
+    reconcile = runtime_sync + reconcile_active_quarantines(state, config, now, admin)
    forced_names = forced_account_names()
    gateway_monitor = {"enabled": False, "skipped": "forced-manual-probe"} if forced_names else run_gateway_failure_monitor(state, config, now, kube, admin, profiles)
    if forced_names:
@@ -1899,6 +2009,7 @@ def main():
            "skipped": gateway_monitor.get("skipped"),
            "logErrors": gateway_monitor.get("logErrors"),
        },
+        "runtimeSchedulable": state.get("runtimeSchedulable"),
        "selection": selection,
        "reconcile": reconcile[-20:],
    }
@@ -2382,6 +2382,8 @@ function renderSentinelReport(
    `ok=${parsed.ok === true ? "true" : "false"}`,
    `accounts=${summary.accountCount ?? accounts.length}`,
    `quarantined=${summary.quarantinedCount ?? "?"}`,
+    `schedulable=${summary.runtimeSchedulableCount ?? "?"}`,
+    `unschedulable=${summary.runtimeUnschedulableCount ?? "?"}`,
    `history=${summary.historyCount ?? runs.length}`,
    `window=${formatWindow(summary.historyFrom, summary.historyTo)}`,
  ].join(" "));
@@ -2396,11 +2398,12 @@ function renderSentinelReport(
  lines.push("");
  lines.push("ACCOUNTS");
  lines.push(renderTable([
-    ["ACCOUNT", "STATE", "Q", "T", "PROT", "P_FAIL", "F_MIN", "S_MIN", "S_MAX", "PROBES", "LAST", "HTTP", "M", "KIND", "ACTION", "NEXT", "OBS_MIN"],
+    ["ACCOUNT", "STATE", "Q", "SCH", "T", "PROT", "P_FAIL", "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.runtimeSchedulable === true ? "Y" : account.runtimeSchedulable === false ? "N" : "-",
      account.trustUpstream === true ? "Y" : account.trustUpstream === false ? "N" : "-",
      account.sentinelProtectEnabled === true ? textValue(account.sentinelProtectThreshold) : "-",
      account.sentinelProtectDecision ? `${textValue(account.sentinelProtectFailureCount)}/${textValue(account.sentinelProtectThreshold)}` : "-",
@@ -2437,7 +2440,7 @@ function renderSentinelReport(
    ]));
  }
  lines.push("");
-  lines.push("LEGEND Q=quarantined T=trusted upstream PROT=sentinel protect consecutive-failure threshold P_FAIL=last protect failures/threshold M=marker matched F_MIN=freeze interval S_MIN=success interval S_MAX=success max interval OBS_MIN=last event to next probe minutes TF=transport failures GF=gateway failures GACT=gateway freeze actions");
+  lines.push("LEGEND Q=quarantined SCH=Sub2API runtime schedulable T=trusted upstream PROT=sentinel protect consecutive-failure threshold P_FAIL=last protect failures/threshold M=marker matched F_MIN=freeze interval S_MIN=success interval S_MAX=success max interval OBS_MIN=last event to next probe minutes TF=transport failures GF=gateway failures GACT=gateway freeze actions");
  lines.push("Raw: bun scripts/cli.ts platform-infra sub2api codex-pool sentinel-report --raw");
  return lines.join("\n");
 }
@@ -3665,6 +3668,14 @@ 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,
+            "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,
@@ -3710,6 +3721,8 @@ def report():
            "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)
@@ -3733,10 +3746,13 @@ def report():
        "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,
@@ -4368,13 +4384,6 @@ def account_payload(profile, group_id):
        "confirm_mixed_channel_risk": True,
    }

-def ensure_account_schedulable(token, account_id, account_name, schedulable=True):
-    data = ensure_success(
-        curl_api("POST", f"/api/v1/admin/accounts/{account_id}/schedulable", bearer=token, payload={"schedulable": bool(schedulable)}),
-        f"set account {account_name} schedulable={bool(schedulable)}",
-    )
-    return data if isinstance(data, dict) else {}
-
 def planned_sentinel_account_results(profiles, existing_accounts):
    existing = {item.get("name"): item for item in existing_accounts if isinstance(item, dict)}
    results = []
@@ -4422,10 +4431,6 @@ def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_fr
        else:
            data = ensure_success(curl_api("POST", "/api/v1/admin/accounts", bearer=token, payload=payload), f"create account {profile['accountName']}")
            action = "created"
-        if isinstance(data, dict) and data.get("id") is not None:
-            schedulable_data = ensure_account_schedulable(token, data["id"], profile["accountName"], not keep_frozen)
-            if schedulable_data:
-                data = schedulable_data
        results.append({
            "profile": profile["profile"],
            "accountName": profile["accountName"],
@@ -4445,6 +4450,7 @@ def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_fr
            "sentinelProbePending": quality_gate_required,
            "sentinelDefaultFrozen": False,
            "sentinelFreezeProtected": keep_frozen,
+            "schedulableControl": "sentinel-marker-restore",
            "openaiResponsesWebSocketsV2Mode": profile.get("openaiResponsesWebSocketsV2Mode"),
            "trustUpstream": profile.get("trustUpstream") is True,
            "priority": int(profile.get("priority", POOL_DEFAULT_ACCOUNT_PRIORITY) or POOL_DEFAULT_ACCOUNT_PRIORITY),
@@ -5936,7 +5942,7 @@ def run_sync():
            "pruneMode": "explicit" if prune_removed else "disabled-by-default",
            "items": account_results,
            "prunedItems": pruned_account_results,
-            "processControl": {"schedulableRestore": "POST /api/v1/admin/accounts/:id/schedulable for non-quarantined accounts only", "durableConfig": False},
+            "processControl": {"schedulableRestore": "sentinel marker probe only; sync does not restore schedulable for existing accounts", "durableConfig": False},
            "valuesPrinted": False,
        },
        "capacity": capacity_status,