feat: improve sub2api recent error analysis

2026-07-14 07:07:48 +02:00
parent b9ede0471e
commit da37a96424
5 changed files with 241 additions and 4 deletions
@@ -3006,6 +3006,16 @@ def trace_untried_schedulable_accounts(failovers, final_event, account_snapshot)
            })
    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")
@@ -3032,6 +3042,35 @@ def run_trace():
            "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)}

@@ -3119,7 +3158,7 @@ def run_trace():
                "type": "request-start",
                "at": native_request.get("created_at"),
                "requestId": native_request.get("request_id"),
-                "path": None,
+                "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")),
@@ -3137,7 +3176,7 @@ def run_trace():
            "type": "final",
            "at": native_request.get("created_at"),
            "requestId": native_request.get("request_id"),
-            "path": None,
+            "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")),
@@ -3182,7 +3221,7 @@ def run_trace():
    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:
+    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")
@@ -3191,7 +3230,7 @@ def run_trace():
    else:
        outcome = "incomplete"
    return {
-        "ok": proc.returncode == 0 and len(matched) > 0,
+        "ok": proc.returncode == 0 and (len(matched) > 0 or isinstance(native_request, dict)),
        "mode": "trace",
        "targetId": TARGET_ID,
        "runtimeMode": RUNTIME_MODE,
@@ -172,6 +172,7 @@ function compactRuntimeErrors(value: unknown, options: RuntimeOptions): unknown
        overview: nativeOps.overview,
        responseHeaderTimeout: nativeOps.responseHeaderTimeout,
        requestTiming: nativeOps.requestTiming,
+        customerErrors: nativeOps.customerErrors,
        accountAvailability: {
          status: availability.status,
          totalAccounts: availabilityGroup.total_accounts,
@@ -325,6 +326,7 @@ function renderRuntimeErrors(lines: string[], runtime: Record<string, unknown>,
  const timing = runtimeRecord(nativeOps.requestTiming) ?? {};
  const ttft = runtimeRecord(timing.ttft) ?? {};
  const duration = runtimeRecord(timing.duration) ?? {};
+  const customerErrors = runtimeRecord(nativeOps.customerErrors) ?? {};
  lines.push(`SUB2API RUNTIME ERRORS ok=${runtime.ok === true ? "true" : "false"} scope=group since=${options.since}`);
  lines.push(renderTable([
    ["GROUP", "ID", "PLATFORM", "REQUESTS", "CLIENT_ERR", "ERR_RATE", "UP_RATE", "BUSINESS_LIMIT", "HEALTH"],
@@ -351,6 +353,32 @@ function renderRuntimeErrors(lines: string[], runtime: Record<string, unknown>,
    ]),
  ]));
  if (typeof timing.attribution === "string") lines.push(`NOTE: ${timing.attribution}`);
+  const modelTiming = runtimeRecords(timing.modelTiming);
+  if (modelTiming.length > 0) lines.push("", "MODEL TIMING", renderTable([
+    ["MODEL", "REQUESTS", "FIRST_TOKEN_ROWS", "COVERAGE", "AVG_TTFT_MS", "AVG_DURATION_MS", "AVG_TOKENS_S"],
+    ...modelTiming.map((item) => [runtimeShort(item.model, 32), runtimeText(item.requestCount), runtimeText(item.requestsWithFirstToken), runtimeRate(item.firstTokenCoverage), runtimeText(item.avgFirstTokenMs), runtimeText(item.avgDurationMs), runtimeText(item.avgTokensPerSec)]),
+  ]));
+  const errorModels = runtimeRecords(customerErrors.modelBuckets);
+  const errorAccounts = runtimeRecords(customerErrors.accountBuckets);
+  const errorDimensions = [
+    ...runtimeRecords(customerErrors.statusBuckets).map((item) => ["status", runtimeText(item.value), runtimeText(item.count)]),
+    ...runtimeRecords(customerErrors.streamBuckets).map((item) => ["mode", runtimeText(item.value), runtimeText(item.count)]),
+    ...runtimeRecords(customerErrors.phaseBuckets).map((item) => ["phase", runtimeText(item.value), runtimeText(item.count)]),
+    ...runtimeRecords(customerErrors.endpointBuckets).map((item) => ["endpoint", runtimeText(item.value), runtimeText(item.count)]),
+    ...runtimeRecords(customerErrors.pathBuckets).map((item) => ["path", runtimeText(item.value), runtimeText(item.count)]),
+  ];
+  if (errorModels.length > 0) lines.push("", `CUSTOMER ERRORS source=${runtimeText(customerErrors.source)} total=${runtimeText(customerErrors.total)}`, renderTable([
+    ["MODEL", "COUNT"], ...errorModels.map((item) => [runtimeShort(item.value, 40), runtimeText(item.count)]),
+  ]));
+  if (errorAccounts.length > 0) lines.push(renderTable([
+    ["ACCOUNT_ID", "COUNT"], ...errorAccounts.map((item) => [runtimeText(item.value), runtimeText(item.count)]),
+  ]));
+  if (errorDimensions.length > 0) lines.push(renderTable([["DIMENSION", "VALUE", "COUNT"], ...errorDimensions]));
+  const errorSamples = runtimeRecords(customerErrors.samples);
+  if (errorSamples.length > 0) lines.push("CUSTOMER ERROR INDEX", renderTable([
+    ["REQUEST_ID", "MODEL", "STATUS", "ACCOUNT", "MODE", "PATH", "MESSAGE"],
+    ...errorSamples.map((item) => [runtimeText(item.requestId), runtimeShort(item.model, 26), runtimeText(item.statusCode), runtimeText(item.accountId), runtimeText(item.mode), runtimeShort(item.path, 34), runtimeShort(item.message, 48)]),
+  ]));
  lines.push(renderTable([
    ["ACCOUNTS", "AVAILABLE", "RATE_LIMIT", "ERROR", "IN_USE", "CAPACITY", "QUEUE"],
    [
@@ -1206,6 +1234,32 @@ def native_request_timing(token, group_id, platform, overview):
            "errorDurationAtLeastCount": error_counts.get(seconds * 1000),
            "assessment": "observed-ttft-p99-plus-15s-rounded" if seconds == observed_floor else "ttft-p99-within-candidate" if isinstance(headroom, int) and headroom >= 0 else "ttft-p99-exceeds-candidate" if isinstance(headroom, int) else "insufficient-ttft-evidence",
        })
+    since = str(PAYLOAD.get("since") or "")
+    token_range = {"30m": "30m", "60m": "1h", "1h": "1h", "24h": "1d", "1d": "1d"}.get(since)
+    model_timing = []
+    model_timing_status = "unsupported-window"
+    if token_range:
+        response = curl_api("GET", f"/api/v1/admin/ops/dashboard/openai-token-stats?group_id={group_id}&time_range={token_range}&top_n=100{platform_query}", bearer=token)
+        try:
+            data = data_of(response, "get OpenAI model token stats")
+            rows = items_of(data)
+            for row in rows:
+                if not isinstance(row, dict):
+                    continue
+                request_count = row.get("request_count")
+                first_count = row.get("requests_with_first_token")
+                model_timing.append({
+                    "model": row.get("model"), "requestCount": request_count,
+                    "requestsWithFirstToken": first_count,
+                    "firstTokenCoverage": round(first_count / request_count, 6) if isinstance(first_count, int) and isinstance(request_count, int) and request_count > 0 else None,
+                    "avgFirstTokenMs": row.get("avg_first_token_ms"), "avgDurationMs": row.get("avg_duration_ms"),
+                    "avgTokensPerSec": row.get("avg_tokens_per_sec"),
+                })
+            model_timing.sort(key=lambda item: (-(item.get("avgFirstTokenMs") or -1), str(item.get("model") or "")))
+            model_timing_status = "available"
+        except Exception as exc:
+            model_timing_status = "unavailable"
+            reason = compact_error(str(exc))
    return {
        "status": status,
        "reason": reason,
@@ -1214,10 +1268,71 @@ def native_request_timing(token, group_id, platform, overview):
        "duration": duration,
        "observedFloorSeconds": observed_floor,
        "errorDurationRecordCount": error_duration_record_count,
+        "modelTimingStatus": model_timing_status,
+        "modelTimingWindow": token_range,
+        "modelTiming": model_timing[:20 if PAYLOAD.get("full") is True else 8],
        "candidates": candidates,
        "attribution": "TTFT covers successful streaming requests with first_token_ms. Total and error duration are end-to-end request durations, not response-header wait or guaranteed failover lead time. The observed floor is TTFT P99 plus 15 seconds rounded up to 15 seconds; it is not a deployment recommendation. A dash in ERROR_DUR>= means error duration coverage is unavailable.",
    }

+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 = []
+    seen = set()
+    for row in rows:
+        if not isinstance(row, dict) or not row.get("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"))
+        if key in seen:
+            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")),
+        })
+        if len(samples) >= (30 if PAYLOAD.get("full") is True else 15):
+            break
+    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"),
+        "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")),
+        "samples": samples,
+    }
+
 def all_group_error_overview(token, groups):
    page_size = 20
    cursor = int(PAYLOAD.get("pageToken") or 0)
@@ -1636,6 +1751,7 @@ def observed_runtime_errors(token, accounts, group_id, platform=None):
    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)
    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):
@@ -1771,6 +1887,7 @@ def observed_runtime_errors(token, accounts, group_id, platform=None):
            },
            "responseHeaderTimeout": native_ops.get("responseHeaderTimeout"),
            "requestTiming": native_ops.get("requestTiming"),
+            "customerErrors": native_ops.get("customerErrors"),
            "accountAvailability": {
                "status": (native_ops.get("accountAvailability") or {}).get("status"),
                "group": availability.get("group"),