test: validate Sub2API 2xx reclassification

scripts/src/platform-infra-sub2api-codex.ts

@@ -2517,6 +2517,343 @@ def summarize_temp_unschedulable_rules(rules):
         "hasDescription": bool(rule.get("description")),
     } for rule in rules]
 
+def success_body_reclassification_requirement():
+    for name in sorted(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE):
+        expected = normalize_temp_unschedulable_credentials(EXPECTED_ACCOUNT_TEMP_UNSCHEDULABLE[name])
+        for rule in expected["rules"]:
+            error_code = rule.get("error_code")
+            keywords = rule.get("keywords") or []
+            if isinstance(error_code, int) and 200 <= error_code < 300 and keywords:
+                return {
+                    "required": True,
+                    "sourceAccountName": name,
+                    "statusCode": error_code,
+                    "keywords": keywords,
+                    "probeKeyword": keywords[0],
+                    "durationMinutes": rule.get("duration_minutes"),
+                }
+    return {
+        "required": False,
+        "sourceAccountName": None,
+        "statusCode": None,
+        "keywords": [],
+        "probeKeyword": None,
+        "durationMinutes": None,
+    }
+
+def delete_probe_resource(token, method, path, label):
+    if not path:
+        return {"label": label, "ok": True, "skipped": True}
+    resp = curl_api(method, path, bearer=token)
+    ok = resp.get("ok") is True or resp.get("httpStatus") in (404, 410)
+    return {
+        "label": label,
+        "ok": ok,
+        "method": method,
+        "path": path,
+        "httpStatus": resp.get("httpStatus"),
+        "transportExitCode": resp.get("transportExitCode"),
+        "bodyPreview": "" if ok else text(resp.get("body", ""), 500),
+        "valuesPrinted": False,
+    }
+
+def launch_success_body_mock_upstream(status_code, body_text):
+    port = 28000 + secrets.randbelow(2000)
+    body_b64 = base64.b64encode(body_text.encode("utf-8")).decode("ascii")
+    script = r'''
+set -eu
+port="$1"
+status="$2"
+body_b64="$3"
+body="$(printf "%s" "$body_b64" | base64 -d)"
+length="$(printf "%s" "$body" | wc -c | tr -d " ")"
+{ printf "HTTP/1.1 %s OK\r\nContent-Type: application/json\r\nContent-Length: %s\r\nConnection: close\r\n\r\n" "$status" "$length"; printf "%s" "$body"; } | nc -l -p "$port" -w 5
+'''
+    proc = subprocess.Popen([
+        "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
+        "--", "sh", "-c", script, "sh", str(port), str(status_code), body_b64,
+    ], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    time.sleep(0.35)
+    if proc.poll() is None:
+        return port, proc, None
+    stdout, stderr = proc.communicate(timeout=2)
+    return None, None, {
+        "ok": False,
+        "error": "mock-upstream-exited-before-probe",
+        "exitCode": proc.returncode,
+        "stdoutTail": text(stdout, 1000),
+        "stderrTail": text(stderr, 1000),
+    }
+
+def finish_mock_upstream(proc):
+    if proc is None:
+        return None
+    timed_out = False
+    try:
+        stdout, stderr = proc.communicate(timeout=2)
+    except subprocess.TimeoutExpired:
+        timed_out = True
+        proc.kill()
+        stdout, stderr = proc.communicate(timeout=2)
+    return {
+        "exitCode": proc.returncode,
+        "timedOut": timed_out,
+        "stdoutTail": text(stdout, 1000),
+        "stderrTail": text(stderr, 1000),
+    }
+
+def create_success_body_probe_resources(token, base_url, requirement):
+    stamp = str(int(time.time() * 1000))
+    suffix = stamp + "-" + "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(6))
+    group_payload_obj = group_payload()
+    group_payload_obj["name"] = "unidesk-probe-2xx-body-" + suffix
+    group_payload_obj["description"] = "UniDesk validate probe for OpenAI 2xx success-body reclassification."
+    group_id = None
+    account_id = None
+    api_key_id = None
+    try:
+        group = ensure_success(curl_api("POST", "/api/v1/admin/groups", bearer=token, payload=group_payload_obj), "create 2xx success-body probe group")
+        group_id = group.get("id") if isinstance(group, dict) else None
+        if group_id is None:
+            raise RuntimeError("2xx success-body probe group id missing")
+
+        account_payload_obj = {
+            "name": "unidesk-probe-2xx-body-" + suffix,
+            "notes": "Temporary UniDesk validate probe account; safe to delete.",
+            "platform": "openai",
+            "type": "apikey",
+            "credentials": {
+                "api_key": "sk-unidesk-probe-upstream",
+                "base_url": base_url,
+                "temp_unschedulable_enabled": True,
+                "temp_unschedulable_rules": [{
+                    "error_code": requirement["statusCode"],
+                    "keywords": [requirement["probeKeyword"]],
+                    "duration_minutes": 1,
+                    "description": "UniDesk runtime capability probe for OpenAI 2xx success-body reclassification.",
+                }],
+            },
+            "extra": {
+                "openai_responses_mode": "force_responses",
+                "unidesk_probe": "success_body_reclassification",
+            },
+            "concurrency": 1,
+            "priority": 0,
+            "rate_multiplier": 1,
+            "load_factor": 1,
+            "group_ids": [group_id],
+            "confirm_mixed_channel_risk": True,
+        }
+        account = ensure_success(curl_api("POST", "/api/v1/admin/accounts", bearer=token, payload=account_payload_obj), "create 2xx success-body probe account")
+        account_id = account.get("id") if isinstance(account, dict) else None
+        if account_id is None:
+            raise RuntimeError("2xx success-body probe account id missing")
+
+        api_key = "sk-unidesk-probe-" + "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(36))
+        api_key_obj = ensure_success(curl_api("POST", "/api/v1/keys", bearer=token, payload={
+            "name": "unidesk-probe-2xx-body-" + suffix,
+            "group_id": group_id,
+            "custom_key": api_key,
+            "quota": 0,
+            "rate_limit_5h": 0,
+            "rate_limit_1d": 0,
+            "rate_limit_7d": 0,
+        }), "create 2xx success-body probe API key")
+        api_key_id = api_key_obj.get("id") if isinstance(api_key_obj, dict) else None
+        return {
+            "groupId": group_id,
+            "groupName": group_payload_obj["name"],
+            "accountId": account_id,
+            "accountName": account_payload_obj["name"],
+            "apiKeyId": api_key_id,
+            "apiKey": api_key,
+            "keyPreview": api_key_preview(api_key),
+            "valuesPrinted": False,
+        }
+    except Exception:
+        if api_key_id is not None:
+            delete_probe_resource(token, "DELETE", f"/api/v1/keys/{api_key_id}", "api-key")
+        if account_id is not None:
+            delete_probe_resource(token, "DELETE", f"/api/v1/admin/accounts/{account_id}", "account")
+        if group_id is not None:
+            delete_probe_resource(token, "DELETE", f"/api/v1/admin/groups/{group_id}", "group")
+        raise
+
+def gateway_success_body_probe_request(api_key, request_id):
+    payload = {
+        "model": RESPONSES_SMOKE_MODEL,
+        "input": "Reply exactly: success-body probe",
+        "stream": False,
+        "store": False,
+        "max_output_tokens": 8,
+    }
+    body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+    script = r'''
+set -eu
+token="$1"
+request_id="$2"
+tmp="$(mktemp)"
+trap 'rm -f "$tmp"' EXIT
+cat > "$tmp"
+curl -sS -w '\\n__HTTP_CODE__:%{http_code}' -X POST \
+  -H "Authorization: Bearer $token" \
+  -H 'Content-Type: application/json' \
+  -H "X-Request-ID: $request_id" \
+  -H "OpenAI-Client-Request-ID: $request_id" \
+  --data-binary @"$tmp" \
+  http://127.0.0.1:8080/v1/responses
+'''
+    proc = run([
+        "kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD,
+        "--", "sh", "-c", script, "sh", api_key, request_id,
+    ], body)
+    return parse_curl_output(proc)
+
+def account_temp_unschedulable_probe_state(token, account_id):
+    detail = ensure_success(curl_api("GET", f"/api/v1/admin/accounts/{account_id}", bearer=token), "get 2xx success-body probe account")
+    if not isinstance(detail, dict):
+        return {
+            "accountId": account_id,
+            "status": None,
+            "schedulable": None,
+            "tempUnschedulableUntil": None,
+            "tempUnschedulableReasonPreview": "",
+            "tempUnschedulableSet": False,
+        }
+    until = detail.get("temp_unschedulable_until") or detail.get("tempUnschedulableUntil")
+    reason = detail.get("temp_unschedulable_reason") or detail.get("tempUnschedulableReason") or ""
+    return {
+        "accountId": account_id,
+        "status": detail.get("status"),
+        "schedulable": detail.get("schedulable"),
+        "tempUnschedulableUntil": until,
+        "tempUnschedulableReasonPreview": text(str(reason), 500) if reason else "",
+        "tempUnschedulableSet": until is not None or bool(reason),
+    }
+
+def validate_success_body_reclassification(token):
+    requirement = success_body_reclassification_requirement()
+    if not requirement["required"]:
+        return {
+            "ok": True,
+            "required": False,
+            "capability": "openai-2xx-success-body-temp-unschedulable-failover",
+            "outcome": "not-required-by-yaml",
+            "valuesPrinted": False,
+        }
+
+    resources = None
+    cleanup = []
+    mock_proc = None
+    try:
+        keyword = requirement["probeKeyword"]
+        upstream_body = json.dumps({
+            "id": "resp_unidesk_success_body_probe",
+            "object": "response",
+            "created_at": int(time.time()),
+            "status": "completed",
+            "model": RESPONSES_SMOKE_MODEL,
+            "output": [{
+                "type": "message",
+                "role": "assistant",
+                "content": [{"type": "output_text", "text": keyword}],
+            }],
+            "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
+        }, separators=(",", ":"))
+        port, mock_proc, mock_error = launch_success_body_mock_upstream(requirement["statusCode"], upstream_body)
+        if mock_error is not None:
+            return {
+                "ok": False,
+                "required": True,
+                "capability": "openai-2xx-success-body-temp-unschedulable-failover",
+                "outcome": "probe-infrastructure-failed",
+                "requirement": requirement,
+                "mock": mock_error,
+                "valuesPrinted": False,
+            }
+        resources = create_success_body_probe_resources(token, f"http://127.0.0.1:{port}", requirement)
+        request_id = "unidesk-2xx-body-probe-" + str(int(time.time() * 1000))
+        started = time.time()
+        response = gateway_success_body_probe_request(resources["apiKey"], request_id)
+        mock_result = finish_mock_upstream(mock_proc)
+        mock_proc = None
+        state = account_temp_unschedulable_probe_state(token, resources["accountId"])
+        evidence = request_log_evidence(request_id)
+        supported = response.get("ok") is not True and state.get("tempUnschedulableSet") is True
+        return {
+            "ok": supported,
+            "required": True,
+            "capability": "openai-2xx-success-body-temp-unschedulable-failover",
+            "outcome": "supported" if supported else "unsupported-runtime-image",
+            "requirement": requirement,
+            "probe": {
+                "requestId": request_id,
+                "durationMs": int((time.time() - started) * 1000),
+                "httpStatus": response.get("httpStatus"),
+                "transportExitCode": response.get("transportExitCode"),
+                "responseOk": response.get("ok"),
+                "bodyPreview": text(response.get("body", ""), 800),
+                "stderr": response.get("stderr", ""),
+                "accountState": state,
+                "logEvidence": evidence,
+            },
+            "mock": mock_result,
+            "resources": {
+                "groupId": resources["groupId"],
+                "accountId": resources["accountId"],
+                "apiKeyId": resources["apiKeyId"],
+                "keyPreview": resources["keyPreview"],
+                "valuesPrinted": False,
+            },
+            "cleanup": cleanup,
+            "message": "Sub2API image must reclassify matching 2xx OpenAI bodies into failover/temp-unschedulable before statusCode=200 YAML rules are effective." if not supported else "Matching 2xx OpenAI bodies are reclassified into account failover/temp-unschedulable.",
+            "valuesPrinted": False,
+        }
+    except Exception as exc:
+        return {
+            "ok": False,
+            "required": True,
+            "capability": "openai-2xx-success-body-temp-unschedulable-failover",
+            "outcome": "probe-failed",
+            "requirement": requirement,
+            "error": str(exc),
+            "cleanup": cleanup,
+            "valuesPrinted": False,
+        }
+    finally:
+        if mock_proc is not None:
+            _ = finish_mock_upstream(mock_proc)
+        if resources is not None:
+            cleanup.append(delete_probe_resource(token, "DELETE", f"/api/v1/keys/{resources['apiKeyId']}" if resources.get("apiKeyId") is not None else "", "api-key"))
+            cleanup.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/accounts/{resources['accountId']}", "account"))
+            cleanup.append(delete_probe_resource(token, "DELETE", f"/api/v1/admin/groups/{resources['groupId']}", "group"))
+
+def validate_runtime_capabilities(token):
+    success_body = validate_success_body_reclassification(token)
+    return {
+        "ok": success_body.get("ok") is True,
+        "runtimeImage": app_pod_runtime_image(),
+        "successBodyReclassification": success_body,
+        "valuesPrinted": False,
+    }
+
+def app_pod_runtime_image():
+    try:
+        pod = kube_json(["-n", NAMESPACE, "get", "pod", APP_POD], f"pod/{APP_POD}")
+        spec_containers = ((pod.get("spec") or {}).get("containers") or []) if isinstance(pod, dict) else []
+        status_containers = ((pod.get("status") or {}).get("containerStatuses") or []) if isinstance(pod, dict) else []
+        spec = next((item for item in spec_containers if item.get("name") == "sub2api"), spec_containers[0] if spec_containers else {})
+        status = next((item for item in status_containers if item.get("name") == "sub2api"), status_containers[0] if status_containers else {})
+        return {
+            "pod": APP_POD,
+            "image": spec.get("image"),
+            "imageID": status.get("imageID"),
+            "ready": status.get("ready"),
+            "restartCount": status.get("restartCount"),
+        }
+    except Exception as exc:
+        return {"pod": APP_POD, "error": str(exc)}
+
 def get_account_detail(token, account):
     account_id = account.get("id") if isinstance(account, dict) else None
     if account_id is None:
@@ -2764,9 +3101,10 @@ def run_sync():
     gateway = validate_gateway(api_key)
     responses_smoke = validate_gateway_responses(api_key)
     compact_evidence = recent_compact_gateway_evidence()
+    runtime_capabilities = validate_runtime_capabilities(token)
     return {
         "ok": gateway["ok"] is True and responses_smoke["ok"] is True and owner_concurrency["ok"] is True and capacity_status["ok"] is True and load_factor_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True,
-        "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")),
+        "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")) or runtime_capabilities.get("ok") is not True,
         "mode": "sync",
         "namespace": NAMESPACE,
         "serviceDns": SERVICE_DNS,
@@ -2800,6 +3138,7 @@ def run_sync():
         },
         "ownerBalance": owner_balance,
         "ownerConcurrency": owner_concurrency,
+        "runtimeCapabilities": runtime_capabilities,
         "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayCompactRecent": compact_evidence},
     }
 
@@ -2821,9 +3160,10 @@ def run_validate():
     gateway = validate_gateway(api_key)
     responses_smoke = validate_gateway_responses(api_key)
     compact_evidence = recent_compact_gateway_evidence()
+    runtime_capabilities = validate_runtime_capabilities(token)
     return {
         "ok": gateway["ok"] is True and responses_smoke["ok"] is True and (owner_concurrency is None or owner_concurrency["ok"] is True) and capacity_status["ok"] is True and load_factor_status["ok"] is True and ws_v2_status["ok"] is True and temp_unschedulable_status["ok"] is True,
-        "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")),
+        "degraded": bool(responses_smoke.get("degraded")) or bool(compact_evidence.get("degraded")) or runtime_capabilities.get("ok") is not True,
         "mode": "validate",
         "namespace": NAMESPACE,
         "serviceDns": SERVICE_DNS,
@@ -2842,6 +3182,7 @@ def run_validate():
         "loadFactor": load_factor_status,
         "webSocketsV2": ws_v2_status,
         "tempUnschedulable": temp_unschedulable_status,
+        "runtimeCapabilities": runtime_capabilities,
         "validation": {"gatewayModels": gateway, "gatewayResponses": responses_smoke, "gatewayCompactRecent": compact_evidence},
     }