export function runtimeRemoteScriptTail(): string { return `def native_customer_error_profile(token, group_id, platform): start_time = quote(since_start_time(PAYLOAD.get("since")), safe="") platform_query = "&platform=" + quote(str(platform), safe="") if platform else "" page = 1 page_size = 500 rows = [] total = None try: while True: response = curl_api("GET", f"/api/v1/admin/ops/request-errors?group_id={group_id}&start_time={start_time}&view=all&page={page}&page_size={page_size}{platform_query}", bearer=token) data = data_of(response, "get customer-visible request errors") page_rows = items_of(data) rows.extend(page_rows) if total is None and isinstance(data, dict) and isinstance(data.get("total"), int): total = data.get("total") if not page_rows or len(page_rows) < page_size or (isinstance(total, int) and len(rows) >= total): break page += 1 except Exception as exc: return {"status": "unavailable", "source": "sub2api-native-admin-ops-request-errors", "reason": compact_error(str(exc))} if total is None: total = len(rows) def buckets(key_fn): counts = {} for row in rows: value = key_fn(row) if isinstance(row, dict) else None label = str(value) if value not in (None, "") else "unknown" counts[label] = counts.get(label, 0) + 1 return [{"value": key, "count": count} for key, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))[:20 if PAYLOAD.get("full") is True else 8]] samples = [] account_metrics = {} def customer_error_attribution(row): message = str(row.get("message") or row.get("error_message") or "").lower() if "context window" in message or "context_length_exceeded" in message: return False, "context-window" if "input must be a list" in message: return False, "invalid-client-input" if "not supported by any configured account" in message or "no available channel for model" in message: return False, "model-route" phase = str(row.get("phase") or "").lower() if phase in ("internal", "client", "business"): return False, "non-upstream-phase" if row.get("account_id") is None: return False, "no-account-attribution" category = " ".join(str(row.get(key) or "").lower() for key in ("type", "category", "error_type")) if phase == "upstream" or "upstream" in category: return True, "explicit-upstream" upstream_messages = ( "upstream service temporarily unavailable", "upstream request failed", "bad gateway", "gateway timeout", "error code: 502", "error code: 503", "error code: 504", "error code: 524", ) if any(marker in message for marker in upstream_messages): return True, "stable-upstream-message" return False, "unattributed-customer-error" seen = set() for row in rows: if not isinstance(row, dict): continue account_id = row.get("account_id") account_key = str(account_id) if account_id is not None else "unknown" if account_key not in account_metrics: account_metrics[account_key] = { "accountId": account_id, "customerErrorEvents": 0, "customerErrorRequestIds": set(), "upstreamRequestIds": set(), "excludedRequestIds": set(), "excludedReasons": {}, "statusCodes": {}, } metric = account_metrics[account_key] metric["customerErrorEvents"] += 1 request_id = row.get("request_id") if isinstance(request_id, str) and request_id: metric["customerErrorRequestIds"].add(request_id) scoreable, reason = customer_error_attribution(row) target = metric["upstreamRequestIds"] if scoreable else metric["excludedRequestIds"] target.add(request_id) if not scoreable: metric["excludedReasons"][reason] = metric["excludedReasons"].get(reason, 0) + 1 status_code = row.get("status_code") if isinstance(status_code, int): status_key = str(status_code) metric["statusCodes"][status_key] = metric["statusCodes"].get(status_key, 0) + 1 if not request_id: continue key = (row.get("requested_model") or row.get("model"), row.get("status_code"), row.get("account_id"), row.get("stream"), row.get("request_path") or row.get("inbound_endpoint")) sample_limit = 30 if PAYLOAD.get("full") is True else 15 if key in seen or len(samples) >= sample_limit: continue seen.add(key) samples.append({ "requestId": row.get("request_id"), "model": key[0], "statusCode": key[1], "accountId": key[2], "mode": "stream" if key[3] is True else "sync" if key[3] is False else "unknown", "path": key[4], "message": compact_error(row.get("message") or row.get("error_message")), }) return { "status": "available", "source": "sub2api-native-admin-ops-request-errors", "total": total, "returned": len(rows), "pagesRead": page, "recordsTruncated": False, "modelBuckets": buckets(lambda row: row.get("requested_model") or row.get("model")), "accountBuckets": buckets(lambda row: row.get("account_id")), "statusBuckets": buckets(lambda row: row.get("status_code")), "streamBuckets": buckets(lambda row: "stream" if row.get("stream") is True else "sync" if row.get("stream") is False else "unknown"), "messageBuckets": buckets(lambda row: compact_error(row.get("message") or row.get("error_message"))), "phaseBuckets": buckets(lambda row: row.get("phase")), "endpointBuckets": buckets(lambda row: row.get("inbound_endpoint") or row.get("request_path")), "pathBuckets": buckets(lambda row: row.get("request_path") or row.get("inbound_endpoint")), "accountMetrics": [ { "accountId": item["accountId"], "customerErrorEvents": item["customerErrorEvents"], "customerErrorRequests": len(item["customerErrorRequestIds"]), "_requestIds": sorted(item["customerErrorRequestIds"]), "_upstreamRequestIds": sorted(item["upstreamRequestIds"]), "_excludedRequestIds": sorted(item["excludedRequestIds"]), "scoreableUpstreamErrorRequests": len(item["upstreamRequestIds"]), "excludedNonUpstreamErrorRequests": len(item["excludedRequestIds"]), "excludedReasonBuckets": [{"reason": reason, "count": count} for reason, count in sorted(item["excludedReasons"].items())], "statusBuckets": [ {"statusCode": int(code), "count": count} for code, count in sorted(item["statusCodes"].items(), key=lambda pair: int(pair[0])) ], } for item in sorted(account_metrics.values(), key=lambda item: int(item.get("accountId") or 0)) ], "samples": samples, } def parse_native_time(value): if not isinstance(value, str) or not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) except ValueError: return None def metric_percentile(values, fraction): ordered = sorted(value for value in values if isinstance(value, (int, float)) and value >= 0) if not ordered: return None position = (len(ordered) - 1) * fraction lower = int(position) upper = min(lower + 1, len(ordered) - 1) return round(ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower)) def numeric_value(value): try: return float(value or 0) except (TypeError, ValueError): return 0.0 def account_cost_rate(account_name): match = re.search(r"(\\d+(?:\\.\\d+)?)$", str(account_name or "")) return float(match.group(1)) if match else None def account_usage_metrics(token, accounts, group_id): start_at = parse_native_time(since_start_time(PAYLOAD.get("since"))) end_at = datetime.now(timezone.utc) result = {} for account in accounts: account_id = account.get("id") key = str(account_id) rows = [] page = 1 status = "available" reason = None try: while True: path = ( f"/api/v1/admin/usage?page={page}&page_size=100&group_id={group_id}&account_id={account_id}" f"&start_date={start_at.date().isoformat()}&end_date={end_at.date().isoformat()}&timezone=UTC" "&sort_by=created_at&sort_order=desc" ) data = data_of(curl_api("GET", path, bearer=token), "list account usage") page_rows = items_of(data) reached_start = False for row in page_rows: created_at = parse_native_time(row.get("created_at") if isinstance(row, dict) else None) if created_at is None or created_at < start_at or created_at > end_at: if created_at is not None and created_at < start_at: reached_start = True continue rows.append(row) if not page_rows or len(page_rows) < 100 or reached_start: break page += 1 except Exception as exc: status = "unavailable" reason = compact_error(str(exc)) stream_rows = [row for row in rows if isinstance(row, dict) and row.get("stream") is True] ttft_values = [row.get("first_token_ms") for row in stream_rows if isinstance(row.get("first_token_ms"), int)] duration_values = [row.get("duration_ms") for row in rows if isinstance(row.get("duration_ms"), int)] models = {} for row in rows: model = str(row.get("model") or "unknown") models[model] = models.get(model, 0) + 1 result[key] = { "status": status, "reason": reason, "source": "sub2api-native-admin-usage", "pagesRead": page, "successRequests": len(rows), "streamSuccessRequests": len(stream_rows), "requestCount": len(rows), "tokenCount": sum(int(row.get("input_tokens") or 0) + int(row.get("output_tokens") or 0) for row in rows), "apiAmountUsd": round(sum(numeric_value(row.get("actual_cost")) for row in rows), 8), "firstTokenSamples": len(ttft_values), "firstTokenCoverage": round(len(ttft_values) / len(stream_rows), 6) if stream_rows else None, "ttftP50Ms": metric_percentile(ttft_values, 0.50), "ttftP95Ms": metric_percentile(ttft_values, 0.95), "ttftP99Ms": metric_percentile(ttft_values, 0.99), "ttftMaxMs": max(ttft_values) if ttft_values else None, "durationP95Ms": metric_percentile(duration_values, 0.95), "modelBuckets": [{"model": model, "count": count} for model, count in sorted(models.items(), key=lambda item: (-item[1], item[0]))[:8]], } return result def reliability_score_points(failure_rate): if not isinstance(failure_rate, (int, float)): return None return round(60 * (1 - min(max(failure_rate, 0), 0.20) / 0.20), 2) def latency_score_points(ttft_p95_ms): if not isinstance(ttft_p95_ms, (int, float)): return None return round(25 * (1 - min(max(ttft_p95_ms - 10000, 0), 170000) / 170000), 2) def account_quality_scores(token, accounts, group_id, native_ops, effectiveness): usage_by_account = account_usage_metrics(token, accounts, group_id) customer_profile = native_ops.get("customerErrors") if isinstance(native_ops.get("customerErrors"), dict) else {} customer_by_account = {str(item.get("accountId")): item for item in customer_profile.get("accountMetrics") or [] if isinstance(item, dict)} policy_by_account = {str(item.get("accountId")): item for item in effectiveness.get("accountPolicyMetrics") or [] if isinstance(item, dict)} availability = ((native_ops.get("accountAvailability") or {}).get("summary") or {}) unavailable_ids = set(str(item.get("accountId")) for item in availability.get("unavailableAccounts") or [] if isinstance(item, dict)) rows = [] for account in accounts: account_id = account.get("id") key = str(account_id) usage = usage_by_account.get(key) or {} customer = customer_by_account.get(key) or {} policy = policy_by_account.get(key) or {} policy_failure_ids = set(policy.pop("_failureRequestIds", []) or []) customer_failure_ids = set(customer.get("_upstreamRequestIds", []) or []) excluded_failure_ids = set(customer.get("_excludedRequestIds", []) or []) failure_requests = len((policy_failure_ids | customer_failure_ids) - excluded_failure_ids) success_requests = usage.get("successRequests") if isinstance(usage.get("successRequests"), int) else 0 observed_attempts = success_requests + failure_requests failure_rate = round(failure_requests / observed_attempts, 6) if observed_attempts > 0 else None reliability_points = reliability_score_points(failure_rate) ttft_samples = usage.get("firstTokenSamples") if isinstance(usage.get("firstTokenSamples"), int) else 0 latency_points = latency_score_points(usage.get("ttftP95Ms")) if ttft_samples >= 5 else None current_available = key not in unavailable_ids and account.get("status") == "active" and account.get("schedulable") is not False availability_points = 15 if current_available else 8 if account.get("status") == "active" else 0 earned = (reliability_points or 0) + (latency_points or 0) + availability_points available_weight = (60 if reliability_points is not None else 0) + (25 if latency_points is not None else 0) + 15 score = round(earned / available_weight * 100, 1) if observed_attempts > 0 and available_weight > 0 else None score_comparable = observed_attempts >= 10 and ttft_samples >= 5 grade = "insufficient" if isinstance(score, (int, float)) and (score_comparable or (score < 60 and observed_attempts >= 10)): grade = "A" if score >= 90 else "B" if score >= 80 else "C" if score >= 70 else "D" if score >= 60 else "E" assessment = "preferred" if grade == "A" else "healthy" if grade == "B" else "watch" if grade == "C" else "degraded" if grade == "D" else "poor" if grade == "E" else "insufficient-evidence" confidence = "high" if observed_attempts >= 50 and ttft_samples >= 20 else "medium" if observed_attempts >= 10 and ttft_samples >= 5 else "low" reasons = [] if isinstance(failure_rate, (int, float)) and failure_rate >= 0.10: reasons.append("failure-rate>=10%") elif isinstance(failure_rate, (int, float)) and failure_rate >= 0.03: reasons.append("failure-rate>=3%") if isinstance(usage.get("ttftP95Ms"), int) and usage.get("ttftP95Ms") >= 120000: reasons.append("ttft-p95>=120s") if policy.get("failoverRequests", 0) > 0: reasons.append("upstream-failover-triggered") if policy.get("sameAccountRetryEvents", 0) > 0: reasons.append("same-account-retry-observed") if not current_available: reasons.append("currently-unavailable") if ttft_samples < 5: reasons.append("ttft-evidence-insufficient") if observed_attempts < 10: reasons.append("request-evidence-insufficient") cost_rate = account_cost_rate(account.get("name")) api_amount = usage.get("apiAmountUsd") rows.append({ "accountId": account_id, "accountName": account.get("name"), "status": account.get("status"), "schedulable": account.get("schedulable"), "currentlyAvailable": current_available, "score": score, "grade": grade, "assessment": assessment, "confidence": confidence, "scoreComparable": score_comparable, "observedAttempts": observed_attempts, "successRequests": success_requests, "failureRequests": failure_requests, "failureRate": failure_rate, "streamSuccessRequests": usage.get("streamSuccessRequests"), "firstTokenSamples": ttft_samples, "firstTokenCoverage": usage.get("firstTokenCoverage"), "ttftP50Ms": usage.get("ttftP50Ms"), "ttftP95Ms": usage.get("ttftP95Ms"), "ttftP99Ms": usage.get("ttftP99Ms"), "ttftMaxMs": usage.get("ttftMaxMs"), "durationP95Ms": usage.get("durationP95Ms"), "modelBuckets": usage.get("modelBuckets"), "customerErrorRequests": customer.get("customerErrorRequests", 0), "scoreableUpstreamErrorRequests": customer.get("scoreableUpstreamErrorRequests", 0), "excludedNonUpstreamErrorRequests": customer.get("excludedNonUpstreamErrorRequests", 0), "excludedReasonBuckets": customer.get("excludedReasonBuckets", []), "usage": { "requestCount": usage.get("requestCount"), "tokenCount": usage.get("tokenCount"), "apiAmountUsd": api_amount, "costRateCnyPerApiUsd": cost_rate, "upstreamCostCny": round(api_amount * cost_rate, 8) if isinstance(api_amount, (int, float)) and isinstance(cost_rate, (int, float)) else None, }, "failoverRequests": policy.get("failoverRequests", 0), "failoverRecovered": policy.get("failoverRecovered", 0), "failoverFailed": policy.get("failoverFailed", 0), "failoverOutcomeMissing": policy.get("failoverOutcomeMissing", 0), "sameAccountRetryEvents": policy.get("sameAccountRetryEvents", 0), "tempUnschedulableEvents": policy.get("tempUnschedulableEvents", 0), "forwardFailedRequests": policy.get("forwardFailedRequests", 0), "upstreamStatusBuckets": policy.get("upstreamStatusBuckets", []), "scoreComponents": {"reliability": reliability_points, "latency": latency_points, "availability": availability_points, "availableWeight": available_weight}, "reasons": reasons, "usageStatus": usage.get("status"), "usageReason": usage.get("reason"), }) rows.sort(key=lambda item: (item.get("score") is None, -(item.get("score") or -1), int(item.get("accountId") or 0))) return { "source": "sub2api-native-admin-usage-request-errors-and-system-logs", "scope": "group-account-window", "window": PAYLOAD.get("since"), "scoreRule": { "reliabilityWeight": 60, "latencyWeight": 25, "availabilityWeight": 15, "failureDefinition": "unique request ids with account failover, forward_failed, or explicitly attributable upstream customer error; client input, model-route, internal/business, no-account, and unattributed errors are excluded", "reliability": "linear from 60 points at 0% observed failures to 0 points at 20%", "latency": "requires at least 5 first-token samples; linear from 25 points at <=10s TTFT P95 to 0 points at >=180s", "availability": "15 points when active, schedulable, and currently available; 8 when active but unavailable; otherwise 0", "missingMetric": "missing latency is excluded from the denominator and lowers confidence; it is not treated as zero latency", }, "accounts": rows, "attribution": "Success denominators, usage, token count, API amount, and first-token latency come from native admin usage. Reliability only includes explicitly attributable upstream customer errors and deduplicated failover/forward request ids; all excluded reasons remain visible.", } def all_group_error_overview(token, groups): page_size = 20 cursor = int(PAYLOAD.get("pageToken") or 0) ordered = sorted( [item for item in groups if isinstance(item, dict) and isinstance(item.get("id"), int)], key=lambda item: item["id"], ) remaining = [item for item in ordered if item["id"] > cursor] page = remaining[:page_size] summaries = [] totals = { "requestCount": 0, "errorCount": 0, "businessLimitedCount": 0, "upstreamErrorCount": 0, "groupsNeedingAttention": 0, } for group in page: group_id = group["id"] platform = PAYLOAD.get("platform") or group.get("platform") snapshot = native_ops_snapshot(token, group_id, platform) overview = ((snapshot.get("overview") or {}).get("summary") or {}) availability = ((snapshot.get("accountAvailability") or {}).get("summary") or {}) availability_group = availability.get("group") if isinstance(availability.get("group"), dict) else {} concurrency = ((snapshot.get("concurrency") or {}).get("summary") or {}) concurrency_group = concurrency.get("group") if isinstance(concurrency.get("group"), dict) else {} request_count = overview.get("request_count_total") error_count = overview.get("error_count_total") business_limited = overview.get("business_limited_count") upstream_error_count = sum( value for value in ( overview.get("upstream_error_count_excl_429_529"), overview.get("upstream_429_count"), overview.get("upstream_529_count"), ) if isinstance(value, int) ) unavailable_count = availability.get("unavailableAccountCount") needs_attention = ( (isinstance(error_count, int) and error_count > 0) or upstream_error_count > 0 or (isinstance(unavailable_count, int) and unavailable_count > 0) ) totals["requestCount"] += request_count if isinstance(request_count, int) else 0 totals["errorCount"] += error_count if isinstance(error_count, int) else 0 totals["businessLimitedCount"] += business_limited if isinstance(business_limited, int) else 0 totals["upstreamErrorCount"] += upstream_error_count totals["groupsNeedingAttention"] += 1 if needs_attention else 0 summaries.append({ "groupId": group_id, "groupName": group.get("name"), "platform": platform, "status": group.get("status"), "requestCount": request_count, "errorCount": error_count, "errorRate": overview.get("error_rate"), "upstreamErrorCount": upstream_error_count, "upstreamErrorRate": overview.get("upstream_error_rate"), "businessLimitedCount": business_limited, "ttftP99Ms": ((overview.get("ttft") or {}).get("p99_ms") if isinstance(overview.get("ttft"), dict) else None), "durationP99Ms": ((overview.get("duration") or {}).get("p99_ms") if isinstance(overview.get("duration"), dict) else None), "totalAccounts": availability_group.get("total_accounts"), "availableCount": availability_group.get("available_count"), "rateLimitCount": availability_group.get("rate_limit_count"), "errorAccountCount": availability_group.get("error_count"), "unavailableAccountCount": unavailable_count, "currentInUse": concurrency_group.get("current_in_use"), "maxCapacity": concurrency_group.get("max_capacity"), "waitingInQueue": concurrency_group.get("waiting_in_queue"), "needsAttention": needs_attention, "opsStatus": { "overview": (snapshot.get("overview") or {}).get("status"), "accountAvailability": (snapshot.get("accountAvailability") or {}).get("status"), "concurrency": (snapshot.get("concurrency") or {}).get("status"), }, }) next_token = str(page[-1]["id"]) if len(remaining) > page_size and page else None totals["errorRate"] = round(totals["errorCount"] / totals["requestCount"], 6) if totals["requestCount"] > 0 else None totals["upstreamErrorRate"] = round(totals["upstreamErrorCount"] / totals["requestCount"], 6) if totals["requestCount"] > 0 else None ttft_p99_values = [item.get("ttftP99Ms") for item in summaries if isinstance(item.get("ttftP99Ms"), int)] max_ttft_p99 = max(ttft_p99_values) if ttft_p99_values else None observed_floor = max(60, ((max_ttft_p99 + 29999) // 15000) * 15) if isinstance(max_ttft_p99, int) else None return { "scope": "all-groups-overview", "window": {"since": PAYLOAD.get("since")}, "platform": PAYLOAD.get("platform"), "totalGroups": len(ordered), "returnedGroups": len(summaries), "pageSize": page_size, "pageToken": str(cursor) if cursor > 0 else None, "nextPageToken": next_token, "pageTotals": totals, "responseHeaderTimeout": runtime_response_header_timeout(), "requestTiming": { "maxTtftP99Ms": max_ttft_p99, "observedFloorSeconds": observed_floor, "scope": "returned-groups-page", "attribution": "The observed floor is the maximum returned-group TTFT P99 plus 15 seconds rounded up to 15 seconds. It is analytical only and must not be treated as a safe timeout.", }, "groups": summaries, "disclosure": "Page totals cover only returned groups. Follow nextPageToken for remaining groups and use --group for account root causes and request indexes.", } def native_system_log_lines(token, platform=None): start_time = quote(since_start_time(PAYLOAD.get("since")), safe="") platform_query = "&platform=" + quote(str(platform), safe="") if platform else "" markers = ( "account_temp_unschedulable", "openai.upstream_failover_switching", "openai.account_select_failed", "openai.forward_failed", "openai.pool_mode_same_account_retry", "gateway.failover_same_account_retry", "http request completed", ) rows_by_id = {} queries = [] for marker in markers: page = 1 returned = 0 total = None try: while True: path = f"/api/v1/admin/ops/system-logs?page={page}&page_size=200&start_time={start_time}&q={quote(marker, safe='')}{platform_query}" data = data_of(curl_api("GET", path, bearer=token), "list ops system logs") rows = items_of(data) returned += len(rows) if total is None and isinstance(data, dict) and isinstance(data.get("total"), int): total = data.get("total") for row in rows: if not isinstance(row, dict): continue row_id = row.get("id") key = str(row_id) if row_id is not None else json.dumps(row, sort_keys=True, default=str) rows_by_id[key] = row if not rows or len(rows) < 200 or (isinstance(total, int) and returned >= total): break page += 1 except Exception as exc: return None, { "status": "unavailable", "source": "sub2api-native-admin-ops-system-logs", "reason": compact_error(str(exc)), "fallback": "runtime-container-logs", } queries.append({"marker": marker, "returned": returned, "total": total if total is not None else returned, "pagesRead": page, "truncated": False}) lines = [] for row in sorted(rows_by_id.values(), key=lambda item: str(item.get("created_at") or "")): extra = dict(row.get("extra") or {}) if isinstance(row.get("extra"), dict) else {} for source_key, target_key in ( ("request_id", "request_id"), ("client_request_id", "client_request_id"), ("account_id", "account_id"), ("platform", "platform"), ("model", "model"), ): if extra.get(target_key) is None and row.get(source_key) is not None: extra[target_key] = row.get(source_key) message = str(row.get("message") or "") lines.append(str(row.get("created_at") or "") + "\\t" + message + "\\t" + json.dumps(extra, separators=(",", ":"), default=str)) return lines, { "status": "available", "source": "sub2api-native-admin-ops-system-logs", "uniqueEventCount": len(lines), "queries": queries, "fallbackUsed": False, } def policy_effectiveness(lines, accounts, source, group_id=None): account_names = {str(item.get("id")): item.get("name") for item in accounts} account_ids = set(account_names.keys()) account_policy = { key: { "accountId": item.get("id"), "accountName": item.get("name"), "tempUnschedulableEvents": 0, "failoverEvents": 0, "failoverRequestIds": set(), "failoverRecoveredRequestIds": set(), "failoverFailedRequestIds": set(), "failoverOutcomeMissingRequestIds": set(), "forwardFailedEvents": 0, "forwardFailedRequestIds": set(), "sameAccountRetryEvents": 0, "upstreamStatusCodes": {}, } for key, item in ((str(item.get("id")), item) for item in accounts) } requests = {} counts = { "tempUnschedulableEvents": 0, "failoverEvents": 0, "selectFailedEvents": 0, "forwardFailedEvents": 0, "sameAccountRetryEvents": 0, "gatewayCompletedEvents": 0, "gatewayFinalErrorEvents": 0, } missing_request_ids = { "tempUnschedulable": 0, "failover": 0, "selectFailed": 0, "forwardFailed": 0, "sameAccountRetry": 0, "gatewayCompleted": 0, } temp_accounts = {} temp_samples = [] failover_samples = [] select_failed_samples = [] final_statuses = {} event_line_count = 0 def request_entry(request_id): if not isinstance(request_id, str) or not request_id: return None if request_id not in requests: requests[request_id] = { "requestId": request_id, "tempUnschedulableCount": 0, "failoverCount": 0, "selectFailedCount": 0, "forwardFailedCount": 0, "sameAccountRetryCount": 0, "finalStatus": None, } return requests[request_id] for line in lines: if not any(marker in line for marker in ( "account_temp_unschedulable", "openai.upstream_failover_switching", "openai.account_select_failed", "openai.forward_failed", "openai.pool_mode_same_account_retry", "gateway.failover_same_account_retry", "http request completed", )): continue json_start = line.find("{") if json_start < 0: continue try: item = json.loads(line[json_start:]) except json.JSONDecodeError: continue if not isinstance(item, dict): continue at = line.split("\\t", 1)[0].strip() or None request_id = item.get("request_id") account_id = item.get("account_id") event_group_id = item.get("group_id") if event_group_id is not None and group_id is not None and str(event_group_id) != str(group_id): continue if account_id is not None and str(account_id) not in account_ids: continue explicitly_scoped = account_id is not None or event_group_id is not None if not explicitly_scoped and (not isinstance(request_id, str) or request_id not in requests): continue request = request_entry(request_id) base = { "at": at, "requestId": request_id, "accountId": account_id, "accountName": account_names.get(str(account_id)), } raw_account_stat = account_policy.get(str(account_id)) account_stat = raw_account_stat if group_id is None or event_group_id is not None else None if "account_temp_unschedulable" in line: event_line_count += 1 counts["tempUnschedulableEvents"] += 1 if request is not None: request["tempUnschedulableCount"] += 1 else: missing_request_ids["tempUnschedulable"] += 1 key = str(account_id) if account_id is not None else "unknown" if key not in temp_accounts: temp_accounts[key] = { "accountId": account_id, "accountName": account_names.get(str(account_id)), "count": 0, } temp_accounts[key]["count"] += 1 if raw_account_stat is not None: raw_account_stat["tempUnschedulableEvents"] += 1 if len(temp_samples) < 20: temp_samples.append({ **base, "statusCode": item.get("status_code"), "ruleIndex": item.get("rule_index"), "matchedKeyword": item.get("matched_keyword"), "until": item.get("until") or item.get("temp_unschedulable_until"), "reason": compact_error(item.get("reason") or item.get("error")), }) elif "openai.upstream_failover_switching" in line: event_line_count += 1 counts["failoverEvents"] += 1 if request is not None: request["failoverCount"] += 1 else: missing_request_ids["failover"] += 1 if account_stat is not None: account_stat["failoverEvents"] += 1 if isinstance(request_id, str) and request_id: account_stat["failoverRequestIds"].add(request_id) status_code = item.get("upstream_status") if isinstance(status_code, int): key = str(status_code) account_stat["upstreamStatusCodes"][key] = account_stat["upstreamStatusCodes"].get(key, 0) + 1 if len(failover_samples) < 20: failover_samples.append({ **base, "switchCount": item.get("switch_count"), "maxSwitches": item.get("max_switches"), }) elif "openai.account_select_failed" in line: event_line_count += 1 counts["selectFailedEvents"] += 1 if request is not None: request["selectFailedCount"] += 1 else: missing_request_ids["selectFailed"] += 1 if len(select_failed_samples) < 20: select_failed_samples.append({ **base, "excludedAccountCount": item.get("excluded_account_count"), "error": compact_error(item.get("error")), }) elif "openai.forward_failed" in line: event_line_count += 1 counts["forwardFailedEvents"] += 1 if request is not None: request["forwardFailedCount"] += 1 else: missing_request_ids["forwardFailed"] += 1 if account_stat is not None: account_stat["forwardFailedEvents"] += 1 if isinstance(request_id, str) and request_id: account_stat["forwardFailedRequestIds"].add(request_id) elif "openai.pool_mode_same_account_retry" in line or "gateway.failover_same_account_retry" in line: event_line_count += 1 counts["sameAccountRetryEvents"] += 1 if request is not None: request["sameAccountRetryCount"] += 1 else: missing_request_ids["sameAccountRetry"] += 1 if raw_account_stat is not None: raw_account_stat["sameAccountRetryEvents"] += 1 elif "http request completed" in line and gateway_request_path(item.get("path")): event_line_count += 1 counts["gatewayCompletedEvents"] += 1 status_code = item.get("status_code") if request is not None and isinstance(status_code, int): request["finalStatus"] = status_code elif request is None: missing_request_ids["gatewayCompleted"] += 1 if isinstance(status_code, int): key = str(status_code) final_statuses[key] = final_statuses.get(key, 0) + 1 if status_code >= 400: counts["gatewayFinalErrorEvents"] += 1 correlated = [item for item in requests.values() if item["failoverCount"] > 0] succeeded = [item for item in correlated if isinstance(item.get("finalStatus"), int) and item["finalStatus"] < 400] failed = [item for item in correlated if isinstance(item.get("finalStatus"), int) and item["finalStatus"] >= 400] incomplete = [item for item in correlated if not isinstance(item.get("finalStatus"), int)] cooled_then_failover = [item for item in correlated if item["tempUnschedulableCount"] > 0] forward_with_http_success = [ item for item in requests.values() if item["forwardFailedCount"] > 0 and isinstance(item.get("finalStatus"), int) and item["finalStatus"] < 400 ] forward_with_http_error = [ item for item in requests.values() if item["forwardFailedCount"] > 0 and isinstance(item.get("finalStatus"), int) and item["finalStatus"] >= 400 ] forward_outcome_missing = [ item for item in requests.values() if item["forwardFailedCount"] > 0 and not isinstance(item.get("finalStatus"), int) ] for request in correlated: request_id = request.get("requestId") if not isinstance(request_id, str): continue for current in account_policy.values(): if request_id not in current["failoverRequestIds"]: continue final_status = request.get("finalStatus") if isinstance(final_status, int) and final_status < 400: current["failoverRecoveredRequestIds"].add(request_id) elif isinstance(final_status, int): current["failoverFailedRequestIds"].add(request_id) else: current["failoverOutcomeMissingRequestIds"].add(request_id) account_policy_rows = [] for current in account_policy.values(): account_policy_rows.append({ "accountId": current["accountId"], "accountName": current["accountName"], "tempUnschedulableEvents": current["tempUnschedulableEvents"], "failoverEvents": current["failoverEvents"], "failoverRequests": len(current["failoverRequestIds"]), "failoverRecovered": len(current["failoverRecoveredRequestIds"]), "failoverFailed": len(current["failoverFailedRequestIds"]), "failoverOutcomeMissing": len(current["failoverOutcomeMissingRequestIds"]), "forwardFailedEvents": current["forwardFailedEvents"], "forwardFailedRequests": len(current["forwardFailedRequestIds"]), "sameAccountRetryEvents": current["sameAccountRetryEvents"], "_failureRequestIds": sorted(current["failoverRequestIds"] | current["forwardFailedRequestIds"]), "upstreamStatusBuckets": [ {"statusCode": int(code), "count": count} for code, count in sorted(current["upstreamStatusCodes"].items(), key=lambda pair: int(pair[0])) ], }) sample_limit = 20 if PAYLOAD.get("full") is True else (2 if PAYLOAD.get("account") is not None else 5) completed = counts["gatewayCompletedEvents"] final_errors = counts["gatewayFinalErrorEvents"] result = { "source": source, "scope": "group-membership-window", "attribution": "Events with group_id are exact. Events without group_id are attributed by account membership and request_id correlation; shared-account events can appear in more than one group drilldown.", "eventLineCount": event_line_count, **counts, "tempUnschedulableAccounts": sorted(temp_accounts.values(), key=lambda item: (-item["count"], str(item.get("accountName") or ""))), "accountPolicyMetrics": sorted(account_policy_rows, key=lambda item: int(item.get("accountId") or 0)), "accountPolicyAttribution": "Failover and forward-failure request ids enter group scoring only when system-log events carry group_id. Account temp-unschedulable and same-account retry counts are account-wide operational signals and are not added to the failure-rate numerator.", "failoverRequestCount": len(correlated), "failoverSucceededRequestCount": len(succeeded), "failoverFailedRequestCount": len(failed), "failoverOutcomeMissingRequestCount": len(incomplete), "cooledThenFailoverRequestCount": len(cooled_then_failover), "forwardFailureWithHttpSuccessRequestCount": len(forward_with_http_success), "forwardFailureWithHttpErrorRequestCount": len(forward_with_http_error), "forwardFailureOutcomeMissingRequestCount": len(forward_outcome_missing), "requestIdCoverage": { "missingByEventType": missing_request_ids, "tempToFailoverCorrelationAvailable": counts["tempUnschedulableEvents"] > missing_request_ids["tempUnschedulable"], }, "gatewayFinalStatusBuckets": [ {"statusCode": int(code), "count": count} for code, count in sorted(final_statuses.items(), key=lambda pair: int(pair[0])) ], "gatewayOutcomeEvidence": { "completedLogEventCount": completed, "finalErrorLogEventCount": final_errors, "observedFinalErrorShare": round(final_errors / completed, 4) if completed > 0 else None, "denominator": "gateway http request completed log events for Responses, Messages, and Chat Completions paths", "coverage": "since-window-and-tail-bounded", "completeTrafficDenominatorGuaranteed": False, }, "requestDrilldown": { "succeededAfterFailover": [item["requestId"] for item in succeeded[:sample_limit]], "failedAfterFailover": [item["requestId"] for item in failed[:sample_limit]], "outcomeMissingAfterFailover": [item["requestId"] for item in incomplete[:sample_limit]], "selectFailed": [item["requestId"] for item in requests.values() if item["selectFailedCount"] > 0][:sample_limit], "forwardFailureWithHttpSuccess": [item["requestId"] for item in forward_with_http_success[:sample_limit]], "forwardFailureWithHttpError": [item["requestId"] for item in forward_with_http_error[:sample_limit]], "forwardFailureOutcomeMissing": [item["requestId"] for item in forward_outcome_missing[:sample_limit]], }, "assessment": ( "failover-observed-with-success" if succeeded else "failover-observed-without-success" if correlated else "no-failover-observed" ), "disclosure": "Rule matches require account_temp_unschedulable evidence; status codes alone are not treated as rule hits. Request outcomes are joined only by request_id. HTTP success can coexist with forward_failed on streaming responses and does not prove the client received a complete terminal event.", } if PAYLOAD.get("full") is True: result["samples"] = { "tempUnschedulable": temp_samples, "failovers": failover_samples, "selectFailures": select_failed_samples, } return result def observed_runtime_errors(token, accounts, group_id, platform=None): selector = PAYLOAD.get("account") selected = [item for item in accounts if selector is None or str(item.get("id")) == str(selector) or item.get("name") == selector] if selector is not None and len(selected) != 1: raise RuntimeError("account-not-found-or-ambiguous: " + str(selector)) selected_ids = set(str(item.get("id")) for item in selected) stats = {} for item in selected: stats[str(item.get("id"))] = { "accountName": item.get("name"), "accountId": item.get("id"), "upstreamEventCount": 0, "observedStatusCodes": {}, "forwardFailureCount": 0, "forwardFailureStatusCodes": {}, "lastSeenAt": None, "errorSamples": [], } native_ops = native_ops_snapshot(token, group_id, platform) lines, system_log_source = native_system_log_lines(token, platform) if lines is None: lines = runtime_log_lines() system_log_source["fallbackUsed"] = True else: marker_queries = [item for item in system_log_source.get("queries", []) if item.get("marker") != "http request completed"] marker_total = sum(item.get("total", 0) for item in marker_queries if isinstance(item.get("total"), int)) overview_summary = ((native_ops.get("overview") or {}).get("summary") or {}) upstream_count = overview_summary.get("upstream_error_count_excl_429_529") or 0 upstream_count += overview_summary.get("upstream_429_count") or 0 upstream_count += overview_summary.get("upstream_529_count") or 0 if marker_total == 0 and upstream_count > 0: runtime_event_lines = [ line for line in runtime_log_lines() if any(marker in line for marker in ( "account_temp_unschedulable", "openai.upstream_failover_switching", "openai.account_select_failed", "openai.forward_failed", )) ] lines.extend(runtime_event_lines) system_log_source["source"] = "sub2api-native-admin-ops-with-runtime-event-fallback" system_log_source["fallbackUsed"] = True system_log_source["fallbackReason"] = "native system-log index returned no policy events while the native overview reported upstream errors" system_log_source["fallbackEventCount"] = len(runtime_event_lines) effectiveness = policy_effectiveness(lines, selected if selector is not None else accounts, system_log_source.get("source"), group_id) effectiveness["sourceStatus"] = system_log_source overview_summary = ((native_ops.get("overview") or {}).get("summary") or {}) native_ops["responseHeaderTimeout"] = runtime_response_header_timeout() native_ops["requestTiming"] = native_request_timing(token, group_id, platform, overview_summary) native_ops["customerErrors"] = native_customer_error_profile(token, group_id, platform) native_ops["accountQuality"] = account_quality_scores(token, selected, group_id, native_ops, effectiveness) request_total = overview_summary.get("request_count_total") error_total = overview_summary.get("error_count_total") if isinstance(request_total, int) and isinstance(error_total, int): effectiveness["gatewayOutcomeEvidence"] = { "source": "sub2api-native-admin-ops-dashboard-overview", "requestCountTotal": request_total, "errorCountTotal": error_total, "errorRate": overview_summary.get("error_rate"), "upstreamErrorRate": overview_summary.get("upstream_error_rate"), "businessLimitedCount": overview_summary.get("business_limited_count"), "startTime": overview_summary.get("start_time"), "endTime": overview_summary.get("end_time"), "completeTrafficDenominatorGuaranteed": True, } scanned = 0 for line in lines: if "account_upstream_error" not in line and "openai.forward_failed" not in line: continue json_start = line.find("{") if json_start < 0: continue try: item = json.loads(line[json_start:]) except json.JSONDecodeError: continue account_id = item.get("account_id") if isinstance(item, dict) else None account_key = str(account_id) if account_key not in selected_ids: continue scanned += 1 at = line.split("\t", 1)[0].strip() or None current = stats[account_key] if at and (current["lastSeenAt"] is None or at > current["lastSeenAt"]): current["lastSeenAt"] = at if "account_upstream_error" in line: code = item.get("status_code") if isinstance(code, int): key = str(code) current["observedStatusCodes"][key] = current["observedStatusCodes"].get(key, 0) + 1 current["upstreamEventCount"] += 1 continue current["forwardFailureCount"] += 1 error = compact_error(item.get("error")) match = re.search(r"(?:upstream error:\\s*|status(?:_code)?[=:]\\s*)(\\d{3})", error, re.IGNORECASE) if match: key = match.group(1) current["forwardFailureStatusCodes"][key] = current["forwardFailureStatusCodes"].get(key, 0) + 1 if error and error not in current["errorSamples"] and len(current["errorSamples"]) < 3: current["errorSamples"].append(error) rows = [] aggregate = {} for item in selected: current = stats[str(item.get("id"))] observed = [{"statusCode": int(code), "count": count} for code, count in sorted(current["observedStatusCodes"].items(), key=lambda pair: int(pair[0]))] forwarded = [{"statusCode": int(code), "count": count} for code, count in sorted(current["forwardFailureStatusCodes"].items(), key=lambda pair: int(pair[0]))] for entry in observed: key = str(entry["statusCode"]) aggregate[key] = aggregate.get(key, 0) + entry["count"] current["observedStatusCodes"] = observed current["forwardFailureStatusCodes"] = forwarded analysis = ops_error_analysis(token, item, observed) has_ops_errors = analysis.get("status") == "available" and (analysis.get("opsRecordCount") or 0) > 0 if current["upstreamEventCount"] > 0 or current["forwardFailureCount"] > 0 or has_ops_errors: if PAYLOAD.get("full") is not True and analysis.get("status") == "available": analysis = { "source": analysis.get("source"), "status": analysis.get("status"), "opsRecordCount": analysis.get("opsRecordCount"), "recordsTruncated": analysis.get("recordsTruncated"), "statusBuckets": analysis.get("statusBuckets"), "causeBuckets": analysis.get("causeBuckets"), "rootCause": analysis.get("rootCause"), "errorRateEvidence": analysis.get("errorRateEvidence"), "concurrencyRecommendation": analysis.get("concurrencyRecommendation"), "runtimeConcurrency": analysis.get("runtimeConcurrency"), "samples": (analysis.get("samples") or [])[:1], } current["opsAnalysis"] = analysis if PAYLOAD.get("full") is not True: current["errorSamples"] = current["errorSamples"][:1] if current["upstreamEventCount"] > 0 or current["forwardFailureCount"] > 0 or has_ops_errors or selector is not None: rows.append(current) root_causes = {} account_roots = [] for row in rows: analysis = row.get("opsAnalysis") if not isinstance(analysis, dict) or analysis.get("status") != "available": continue for bucket in analysis.get("causeBuckets") or []: cause = bucket.get("cause") count = bucket.get("count") if isinstance(cause, str) and isinstance(count, int): root_causes[cause] = root_causes.get(cause, 0) + count root = analysis.get("rootCause") or {} account_roots.append({ "accountId": row.get("accountId"), "accountName": row.get("accountName"), "dominantCause": root.get("dominantCause"), "dominantCount": root.get("dominantCount"), "dominantShare": root.get("dominantShare"), }) ranked_causes = sorted(root_causes.items(), key=lambda pair: (-pair[1], pair[0])) native_overview = native_ops.get("overview") or {} native_overview_values = native_overview.get("summary") if isinstance(native_overview.get("summary"), dict) else native_overview native_error_rate_available = isinstance(native_overview_values.get("error_rate") if isinstance(native_overview_values, dict) else None, (int, float)) account_count_with_errors = sum( 1 for item in rows if item["upstreamEventCount"] > 0 or item["forwardFailureCount"] > 0 or (((item.get("opsAnalysis") or {}).get("opsRecordCount") or 0) > 0) ) if selector is None and PAYLOAD.get("full") is not True: for row in rows: row.pop("opsAnalysis", None) if PAYLOAD.get("full") is not True: overview = ((native_ops.get("overview") or {}).get("summary") or {}) availability = ((native_ops.get("accountAvailability") or {}).get("summary") or {}) concurrency = ((native_ops.get("concurrency") or {}).get("summary") or {}) sink = ((native_ops.get("systemLogSink") or {}).get("health") or {}) unavailable_accounts = availability.get("unavailableAccounts") or [] if selector is not None: unavailable_accounts = [item for item in unavailable_accounts if item.get("accountId") in selected_ids] native_ops = { "source": native_ops.get("source"), "overview": { "status": (native_ops.get("overview") or {}).get("status"), "healthScore": overview.get("health_score"), "requestCount": overview.get("request_count_total"), "errorCount": overview.get("error_count_total"), "errorRate": overview.get("error_rate"), "upstreamErrorRate": overview.get("upstream_error_rate"), "businessLimitedCount": overview.get("business_limited_count"), }, "responseHeaderTimeout": native_ops.get("responseHeaderTimeout"), "requestTiming": native_ops.get("requestTiming"), "customerErrors": native_ops.get("customerErrors"), "accountQuality": native_ops.get("accountQuality"), "accountAvailability": { "status": (native_ops.get("accountAvailability") or {}).get("status"), "group": availability.get("group"), "unavailableAccountCount": availability.get("unavailableAccountCount"), "unavailableAccounts": unavailable_accounts, }, "concurrency": { "status": (native_ops.get("concurrency") or {}).get("status"), "group": concurrency.get("group"), }, "systemLogSink": { "status": (native_ops.get("systemLogSink") or {}).get("status"), "queueDepth": sink.get("queue_depth"), "droppedCount": sink.get("dropped_count"), "writeFailedCount": sink.get("write_failed_count"), }, } source_status = effectiveness.get("sourceStatus") or {} compact_source_status = { "status": source_status.get("status"), "source": source_status.get("source"), "fallbackUsed": source_status.get("fallbackUsed"), "fallbackReason": source_status.get("fallbackReason"), "fallbackEventCount": source_status.get("fallbackEventCount"), "nativeIndexedEventCounts": { item.get("marker"): item.get("total") for item in source_status.get("queries", []) if isinstance(item, dict) }, } drilldown = effectiveness.get("requestDrilldown") or {} effectiveness = { "source": effectiveness.get("source"), "scope": effectiveness.get("scope"), "attribution": effectiveness.get("attribution"), "assessment": effectiveness.get("assessment"), "ruleMatches": { "tempUnschedulableEvents": effectiveness.get("tempUnschedulableEvents"), "accounts": effectiveness.get("tempUnschedulableAccounts"), }, "failover": { "events": effectiveness.get("failoverEvents"), "requests": effectiveness.get("failoverRequestCount"), "succeeded": effectiveness.get("failoverSucceededRequestCount"), "failed": effectiveness.get("failoverFailedRequestCount"), "outcomeMissing": effectiveness.get("failoverOutcomeMissingRequestCount"), }, "forwardFailure": { "events": effectiveness.get("forwardFailedEvents"), "withHttpSuccess": effectiveness.get("forwardFailureWithHttpSuccessRequestCount"), "withHttpError": effectiveness.get("forwardFailureWithHttpErrorRequestCount"), "outcomeMissing": effectiveness.get("forwardFailureOutcomeMissingRequestCount"), }, "selectFailedEvents": effectiveness.get("selectFailedEvents"), "requestIdCoverage": effectiveness.get("requestIdCoverage"), "gatewayOutcomeEvidence": effectiveness.get("gatewayOutcomeEvidence"), "requestDrilldown": {key: value for key, value in drilldown.items() if value}, "sourceStatus": compact_source_status, "disclosure": effectiveness.get("disclosure"), } if selector is not None: native_ops["accountAvailability"].pop("group", None) native_ops.pop("concurrency", None) native_ops.pop("systemLogSink", None) effectiveness["sourceStatus"] = { "source": compact_source_status.get("source"), "fallbackUsed": compact_source_status.get("fallbackUsed"), "fallbackReason": compact_source_status.get("fallbackReason"), } return { "window": {"since": PAYLOAD.get("since"), "tail": PAYLOAD.get("tail"), "matchedLogEvents": scanned}, "aggregateObservedStatusCodes": [{"statusCode": int(code), "count": count} for code, count in sorted(aggregate.items(), key=lambda pair: int(pair[0]))], "accountCountWithErrors": account_count_with_errors, "rootCauseAnalysis": { "source": "ops-upstream-errors", "scope": "account-window-for-group-members", "attribution": "The upstream-errors endpoint is account scoped. If accounts belong to multiple groups, these cause buckets are shared evidence and must not be added across groups.", "causeBuckets": [{"cause": cause, "count": count} for cause, count in ranked_causes], "dominantCause": ranked_causes[0][0] if ranked_causes else None, "dominantCount": ranked_causes[0][1] if ranked_causes else 0, "accountRoots": account_roots, "errorRateAvailable": native_error_rate_available, }, "nativeOps": native_ops, "policyEffectiveness": effectiveness, **( {"accounts": rows} if selector is not None or PAYLOAD.get("full") is True else {"accountDrilldown": "rerun runtime errors with --account for per-account evidence"} ), "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 PAYLOAD.get("kind") == "temp-unschedulable" and 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): if PAYLOAD.get("kind") == "account-name": before_name = str(detail.get("name") or "") desired_name = str(PAYLOAD.get("name") or "") return { "accountName": before_name, "accountId": detail.get("id"), "change": "noop" if before_name == desired_name else "update", "before": {"name": before_name}, "desired": {"name": desired_name}, "template": None, "_kind": "account-name", "_desiredName": desired_name, } if PAYLOAD.get("kind") == "priority": before_priority = int(detail.get("priority") or 0) desired_priority = int(PAYLOAD.get("priority")) return { "accountName": detail.get("name"), "accountId": detail.get("id"), "change": "noop" if before_priority == desired_priority else "update", "before": {"priority": before_priority}, "desired": {"priority": desired_priority}, "template": None, "_kind": "priority", "_desiredPriority": desired_priority, } 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, "_kind": "temp-unschedulable", } def write_mutation_plan(token, plan): if plan["change"] == "noop": return {"attempted": False, "status": "noop", "error": None} try: if plan["_kind"] == "priority": payload = {"priority": plan["_desiredPriority"]} elif plan["_kind"] == "account-name": payload = {"name": plan["_desiredName"]} else: payload = {"credentials": plan["_desiredCredentials"]} data_of(curl_api("PUT", f"/api/v1/admin/accounts/{plan['accountId']}", bearer=token, payload=payload), "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"]) if plan["_kind"] == "priority": actual_priority = int(actual_detail.get("priority") or 0) item["actual"] = {"priority": actual_priority} item["reconciled"] = actual_priority == plan["_desiredPriority"] elif plan["_kind"] == "account-name": actual_name = str(actual_detail.get("name") or "") item["actual"] = {"name": actual_name} item["reconciled"] = actual_name == plan["_desiredName"] else: 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() action = PAYLOAD["action"] if action == "errors" and PAYLOAD.get("allGroups") is True: groups = list_groups(token, PAYLOAD.get("platform")) overview = all_group_error_overview(token, groups) return { "ok": True, "operation": "errors", "mutation": False, "endpoint": "host-docker" if RUNTIME_MODE == "host-docker" else "k3s-service", "allGroups": overview, "valuesPrinted": False, } group_selector = PAYLOAD.get("group") group_platform = PAYLOAD.get("platform") if action in ("errors", "infrastructure") else "openai" if action in ("errors", "infrastructure") and group_selector is None: group_platform = "openai" groups = list_groups(token, group_platform) selector = str(group_selector if group_selector is not None else PAYLOAD["groupName"]) matching_groups = [item for item in groups if str(item.get("id")) == selector or item.get("name") == selector] if len(matching_groups) > 1: raise RuntimeError("group-selector-ambiguous: " + selector) group = matching_groups[0] if matching_groups else None if not isinstance(group, dict) or group.get("id") is None: raise RuntimeError("group-not-found: " + selector) effective_platform = group_platform or group.get("platform") accounts = list_group_accounts(token, group["id"], effective_platform) base = { "group": {"id": group.get("id"), "name": group.get("name"), "platform": effective_platform, "status": group.get("status")}, "templates": [template_summary(item) for item in PAYLOAD["templates"]], "endpoint": "host-docker" if RUNTIME_MODE == "host-docker" else "k3s-service", "valuesPrinted": False, } if action == "list": details = [account_detail(token, item["id"]) for item in accounts] return {**base, "ok": True, "operation": "list", "mutation": False, "accountCount": len(details), "accounts": [account_summary(item) for item in sorted(details, key=lambda value: value.get("name") or "")]} if action == "errors": errors = observed_runtime_errors(token, accounts, group["id"], effective_platform) return {**base, "ok": True, "operation": "errors", "mutation": False, "errors": errors} detail = find_account(token, accounts, str(PAYLOAD["account"])) if action in ("get", "infrastructure") else None if action == "get": return {**base, "ok": True, "operation": "get", "mutation": False, "account": account_summary(detail, True)} if action == "infrastructure": return { **base, "ok": True, "operation": "infrastructure", "mutation": False, "account": account_summary(detail, False), "infrastructure": infrastructure_snapshot(token, detail), } include_plan_rules = PAYLOAD.get("full") is True 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() except Exception as exc: result = {"ok": False, "operation": PAYLOAD.get("action"), "mutation": False, "error": str(exc), "valuesPrinted": False} print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0 if result.get("ok") else 1) PY `; }