export const remotePythonTraceScript = `def parse_log_line(line): json_start = line.find("{") if json_start < 0: return None prefix = line[:json_start].rstrip() try: item = json.loads(line[json_start:]) except Exception: return None if not isinstance(item, dict): return None at = None parts = prefix.split() if parts: at = parts[0] message = "" if len(parts) >= 4: message = " ".join(parts[3:]) elif len(parts) >= 3: message = parts[2] elif len(parts) >= 1: message = parts[-1] item["_at"] = at item["_message"] = message item["_line"] = line return item def log_time_epoch(item): at = item.get("_at") if isinstance(item, dict) else None if not isinstance(at, str) or not at: return None try: return datetime.strptime(at, "%Y-%m-%dT%H:%M:%S.%f%z").timestamp() except Exception: try: return datetime.fromisoformat(at.replace("Z", "+00:00")).timestamp() except Exception: return None def event_base(item, account_names_by_id): account_id = item.get("account_id") if isinstance(account_id, str) and account_id.isdigit(): account_id = int(account_id) account_name = account_names_by_id.get(account_id) return { "at": item.get("_at"), "message": item.get("_message"), "requestId": item.get("request_id"), "clientRequestId": item.get("client_request_id"), "path": item.get("path"), "method": item.get("method"), "model": item.get("model"), "accountId": account_id, "accountName": account_name, "statusCode": item.get("status_code"), "upstreamStatus": item.get("upstream_status"), "latencyMs": item.get("latency_ms"), } def classify_trace_event(item, account_names_by_id): message = str(item.get("_message") or "") event = event_base(item, account_names_by_id) if "content_moderation.gateway_check_start" in message: event.update({ "type": "request-start", "stream": item.get("stream"), "bodyBytes": item.get("body_bytes"), "groupId": item.get("group_id"), "groupName": item.get("group_name"), "apiKeyName": item.get("api_key_name"), }) elif "content_moderation.gateway_check_done" in message: event.update({ "type": "gateway-check", "allowed": item.get("allowed"), "blocked": item.get("blocked"), "action": item.get("action"), }) elif "openai.upstream_failover_switching" in message: event.update({ "type": "failover", "switchCount": item.get("switch_count"), "maxSwitches": item.get("max_switches"), }) elif "openai.account_select_failed" in message: event.update({ "type": "select-failed", "error": item.get("error"), "excludedAccountCount": item.get("excluded_account_count"), }) elif "openai.forward_failed" in message: event.update({ "type": "forward-failed", "error": item.get("error"), }) elif "account_upstream_error" in message: event.update({ "type": "upstream-error", "error": item.get("error"), }) elif "account_temp_unschedulable" in message: event.update({ "type": "temp-unschedulable", "until": item.get("until") or item.get("temp_unschedulable_until"), "ruleIndex": item.get("rule_index"), "matchedKeyword": item.get("matched_keyword"), "reason": item.get("reason") or item.get("error"), }) elif "http request completed" in message: event.update({ "type": "final", "clientIp": item.get("client_ip"), "protocol": item.get("protocol"), "platform": item.get("platform"), "completedAt": item.get("completed_at"), }) elif "admin account schedulable updated" in message or "account schedulable updated" in message or "/schedulable" in str(item.get("path") or ""): event.update({ "type": "admin-schedulable", "schedulable": item.get("schedulable"), }) else: event.update({"type": "other"}) return event def with_trace_phase(event, first_epoch, last_epoch): epoch = None at = event.get("at") if isinstance(event, dict) else None if isinstance(at, str) and at: epoch = log_time_epoch({"_at": at}) if epoch is None or first_epoch is None: phase = "unknown" elif epoch < first_epoch: phase = "before-request" elif last_epoch is not None and epoch > last_epoch: phase = "after-request" else: phase = "during-request" event["phase"] = phase return event def account_snapshot_from_runtime(token): try: accounts = list_accounts(token) except Exception as exc: return [], {"error": str(exc)} rows = [] for item in accounts: if not isinstance(item, dict): continue rows.append({ "accountId": item.get("id"), "accountName": item.get("name"), "schedulable": item.get("schedulable"), "status": item.get("status"), "concurrency": item.get("concurrency"), "priority": item.get("priority"), "tempUnschedulableUntil": item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil"), "tempUnschedulableSet": (item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil")) is not None or bool(item.get("temp_unschedulable_reason") or item.get("tempUnschedulableReason")), }) rows.sort(key=lambda row: (str(row.get("accountName") or ""), int(row.get("accountId") or 0))) return rows, None def trace_reason(events, final_event): failovers = [item for item in events if item.get("type") == "failover"] select_failures = [item for item in events if item.get("type") == "select-failed"] upstream_errors = [item for item in events if item.get("type") == "upstream-error"] forward_failures = [item for item in events if item.get("type") == "forward-failed"] if failover_budget_exhausted(failovers, final_event): return "failover-budget-exhausted" if failovers and select_failures: return "failover-attempted-no-candidate" if forward_failures and isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") < 400: return "forward-failed-http-success" if forward_failures: return "forward-failed" if failovers: return "failover-attempted" if select_failures: return "account-select-failed" if upstream_errors: return "upstream-error" if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") >= 400: return "final-http-error" if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int): return "completed" return "unknown" def failover_budget_exhausted(failovers, final_event): if not failovers or not isinstance(final_event, dict): return False last = failovers[-1] switch_count = last.get("switchCount") max_switches = last.get("maxSwitches") final_status = final_event.get("statusCode") return ( isinstance(switch_count, int) and isinstance(max_switches, int) and max_switches > 0 and switch_count >= max_switches and isinstance(final_status, int) and final_status >= 500 ) def trace_untried_schedulable_accounts(failovers, final_event, account_snapshot): if not failover_budget_exhausted(failovers, final_event): return [] tried = set() for item in failovers: account_id = item.get("accountId") if isinstance(account_id, int): tried.add(account_id) final_account = final_event.get("accountId") if isinstance(final_event, dict) else None if isinstance(final_account, int): tried.add(final_account) result = [] for item in account_snapshot: account_id = item.get("accountId") if not isinstance(account_id, int) or account_id in tried: continue if item.get("schedulable") is True and item.get("status") == "active" and item.get("tempUnschedulableSet") is not True: result.append({ "accountId": account_id, "accountName": item.get("accountName"), "priority": item.get("priority"), "concurrency": item.get("concurrency"), }) return result def ops_start_time_for_since(since): if not isinstance(since, str): return None match = re.fullmatch(r"\s*(\d+)\s*([smhd])\s*", since.lower()) if match is None: return None value = int(match.group(1)) multiplier = {"s": 1, "m": 60, "h": 3600, "d": 86400}[match.group(2)] return (datetime.now(timezone.utc) - timedelta(seconds=value * multiplier)).strftime("%Y-%m-%dT%H:%M:%SZ") def run_trace(): payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {} request_id = payload.get("requestId") since = payload.get("since") or "24h" tail = int(payload.get("tail") or 20000) context_seconds = int(payload.get("contextSeconds") or 300) show_lines = bool(payload.get("showLines")) if not isinstance(request_id, str) or not request_id: raise RuntimeError("trace payload missing requestId") admin_email, token, admin_compliance = login() request_query = quote(request_id, safe="") native_request_status = {"status": "unavailable"} native_request = None try: request_data = ensure_success( curl_api("GET", f"/api/v1/admin/ops/requests?request_id={request_query}&kind=all&page=1&page_size=100", bearer=token), "get ops request details", ) request_items = extract_items(request_data) native_request = next((item for item in request_items if isinstance(item, dict) and item.get("request_id") == request_id), None) native_request_status = { "status": "available", "matched": native_request is not None, "returned": len(request_items), "total": request_data.get("total", len(request_items)) if isinstance(request_data, dict) else len(request_items), } if native_request is None: ops_start_time = ops_start_time_for_since(since) error_query = f"/api/v1/admin/ops/request-errors?q={request_query}&view=all&page=1&page_size=20" if isinstance(ops_start_time, str): error_query += "&start_time=" + quote(ops_start_time, safe="") error_data = ensure_success( curl_api("GET", error_query, bearer=token), "get ops request error fallback", ) error_items = extract_items(error_data) native_request = next((item for item in error_items if isinstance(item, dict) and item.get("request_id") == request_id), None) native_request_status.update({ "fallback": "sub2api-native-admin-ops-request-errors", "fallbackStartTime": ops_start_time, "fallbackReturned": len(error_items), "fallbackTotal": error_data.get("total", len(error_items)) if isinstance(error_data, dict) else len(error_items), }) if isinstance(native_request, dict): native_request = dict(native_request) native_request["kind"] = "error" error_id = native_request.get("id") if isinstance(error_id, int): try: detail = ensure_success(curl_api("GET", f"/api/v1/admin/ops/request-errors/{error_id}", bearer=token), "get ops request error detail") if isinstance(detail, dict): native_request.update(detail) except Exception: pass native_request_status.update({"matched": True, "source": "sub2api-native-admin-ops-request-errors"}) except Exception as exc: native_request_status = {"status": "unavailable", "reason": text(str(exc), 500)} native_system_log_status = {"status": "unavailable"} native_matched = [] try: system_log_data = ensure_success( curl_api("GET", f"/api/v1/admin/ops/system-logs?request_id={request_query}&page=1&page_size=200", bearer=token), "get ops request system logs", ) system_log_items = extract_items(system_log_data) for row in system_log_items: if not isinstance(row, dict): continue parsed = dict(row.get("extra") or {}) if isinstance(row.get("extra"), dict) else {} for source_key in ("request_id", "client_request_id", "account_id", "platform", "model"): if parsed.get(source_key) is None and row.get(source_key) is not None: parsed[source_key] = row.get(source_key) parsed["_at"] = row.get("created_at") parsed["_message"] = row.get("message") parsed["_line"] = json.dumps({ "created_at": row.get("created_at"), "message": row.get("message"), "request_id": row.get("request_id"), "account_id": row.get("account_id"), "platform": row.get("platform"), "model": row.get("model"), "extra": parsed, }, ensure_ascii=False, default=str) native_matched.append(parsed) native_system_log_status = { "status": "available", "returned": len(system_log_items), "total": system_log_data.get("total", len(system_log_items)) if isinstance(system_log_data, dict) else len(system_log_items), } except Exception as exc: native_system_log_status = {"status": "unavailable", "reason": text(str(exc), 500)} account_snapshot, account_snapshot_error = account_snapshot_from_runtime(token) account_names_by_id = {} for row in account_snapshot: account_id = row.get("accountId") if isinstance(account_id, str) and account_id.isdigit(): account_id = int(account_id) if isinstance(account_id, int) and isinstance(row.get("accountName"), str): account_names_by_id[account_id] = row.get("accountName") proc = runtime_logs(since, tail) log_bytes = proc.stdout + b"\\n" + proc.stderr stdout = log_bytes.decode("utf-8", errors="replace") parsed_lines = [] runtime_matched = [] for line in stdout.splitlines(): parsed = parse_log_line(line) if parsed is None: continue parsed_lines.append(parsed) if request_id in line: runtime_matched.append(parsed) matched = native_matched if native_matched else runtime_matched matched.sort(key=lambda item: (log_time_epoch(item) is None, log_time_epoch(item) or 0)) trace_event_source = "sub2api-native-admin-ops-system-logs" if native_matched else "runtime-logs-fallback" first_epoch = None last_epoch = None for item in matched: epoch = log_time_epoch(item) if epoch is None: continue first_epoch = epoch if first_epoch is None else min(first_epoch, epoch) last_epoch = epoch if last_epoch is None else max(last_epoch, epoch) window_lines = [] if first_epoch is not None: start_epoch = first_epoch - context_seconds end_epoch = (last_epoch if last_epoch is not None else first_epoch) + context_seconds for item in parsed_lines: epoch = log_time_epoch(item) if epoch is not None and start_epoch <= epoch <= end_epoch: window_lines.append(item) else: window_lines = matched events = [classify_trace_event(item, account_names_by_id) for item in matched] request_start = next((item for item in events if item.get("type") == "request-start"), None) if isinstance(native_request, dict): if not isinstance(request_start, dict): request_start = { "type": "request-start", "at": native_request.get("created_at"), "requestId": native_request.get("request_id"), "path": native_request.get("request_path") or native_request.get("inbound_endpoint"), "model": native_request.get("model"), "accountId": native_request.get("account_id"), "accountName": account_names_by_id.get(native_request.get("account_id")), "stream": native_request.get("stream"), "source": "sub2api-native-admin-ops-requests", } else: if request_start.get("stream") is None: request_start["stream"] = native_request.get("stream") if not request_start.get("model"): request_start["model"] = native_request.get("model") final_event = next((item for item in reversed(events) if item.get("type") == "final"), None) if not isinstance(final_event, dict) and isinstance(native_request, dict) and isinstance(native_request.get("status_code"), int): final_event = { "type": "final", "at": native_request.get("created_at"), "requestId": native_request.get("request_id"), "path": native_request.get("request_path") or native_request.get("inbound_endpoint"), "model": native_request.get("model"), "accountId": native_request.get("account_id"), "accountName": account_names_by_id.get(native_request.get("account_id")), "statusCode": native_request.get("status_code"), "latencyMs": native_request.get("duration_ms"), "source": "sub2api-native-admin-ops-requests", } reference_at = request_start.get("at") if isinstance(request_start, dict) else None reference_epoch = log_time_epoch({"_at": reference_at}) if isinstance(reference_at, str) else first_epoch reference_source = "request-start" if isinstance(reference_at, str) and reference_epoch is not None else "first-matched-event" native_duration_ms = native_request.get("duration_ms") if isinstance(native_request, dict) else None if not isinstance(native_duration_ms, int) and isinstance(final_event, dict): native_duration_ms = final_event.get("latencyMs") if ( isinstance(request_start, dict) and request_start.get("source") == "sub2api-native-admin-ops-requests" and isinstance(native_duration_ms, int) and native_duration_ms >= 0 and reference_epoch is not None ): reference_epoch -= native_duration_ms / 1000 reference_at = datetime.fromtimestamp(reference_epoch, timezone.utc).isoformat().replace("+00:00", "Z") reference_source = "native-completion-minus-duration" for event in events: epoch = log_time_epoch({"_at": event.get("at")}) if isinstance(event.get("at"), str) else None event["elapsedMs"] = round((epoch - reference_epoch) * 1000) if epoch is not None and reference_epoch is not None else None failovers = [item for item in events if item.get("type") == "failover"] select_failures = [item for item in events if item.get("type") == "select-failed"] upstream_errors = [item for item in events if item.get("type") == "upstream-error"] forward_failures = [item for item in events if item.get("type") == "forward-failed"] temp_unsched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if "account_temp_unschedulable" in str(item.get("_message") or "")] admin_sched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if ("schedulable" in str(item.get("_message") or "") or "/schedulable" in str(item.get("path") or ""))] window_events = [classify_trace_event(item, account_names_by_id) for item in window_lines] final_errors = [item for item in window_events if item.get("type") == "final" and isinstance(item.get("statusCode"), int) and item.get("statusCode") >= 400] window_failovers = [item for item in window_events if item.get("type") == "failover"] window_select_failures = [item for item in window_events if item.get("type") == "select-failed"] window_forward_failures = [item for item in window_events if item.get("type") == "forward-failed"] untried_schedulable_accounts = trace_untried_schedulable_accounts(failovers, final_event or {}, account_snapshot) first_failover_elapsed = failovers[0].get("elapsedMs") if failovers else None first_select_failed_elapsed = select_failures[0].get("elapsedMs") if select_failures else None final_epoch = log_time_epoch({"_at": final_event.get("at")}) if isinstance(final_event, dict) and isinstance(final_event.get("at"), str) else None final_elapsed = round((final_epoch - reference_epoch) * 1000) if final_epoch is not None and reference_epoch is not None else None failover_to_final = final_elapsed - first_failover_elapsed if isinstance(final_elapsed, int) and isinstance(first_failover_elapsed, int) else None reason = trace_reason(events, final_event) if not matched and not isinstance(native_request, dict): outcome = "not-found" elif isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") < 400: outcome = "degraded" if forward_failures else ("succeeded-with-failover" if failovers else "succeeded") elif isinstance(final_event, dict): outcome = "failed" else: outcome = "incomplete" return { "ok": proc.returncode == 0 and (len(matched) > 0 or isinstance(native_request, dict)), "mode": "trace", "targetId": TARGET_ID, "runtimeMode": RUNTIME_MODE, "namespace": NAMESPACE, "serviceDns": SERVICE_DNS, "appPod": APP_POD, "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance}, "requestId": request_id, "eventSource": trace_event_source, "summary": { "outcome": outcome, "reason": reason, "eventCount": len(events), "matchedLineCount": len(matched), "firstAt": events[0].get("at") if events else None, "lastAt": events[-1].get("at") if events else None, }, "window": { "since": since, "tail": tail, "beforeSeconds": context_seconds, "afterSeconds": context_seconds, "lineCount": len(window_lines), }, "request": request_start or {}, "final": final_event or {}, "nativeOps": { "requestDetails": native_request_status, "request": native_request, "systemLogs": native_system_log_status, }, "events": events, "failovers": failovers, "selectFailures": select_failures, "upstreamErrors": upstream_errors, "forwardFailures": forward_failures, "tempUnschedulable": temp_unsched, "adminSchedulable": admin_sched[-20:], "windowStats": { "matchedLines": len(matched), "eventCount": len(window_events), "finalErrorCount": len(final_errors), "failoverCount": len(window_failovers), "selectFailedCount": len(window_select_failures), "forwardFailedCount": len(window_forward_failures), "matchedForwardFailedCount": len(forward_failures), "tempUnschedulableCount": len(temp_unsched), "adminSchedulableCount": len(admin_sched), }, "diagnostics": { "failoverBudgetExhausted": failover_budget_exhausted(failovers, final_event or {}), "untriedSchedulableAccounts": untried_schedulable_accounts, "timing": { "referenceAt": reference_at if isinstance(reference_at, str) else (events[0].get("at") if events else None), "referenceSource": reference_source, "firstFailoverElapsedMs": first_failover_elapsed, "firstSelectFailedElapsedMs": first_select_failed_elapsed, "finalElapsedMs": final_elapsed, "failoverToFinalMs": failover_to_final, "clientDeadlineAvailable": False, "clientDeadlineRemainingMs": None, "attribution": "Elapsed values come from indexed event timestamps. The client deadline is not present in Sub2API Ops records, so remaining request budget cannot be calculated.", }, }, "accountSnapshot": account_snapshot, "accountSnapshotError": account_snapshot_error, "rawLines": [{"line": item.get("_line")} for item in matched[-30:]] if show_lines else [], "showLines": show_lines, "logs": { "source": trace_event_source, "contextSource": f"host-docker:{HOST_DOCKER_APP_CONTAINER}" if RUNTIME_MODE == "host-docker" else f"k3s:{NAMESPACE}/deployment/sub2api", "exitCode": proc.returncode, "stderrTail": text(proc.stderr, 1000) if proc.returncode != 0 else "", "stdoutLineCount": len(stdout.splitlines()), }, "valuesPrinted": False, } `;