Merge pull request #277 from pikasTech/fix/sub2api-runtime-failover-249

fix: 收敛 Sub2API Codex pool failover 与 sentinel 边界
2026-06-12 04:11:55 +08:00
parent 440e3a931d 5b8b3e50d3
commit fa277991ea
5 changed files with 176 additions and 355 deletions
@@ -804,16 +804,6 @@ class Sub2ApiAdmin:
        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.account(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_base_url(base_url):
    base = str(base_url).rstrip("/")
    return base if base.endswith("/v1") else base + "/v1"
@@ -1228,18 +1218,20 @@ def apply_result(result, state, config, now, admin, profile):
    was_recovery = bool(quarantine and quarantine.get("active") is True)
    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:
                try:
-                    action = {"taken": True, "type": "restore", "result": admin.set_schedulable(name, True), "recoverState": admin.recover_state(name)}
+                    action = {"taken": True, "type": "restore", "result": admin.set_schedulable(name, True)}
                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}
-            quality_gate = account_state.get("qualityGate") if isinstance(account_state.get("qualityGate"), dict) else None
-            if quality_gate and quality_gate.get("pending") is True:
-                account_state["qualityGate"] = {**quality_gate, "pending": False, "clearedAt": iso(now)}
            account_state["successStreak"] = 0
            account_state["successIntervalMinutes"] = 0
+        elif isinstance(quarantine, dict) and quarantine.get("active") is not True:
+            account_state["quarantine"] = {"active": False, "clearedAt": iso(now), "lastApplied": quarantine.get("applied") is True}
+        if quality_gate and quality_gate.get("pending") is True:
+            account_state["qualityGate"] = {**quality_gate, "pending": False, "clearedAt": iso(now)}
        interval = next_success_interval(account_state, config, profile)
        account_state["successStreak"] = int(account_state.get("successStreak") or 0) + 1
        account_state["successIntervalMinutes"] = interval
@@ -1573,10 +1573,6 @@ function compactTempUnschedulableStatus(block: unknown): Record<string, unknown>
    const result = pickSummaryFields(item, [
      "accountName",
      "accountId",
-      "expectedEnabled",
-      "runtimeEnabled",
-      "expectedRuleCount",
-      "runtimeRuleCount",
      "status",
      "schedulable",
      "tempUnschedulableUntil",
@@ -1602,8 +1598,17 @@ function compactTempUnschedulableStatus(block: unknown): Record<string, unknown>
    mismatched: block.mismatched,
    itemCount: compactItems.length,
    frozenCount: frozenItems.length,
-    frozen: focusedFrozenItems,
-    manuallyUnschedulable: compactItems.filter((item) => item.schedulable === false && item.tempUnschedulableSet !== true),
+    frozenShown: focusedFrozenItems.length,
+    frozen: focusedFrozenItems.map((item) => pickSummaryFields(item, [
+      "accountName",
+      "accountId",
+      "schedulable",
+      "tempUnschedulableUntil",
+      "tempUnschedulableReason",
+    ])),
+    manuallyUnschedulable: compactItems
+      .filter((item) => item.schedulable === false && item.tempUnschedulableSet !== true)
+      .map((item) => pickSummaryFields(item, ["accountName", "accountId", "status", "schedulable"])),
    valuesPrinted: false,
  };
 }
@@ -1636,7 +1641,7 @@ function compactRuntimeCapability(block: unknown): unknown {
  if (!isRecord(block)) return block;
  const probe = isRecord(block.probe) ? block.probe : {};
  const logEvidence = isRecord(probe.logEvidence) ? probe.logEvidence : {};
-  const accountState = isRecord(probe.accountState) ? probe.accountState : {};
+  const accountState = isRecord(probe.accountState) ? probe.accountState : (isRecord(probe.badAccountState) ? probe.badAccountState : {});
  const resources = isRecord(block.resources) ? block.resources : {};
  const requirement = isRecord(block.requirement) ? block.requirement : {};
  return {
@@ -1646,11 +1651,11 @@ function compactRuntimeCapability(block: unknown): unknown {
    outcome: block.outcome,
    requirement: Object.keys(requirement).length === 0 ? undefined : {
      statusCode: requirement.statusCode,
-      probeKeyword: requirement.probeKeyword,
+      representativeKeyword: requirement.representativeKeyword,
      durationMinutes: requirement.durationMinutes,
      sourceAccountName: requirement.sourceAccountName,
    },
-    probe: Object.keys(probe).length === 0 ? undefined : {
+    requestEvidence: Object.keys(probe).length === 0 ? undefined : {
      requestId: probe.requestId,
      durationMs: probe.durationMs,
      httpStatus: probe.httpStatus,
@@ -1766,6 +1771,78 @@ function compactGatewayCompactRecent(block: unknown): unknown {
  };
 }

+function compactSentinelErrorDetails(value: unknown): Record<string, unknown> | undefined {
+  if (!isRecord(value)) return undefined;
+  const body = isRecord(value.body) ? value.body : {};
+  const openaiError = isRecord(value.openaiError) ? value.openaiError : {};
+  return {
+    kind: value.kind,
+    statusCode: value.statusCode,
+    code: value.code ?? body.code ?? openaiError.code,
+    type: value.type ?? body.type ?? openaiError.type,
+    bodyHash: value.bodyHash,
+    valuesPrinted: false,
+  };
+}
+
+function compactSentinelQuarantine(item: unknown): Record<string, unknown> {
+  if (!isRecord(item)) return {};
+  return {
+    accountName: item.accountName,
+    until: item.until,
+    applied: item.applied,
+    reason: item.reason,
+    failureKind: item.failureKind,
+    intervalMinutes: item.intervalMinutes,
+    error: compactSentinelErrorDetails(item.errorDetails),
+  };
+}
+
+function compactSentinelRecentAccount(item: unknown): Record<string, unknown> {
+  if (!isRecord(item)) return {};
+  const action = isRecord(item.action) ? item.action : {};
+  const error = compactSentinelErrorDetails(item.errorDetails);
+  return {
+    accountName: item.accountName,
+    lastProbeAt: item.lastProbeAt,
+    lastStatus: item.lastStatus,
+    nextProbeAfter: item.nextProbeAfter,
+    ok: item.ok,
+    purpose: item.purpose,
+    httpStatus: item.httpStatus,
+    durationMs: item.durationMs,
+    markerMatched: item.markerMatched,
+    failureKind: item.failureKind,
+    requestShape: item.requestShape,
+    action: Object.keys(action).length > 0 ? {
+      taken: action.taken,
+      type: action.type,
+    } : undefined,
+    errorStatusCode: error?.statusCode,
+    errorCode: error?.code,
+    errorBodyHash: error?.bodyHash,
+  };
+}
+
+function compactSentinelLastRun(value: unknown): Record<string, unknown> | undefined {
+  if (!isRecord(value)) return undefined;
+  return {
+    at: value.at,
+    monitorEnabled: value.monitorEnabled,
+    actionsEnabled: value.actionsEnabled,
+    profileCount: value.profileCount,
+    selected: value.selected,
+    okCount: value.okCount,
+    mismatchCount: value.mismatchCount,
+    markerMismatchCount: value.markerMismatchCount,
+    transportFailureCount: value.transportFailureCount,
+    actionsTaken: value.actionsTaken,
+    gatewayFailureMonitor: value.gatewayFailureMonitor,
+    selection: value.selection,
+    reconcileCount: Array.isArray(value.reconcile) ? value.reconcile.length : undefined,
+  };
+}
+
 function compactSentinelStatus(block: unknown): unknown {
  if (!isRecord(block)) return block;
  const runtime = isRecord(block.runtime) ? block.runtime : block;
@@ -1777,6 +1854,13 @@ function compactSentinelStatus(block: unknown): unknown {
  const freezeReassert = isRecord(block.freezeReassert) ? block.freezeReassert : {};
  const qualityGatePrepare = isRecord(block.qualityGatePrepare) ? block.qualityGatePrepare : {};
  const qualityGate = isRecord(block.qualityGate) ? block.qualityGate : {};
+  const quarantined = recordArray(state.quarantined).map(compactSentinelQuarantine);
+  const recentAccounts = recordArray(state.recentAccounts).map(compactSentinelRecentAccount);
+  const recentAttention = recentAccounts.filter((item) => item.ok === false || isRecord(item.action) && item.action.taken === true);
+  const recentHealthy = recentAccounts
+    .filter((item) => item.ok === true)
+    .slice(-3)
+    .map((item) => pickSummaryFields(item, ["accountName", "lastProbeAt", "lastStatus", "nextProbeAfter", "ok", "httpStatus", "markerMatched"]));
  return {
    ok: block.ok,
    action: block.action,
@@ -1813,9 +1897,12 @@ function compactSentinelStatus(block: unknown): unknown {
      exists: state.exists,
      accountCount: state.accountCount,
      quarantinedCount: state.quarantinedCount,
-      quarantined: state.quarantined,
-      recentAccounts: state.recentAccounts,
-      lastRun: state.lastRun,
+      quarantinedShown: Math.min(quarantined.length, 3),
+      quarantined: quarantined.slice(-3),
+      recentAccountCount: recentAccounts.length,
+      recentAttention: uniqueByAccountName(recentAttention.slice(-3)),
+      recentHealthy,
+      lastRun: compactSentinelLastRun(state.lastRun),
      error: state.error,
    },
    freezeReassert: Object.keys(freezeReassert).length > 0 ? {
@@ -1825,7 +1912,7 @@ function compactSentinelStatus(block: unknown): unknown {
      itemCount: freezeReassert.itemCount,
      attentionItems: freezeReassert.attentionItems,
    } : undefined,
-    qualityGatePrepare: Object.keys(qualityGatePrepare).length > 0 ? {
+    pendingProbePrepare: Object.keys(qualityGatePrepare).length > 0 ? {
      ok: qualityGatePrepare.ok,
      skipped: qualityGatePrepare.skipped,
      reason: qualityGatePrepare.reason,
@@ -1834,7 +1921,7 @@ function compactSentinelStatus(block: unknown): unknown {
      pendingOnly: qualityGatePrepare.pendingOnly,
      items: qualityGatePrepare.items,
    } : undefined,
-    qualityGate: Object.keys(qualityGate).length > 0 ? {
+    pendingProbe: Object.keys(qualityGate).length > 0 ? {
      ok: qualityGate.ok,
      skipped: qualityGate.skipped,
      reason: qualityGate.reason,
@@ -3698,7 +3785,8 @@ def planned_sentinel_account_results(profiles, existing_accounts):
            "sentinelProbeConfigFingerprint": profile.get("sentinelProbeConfigFingerprint"),
            "sentinelProbeRequired": quality_gate_required,
            "sentinelChangeReasons": change_reasons if quality_gate_required else [],
-            "sentinelDefaultFrozen": quality_gate_required,
+            "sentinelProbePending": quality_gate_required,
+            "sentinelDefaultFrozen": False,
            "valuesPrinted": False,
        })
    return results
@@ -3728,7 +3816,7 @@ def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_fr
            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 quality_gate_required and not keep_frozen)
+            schedulable_data = ensure_account_schedulable(token, data["id"], profile["accountName"], not keep_frozen)
            if schedulable_data:
                data = schedulable_data
        results.append({
@@ -3746,7 +3834,8 @@ def ensure_accounts(token, profiles, group_id, prune_removed=False, protected_fr
            "sentinelProbeConfigFingerprint": profile.get("sentinelProbeConfigFingerprint"),
            "sentinelProbeRequired": quality_gate_required,
            "sentinelChangeReasons": change_reasons if quality_gate_required else [],
-            "sentinelDefaultFrozen": quality_gate_required,
+            "sentinelProbePending": quality_gate_required,
+            "sentinelDefaultFrozen": False,
            "sentinelFreezeProtected": keep_frozen,
            "openaiResponsesWebSocketsV2Mode": profile.get("openaiResponsesWebSocketsV2Mode"),
            "trustUpstream": profile.get("trustUpstream") is True,
@@ -4145,7 +4234,6 @@ def ensure_sentinel_state_for_sync(account_results, pending_only=False):
        accounts_state = {}
        state["accounts"] = accounts_state
    now = utc_iso()
-    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)
@@ -4167,18 +4255,15 @@ def ensure_sentinel_state_for_sync(account_results, pending_only=False):
            continue
        changed_count += 1
        reasons = item.get("sentinelChangeReasons") if isinstance(item.get("sentinelChangeReasons"), list) else []
-        account_state["quarantine"] = {
-            "active": True,
-            "applied": True,
-            "until": pending_until if pending_only else now,
-            "intervalMinutes": 0,
-            "reason": "yaml-account-change-pending-sentinel-probe",
-            "failureKind": "pending-quality-gate",
-            "changeReasons": reasons,
-            "startedAt": now,
-            "lastBadAt": now,
-        }
-        account_state["nextProbeAfter"] = pending_until if pending_only else now
+        quarantine = account_state.get("quarantine") if isinstance(account_state.get("quarantine"), dict) else None
+        if not (isinstance(quarantine, dict) and quarantine.get("active") is True):
+            account_state["quarantine"] = {
+                "active": False,
+                "reason": "yaml-account-change-pending-sentinel-probe",
+                "lastPendingAt": now,
+                "changeReasons": reasons,
+            }
+        account_state["nextProbeAfter"] = now
        account_state["successStreak"] = 0
        account_state["successIntervalMinutes"] = 0
        profile_config = item.get("profileConfig") if isinstance(item.get("profileConfig"), dict) else {}
@@ -4191,17 +4276,18 @@ def ensure_sentinel_state_for_sync(account_results, pending_only=False):
            "changeReasons": reasons,
            "markedAt": now,
            "pendingOnly": pending_only,
+            "defaultFrozen": False,
        }
-        items.append({"accountName": name, "changeReasons": reasons, "nextProbeAfter": pending_until if pending_only else now, "defaultFrozen": True, "pendingOnly": pending_only})
+        items.append({"accountName": name, "changeReasons": reasons, "nextProbeAfter": now, "defaultFrozen": False, "defaultSchedulable": True, "pendingOnly": pending_only})
    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"
+        reason = "new-or-changed-accounts-pending-probe-prepared-default-schedulable"
    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"
+        reason = "new-or-changed-accounts-default-schedulable-and-sentinel-cadence-clamped"
    elif changed_count > 0:
-        reason = "new-or-changed-accounts-default-frozen"
+        reason = "new-or-changed-accounts-default-schedulable"
    elif len(cadence_clamped_items) > 0:
        reason = "success-cadence-clamped-to-current-config"
    else:
@@ -4777,7 +4863,7 @@ def success_body_reclassification_requirement():
                    "sourceAccountName": name,
                    "statusCode": error_code,
                    "keywords": keywords,
-                    "probeKeyword": keywords[0],
+                    "representativeKeyword": keywords[0],
                    "durationMinutes": rule.get("duration_minutes"),
                }
    return {
@@ -4785,12 +4871,12 @@ def success_body_reclassification_requirement():
        "sourceAccountName": None,
        "statusCode": None,
        "keywords": [],
-        "probeKeyword": None,
+        "representativeKeyword": None,
        "durationMinutes": None,
    }

 def model_routing_400_failover_requirement():
-    preferred = ["暂不支持", "可用模型", "unsupported model", "model not supported", "does not support", "not supported", "model_not_found", "no available channel for model"]
+    preferred = ["invalid_encrypted_content", "encrypted content", "could not be verified", "could not be decrypted", "暂不支持", "可用模型", "unsupported model", "model not supported", "does not support", "not supported", "model_not_found", "no available channel for model"]
    for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE):
        expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name])
        if expected["enabled"] is not True:
@@ -4800,13 +4886,13 @@ def model_routing_400_failover_requirement():
            keywords = rule.get("keywords") or []
            if error_code != 400 or not keywords:
                continue
-            probe_keyword = next((item for item in preferred if item in keywords), keywords[0])
+            representative_keyword = next((item for item in preferred if item in keywords), keywords[0])
            return {
                "required": True,
                "sourceAccountName": name,
                "statusCode": error_code,
                "keywords": keywords,
-                "probeKeyword": probe_keyword,
+                "representativeKeyword": representative_keyword,
                "durationMinutes": rule.get("duration_minutes"),
            }
    return {
@@ -4814,7 +4900,7 @@ def model_routing_400_failover_requirement():
        "sourceAccountName": None,
        "statusCode": None,
        "keywords": [],
-        "probeKeyword": None,
+        "representativeKeyword": None,
        "durationMinutes": None,
    }

@@ -4834,291 +4920,25 @@ def delete_probe_resource(token, method, path, label):
        "valuesPrinted": False,
    }

-def launch_success_body_mock_upstream(status_code, body_text):
-    port = 28000 + secrets.randbelow(2000)
-    body_b64 = base64.b64encode(body_text.encode("utf-8")).decode("ascii")
-    script = r'''
-set -eu
-port="$1"
-status="$2"
-body_b64="$3"
-body="$(printf "%s" "$body_b64" | base64 -d)"
-length="$(printf "%s" "$body" | wc -c | tr -d " ")"
-{ printf "HTTP/1.1 %s OK\r\nContent-Type: application/json\r\nContent-Length: %s\r\nConnection: close\r\n\r\n" "$status" "$length"; printf "%s" "$body"; } | nc -l -p "$port" -w 5
-'''
-    proc = subprocess.Popen([
-        "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
-        "--", "sh", "-c", script, "sh", str(port), str(status_code), body_b64,
-    ], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    time.sleep(0.35)
-    if proc.poll() is None:
-        return port, proc, None
-    stdout, stderr = proc.communicate(timeout=2)
-    return None, None, {
-        "ok": False,
-        "error": "mock-upstream-exited-before-probe",
-        "exitCode": proc.returncode,
-        "stdoutTail": text(stdout, 1000),
-        "stderrTail": text(stderr, 1000),
-    }
-
-def finish_mock_upstream(proc):
-    if proc is None:
-        return None
-    timed_out = False
-    try:
-        stdout, stderr = proc.communicate(timeout=2)
-    except subprocess.TimeoutExpired:
-        timed_out = True
-        proc.kill()
-        stdout, stderr = proc.communicate(timeout=2)
-    return {
-        "exitCode": proc.returncode,
-        "timedOut": timed_out,
-        "stdoutTail": text(stdout, 1000),
-        "stderrTail": text(stderr, 1000),
-    }
-
-def create_success_body_probe_resources(token, base_url, requirement):
-    stamp = str(int(time.time() * 1000))
-    suffix = stamp + "-" + "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(6))
-    group_payload_obj = group_payload()
-    group_payload_obj["name"] = "unidesk-probe-2xx-body-" + suffix
-    group_payload_obj["description"] = "UniDesk validate probe for OpenAI 2xx success-body reclassification."
-    group_id = None
-    account_id = None
-    api_key_id = None
-    try:
-        group = ensure_success(curl_api("POST", "/api/v1/admin/groups", bearer=token, payload=group_payload_obj), "create 2xx success-body probe group")
-        group_id = group.get("id") if isinstance(group, dict) else None
-        if group_id is None:
-            raise RuntimeError("2xx success-body probe group id missing")
-
-        account_payload_obj = {
-            "name": "unidesk-probe-2xx-body-" + suffix,
-            "notes": "Temporary UniDesk validate probe account; safe to delete.",
-            "platform": "openai",
-            "type": "apikey",
-            "credentials": {
-                "api_key": "sk-unidesk-probe-upstream",
-                "base_url": base_url,
-                "temp_unschedulable_enabled": True,
-                "temp_unschedulable_rules": [{
-                    "error_code": requirement["statusCode"],
-                    "keywords": [requirement["probeKeyword"]],
-                    "duration_minutes": 1,
-                    "description": "UniDesk runtime capability probe for OpenAI 2xx success-body reclassification.",
-                }],
-            },
-            "extra": {
-                "openai_responses_mode": "force_responses",
-                "unidesk_probe": "success_body_reclassification",
-            },
-            "concurrency": 1,
-            "priority": 0,
-            "rate_multiplier": 1,
-            "load_factor": 1,
-            "group_ids": [group_id],
-            "confirm_mixed_channel_risk": True,
-        }
-        account = ensure_success(curl_api("POST", "/api/v1/admin/accounts", bearer=token, payload=account_payload_obj), "create 2xx success-body probe account")
-        account_id = account.get("id") if isinstance(account, dict) else None
-        if account_id is None:
-            raise RuntimeError("2xx success-body probe account id missing")
-
-        api_key = "sk-unidesk-probe-" + "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(36))
-        api_key_obj = ensure_success(curl_api("POST", "/api/v1/keys", bearer=token, payload={
-            "name": "unidesk-probe-2xx-body-" + suffix,
-            "group_id": group_id,
-            "custom_key": api_key,
-            "quota": 0,
-            "rate_limit_5h": 0,
-            "rate_limit_1d": 0,
-            "rate_limit_7d": 0,
-        }), "create 2xx success-body probe API key")
-        api_key_id = api_key_obj.get("id") if isinstance(api_key_obj, dict) else None
-        return {
-            "groupId": group_id,
-            "groupName": group_payload_obj["name"],
-            "accountId": account_id,
-            "accountName": account_payload_obj["name"],
-            "apiKeyId": api_key_id,
-            "apiKey": api_key,
-            "keyPreview": api_key_preview(api_key),
-            "valuesPrinted": False,
-        }
-    except Exception:
-        if api_key_id is not None:
-            delete_probe_resource(token, "DELETE", f"/api/v1/keys/{api_key_id}", "api-key")
-        if account_id is not None:
-            delete_probe_resource(token, "DELETE", f"/api/v1/admin/accounts/{account_id}", "account")
-        if group_id is not None:
-            delete_probe_resource(token, "DELETE", f"/api/v1/admin/groups/{group_id}", "group")
-        raise
-
-def gateway_success_body_probe_request(api_key, request_id):
-    payload = {
-        "model": RESPONSES_SMOKE_MODEL,
-        "input": "Reply exactly: success-body probe",
-        "stream": False,
-        "store": False,
-        "max_output_tokens": 8,
-    }
-    body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
-    script = r'''
-set -eu
-token="$1"
-request_id="$2"
-tmp="$(mktemp)"
-trap 'rm -f "$tmp"' EXIT
-cat > "$tmp"
-curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X POST \
-  -H "Authorization: Bearer $token" \
-  -H 'Content-Type: application/json' \
-  -H "X-Request-ID: $request_id" \
-  -H "OpenAI-Client-Request-ID: $request_id" \
-  --data-binary @"$tmp" \
-  http://127.0.0.1:8080/v1/responses
-'''
-    proc = run([
-        "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
-        "--", "sh", "-c", script, "sh", api_key, request_id,
-    ], body)
-    return parse_curl_output(proc)
-
-def account_temp_unschedulable_probe_state(token, account_id):
-    detail = ensure_success(curl_api("GET", f"/api/v1/admin/accounts/{account_id}", bearer=token), "get temp-unschedulable probe account")
-    if not isinstance(detail, dict):
-        return {
-            "accountId": account_id,
-            "status": None,
-            "schedulable": None,
-            "tempUnschedulableUntil": None,
-            "tempUnschedulableReasonPreview": "",
-            "tempUnschedulableSet": False,
-        }
-    until = detail.get("temp_unschedulable_until") or detail.get("tempUnschedulableUntil")
-    reason = detail.get("temp_unschedulable_reason") or detail.get("tempUnschedulableReason") or ""
-    return {
-        "accountId": account_id,
-        "status": detail.get("status"),
-        "schedulable": detail.get("schedulable"),
-        "tempUnschedulableUntil": until,
-        "tempUnschedulableReasonPreview": text(str(reason), 500) if reason else "",
-        "tempUnschedulableSet": until is not None or bool(reason),
-    }
-
-def validate_success_body_reclassification(token):
-    requirement = success_body_reclassification_requirement()
-    if not requirement["required"]:
-        return {
-            "ok": True,
-            "required": False,
-            "capability": "openai-2xx-success-body-temp-unschedulable-failover",
-            "outcome": "not-required-by-yaml",
-            "valuesPrinted": False,
-        }
-
-    resources = None
-    cleanup = []
-    mock_proc = None
-    try:
-        keyword = requirement["probeKeyword"]
-        upstream_body = json.dumps({
-            "id": "resp_unidesk_success_body_probe",
-            "object": "response",
-            "created_at": int(time.time()),
-            "status": "completed",
-            "model": RESPONSES_SMOKE_MODEL,
-            "output": [{
-                "type": "message",
-                "role": "assistant",
-                "content": [{"type": "output_text", "text": keyword}],
-            }],
-            "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
-        }, separators=(",", ":"))
-        port, mock_proc, mock_error = launch_success_body_mock_upstream(requirement["statusCode"], upstream_body)
-        if mock_error is not None:
-            return {
-                "ok": False,
-                "required": True,
-                "capability": "openai-2xx-success-body-temp-unschedulable-failover",
-                "outcome": "probe-infrastructure-failed",
-                "requirement": requirement,
-                "mock": mock_error,
-                "valuesPrinted": False,
-            }
-        resources = create_success_body_probe_resources(token, f"http://127.0.0.1:{port}", requirement)
-        request_id = "unidesk-2xx-body-probe-" + str(int(time.time() * 1000))
-        started = time.time()
-        response = gateway_success_body_probe_request(resources["apiKey"], request_id)
-        mock_result = finish_mock_upstream(mock_proc)
-        mock_proc = None
-        state = account_temp_unschedulable_probe_state(token, resources["accountId"])
-        evidence = request_log_evidence(request_id)
-        supported = response.get("ok") is not True and state.get("tempUnschedulableSet") is True
-        return {
-            "ok": supported,
-            "required": True,
-            "capability": "openai-2xx-success-body-temp-unschedulable-failover",
-            "outcome": "supported" if supported else "unsupported-runtime-image",
-            "requirement": requirement,
-            "probe": {
-                "requestId": request_id,
-                "durationMs": int((time.time() - started) * 1000),
-                "httpStatus": response.get("httpStatus"),
-                "transportExitCode": response.get("transportExitCode"),
-                "responseOk": response.get("ok"),
-                "bodyPreview": text(response.get("body", ""), 800),
-                "stderr": response.get("stderr", ""),
-                "accountState": state,
-                "logEvidence": evidence,
-            },
-            "mock": mock_result,
-            "resources": {
-                "groupId": resources["groupId"],
-                "accountId": resources["accountId"],
-                "apiKeyId": resources["apiKeyId"],
-                "keyPreview": resources["keyPreview"],
-                "valuesPrinted": False,
-            },
-            "cleanup": cleanup,
-            "message": "Sub2API image must reclassify matching 2xx OpenAI bodies into failover/temp-unschedulable before statusCode=200 YAML rules are effective." if not supported else "Matching 2xx OpenAI bodies are reclassified into account failover/temp-unschedulable.",
-            "valuesPrinted": False,
-        }
-    except Exception as exc:
-        return {
-            "ok": False,
-            "required": True,
-            "capability": "openai-2xx-success-body-temp-unschedulable-failover",
-            "outcome": "probe-failed",
-            "requirement": requirement,
-            "error": str(exc),
-            "cleanup": cleanup,
-            "valuesPrinted": False,
-        }
-    finally:
-        if mock_proc is not None:
-            _ = finish_mock_upstream(mock_proc)
-        if resources is not None:
-            cleanup.append(delete_probe_resource(token, "DELETE", f"/api/v1/keys/{resources['apiKeyId']}" if resources.get("apiKeyId") is not None else "", "api-key"))
-            cleanup.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/accounts/{resources['accountId']}", "account"))
-            cleanup.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/groups/{resources['groupId']}", "group"))
-
 def validate_runtime_capabilities(token):
-    success_body = validate_success_body_reclassification(token)
+    success_body = success_body_reclassification_requirement()
    model_routing_400 = model_routing_400_failover_requirement()
    return {
-        "ok": success_body.get("ok") is True,
+        "ok": success_body.get("required") is not True and model_routing_400.get("required") is True,
        "runtimeImage": app_pod_runtime_image(),
-        "successBodyReclassification": success_body,
+        "successBodyReclassification": {
+            "ok": success_body.get("required") is not True,
+            "required": success_body.get("required"),
+            "outcome": "not-required-by-yaml" if success_body.get("required") is not True else "requires-source-review-and-real-traffic-evidence",
+            "requirement": success_body,
+            "valuesPrinted": False,
+        },
        "modelRouting400Failover": {
-            "ok": True,
-            "required": model_routing_400.get("required") is True,
-            "capability": "openai-400-model-routing-temp-unschedulable-failover",
-            "outcome": "declared-by-yaml-and-checked-by-runtime-rules",
+            "ok": model_routing_400.get("required") is True,
+            "required": model_routing_400.get("required"),
+            "outcome": "yaml-rules-present-runtime-observed-by-real-traffic",
            "requirement": model_routing_400,
-            "message": "Default validate stays short; runtime proof comes from synced rules plus real request logs/failover evidence.",
+            "evidence": "Use Sub2API source review plus validation.gatewayResponsesRecent and Artificer/real request ids for runtime proof; default validate does not create mock upstreams or temporary failover accounts.",
            "valuesPrinted": False,
        },
        "valuesPrinted": False,
@@ -5418,7 +5238,7 @@ def run_sync():
    planned_account_results = planned_sentinel_account_results(profiles, existing_accounts)
    sentinel_quality_prepare = ensure_sentinel_state_for_sync(planned_account_results, True)
    if sentinel_quality_prepare.get("ok") is not True:
-        raise RuntimeError("prepare sentinel quality gate failed: " + json.dumps(sentinel_quality_prepare, ensure_ascii=False))
+        raise RuntimeError("prepare sentinel pending probe failed: " + json.dumps(sentinel_quality_prepare, ensure_ascii=False))
    protected_frozen_names = active_sentinel_quarantine_names()
    account_results, pruned_account_results = ensure_accounts(token, profiles, group_id, prune_removed, protected_frozen_names, existing_accounts)
    capacity_status = account_capacity_status(token)
@@ -5438,7 +5258,7 @@ def run_sync():
    sentinel_quality = ensure_sentinel_state_for_sync(account_results)
    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 and sentinel.get("ok") is True and sentinel_quality_prepare.get("ok") is True and sentinel_quality.get("ok") is True and sentinel_reassert.get("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_quality_prepare.get("ok") is True and sentinel_quality.get("ok") is True and sentinel_reassert.get("ok") is True and runtime_capabilities.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,
@@ -5502,7 +5322,7 @@ def run_validate():
    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 and sentinel.get("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 and runtime_capabilities.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,