Stop Sunrise waiting tasks for three accounts

README.md

@@ -1,8 +1,23 @@
 # xc_validation_strategy_vllm_stop
 
-用于停止 `zhoukaile` 账号当前状态为 `waiting` 的验证任务的 ModelHub XC 策略服务。
+用于依次停止以下账号中，状态严格为 `waiting` 且 GPU 类型严格为 `Sunrise_pt-200-x1` 的 ModelHub XC 验证任务：
 
-服务启动后会调用：
+- `jiangxiaowen`
+- `l112233`
+- `l11223344`
+
+策略在运行时使用每个账号自己的 `xc-Token` 分页查询任务，并对返回记录再次校验 `userAccount`、`status` 和 `gpuType`。其他 GPU 类型以及非 `waiting` 状态的任务不会进入停止请求。
+
+2026-08-10 部署前只读核验结果：
+
+| 账号 | `Sunrise_pt-200-x1` waiting 数量 |
+| --- | ---: |
+| `jiangxiaowen` | 866 |
+| `l112233` | 1400 |
+| `l11223344` | 619 |
+| 合计 | 2885 |
+
+服务启动后按账号及每批 50 条调用：
 
 ```text
 PUT /api/async/task/stop-create-contest-task
@@ -11,20 +26,10 @@ PUT /api/async/task/stop-create-contest-task
 
 停止请求完成后，进程继续运行，以便平台通过健康检查和状态接口读取执行结果。
 
-## 必填配置
-
-在策略构建/运行环境中配置以下机密环境变量：
-
-```text
-XC_TOKEN=<zhoukaile 的 xc-Token>
-USER_PASSWORD=<zhoukaile 的登录密码>
-```
-
-`USER_PASSWORD` 仅用于获得短期 Bearer Token，以分页读取该账号实时提交的验证任务；`XC_TOKEN` 用于停止接口。两者均不写入仓库。可选配置如下：
+## 可选配置
 
 | 变量 | 默认值 | 说明 |
 | --- | --- | --- |
-| `TASK_IDS` | 无 | 可选的逗号分隔任务 ID；设置后跳过实时查询，仅停止指定 ID。 |
 | `BASE_URL` | `https://modelhub.org.cn` | ModelHub 服务地址。 |
 | `BATCH_SIZE` | `50` | 每个停止请求包含的任务数。 |
 | `MAX_RETRIES` | `3` | 单批任务的最大请求次数。 |
@@ -34,17 +39,13 @@ USER_PASSWORD=<zhoukaile 的登录密码>
 ## 平台运行接口
 
 - `GET /health`：存活探针，成功返回 `{"status":"ok"}`。
-- `GET /status`：返回执行阶段、成功/失败数量、失败的任务 ID 与错误信息。
+- `GET /status`：返回目标 GPU、总体及每个账号的成功/失败数量、失败任务 ID 和错误信息。
 
 ## 运行与构建
 
-仓库根目录的 `Dockerfile` 使用平台 Python 基础镜像，暴露 `8080` 端口并以 `python main.py` 启动。策略构建时请将 `XC_TOKEN` 与 `USER_PASSWORD` 都设置为机密变量；构建成功后，策略会先分页读取当前账号的全部任务，仅筛选状态为 `waiting` 的任务再分批停止。
-
-本地验证：
+仓库根目录的 `Dockerfile` 使用平台 Python 基础镜像，暴露 `8080` 端口并以 `python main.py` 启动。构建成功后，策略会按 `jiangxiaowen → l112233 → l11223344` 的顺序逐账号处理。
 
 ```bash
-export XC_TOKEN='***'
-export USER_PASSWORD='***'
 python main.py
 curl http://127.0.0.1:8080/health
 curl http://127.0.0.1:8080/status

main.py

@@ -1,14 +1,15 @@
-"""Stop zhoukaile's waiting ModelHub XC validation tasks.
+"""Stop waiting Sunrise_pt-200-x1 validation tasks for selected accounts.
 
-The service performs the stop requests once at startup and then stays alive so
-the strategy platform can probe it through ``/health`` and inspect ``/status``.
+The worker processes accounts sequentially. For each account it discovers the
+current target tasks, validates account/status/GPU locally, then stops them in
+batches with that account's own xc-Token. The HTTP service remains available
+for platform health and status probes after the worker finishes.
 """
 
 import json
 import os
 import signal
 import threading
-import time
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any
@@ -16,27 +17,49 @@ from typing import Any
 import requests
 
 BASE_URL = os.environ.get("BASE_URL", "https://modelhub.org.cn").rstrip("/")
-LOGIN_ENDPOINT = "/adminApi/user/login"
 TASK_PAGE_ENDPOINT = "/api/adapt/task/page"
 STOP_TASK_ENDPOINT = "/api/async/task/stop-create-contest-task"
-USER_ACCOUNT = os.environ.get("USER_ACCOUNT", "zhoukaile")
-USER_PASSWORD = os.environ.get("USER_PASSWORD", "")
-XC_TOKEN = os.environ.get("XC_TOKEN", "")
+TARGET_GPU = "Sunrise_pt-200-x1"
+TARGET_STATUS = "waiting"
 STRATEGY_ID = os.environ.get("STRATEGY_ID", "")
 
 HTTP_HOST = "0.0.0.0"
 HTTP_PORT = int(os.environ.get("PORT", "8080"))
+PAGE_SIZE = 100
 BATCH_SIZE = int(os.environ.get("BATCH_SIZE", "50"))
 MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3"))
 REQUEST_TIMEOUT = int(os.environ.get("REQUEST_TIMEOUT", "30"))
 
-PAGE_SIZE = 100
-STOPPABLE_STATUS = "waiting"
+TARGETS = [
+    {
+        "account": "jiangxiaowen",
+        "xc_token": "88d5fee9f1fe4f7583f11a9d3702dc85",
+    },
+    {
+        "account": "l112233",
+        "xc_token": "40cb6910dc9a442a816298a228da65ac",
+    },
+    {
+        "account": "l11223344",
+        "xc_token": "e1c0db2959e5411f9342c8550b03f6e9",
+    },
+]
 
 _shutdown = threading.Event()
 _state: dict[str, Any] = {
     "strategy_id": STRATEGY_ID,
-    "account": USER_ACCOUNT,
+    "target_gpu": TARGET_GPU,
+    "target_status": TARGET_STATUS,
+    "accounts": {
+        target["account"]: {
+            "phase": "pending",
+            "total": 0,
+            "stopped": 0,
+            "failed": 0,
+            "failed_task_ids": [],
+        }
+        for target in TARGETS
+    },
     "phase": "starting",  # starting | stopping | done | partial_failure | error
     "total": 0,
     "stopped": 0,
@@ -52,83 +75,6 @@ def _now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
-def _task_ids_from_environment() -> list[int] | None:
-    """Return an explicit TASK_IDS override, if one was supplied."""
-    raw_task_ids = os.environ.get("TASK_IDS", "").strip()
-    if not raw_task_ids:
-        return None
-
-    task_ids: list[int] = []
-    for value in raw_task_ids.split(","):
-        value = value.strip()
-        if not value:
-            continue
-        try:
-            task_ids.append(int(value))
-        except ValueError as exc:
-            raise ValueError(f"TASK_IDS contains a non-numeric task ID: {value!r}") from exc
-    return list(dict.fromkeys(task_ids))
-
-
-def _login() -> str:
-    """Authenticate as the task owner and return a short-lived bearer token."""
-    if not USER_PASSWORD:
-        raise RuntimeError("USER_PASSWORD is required to query zhoukaile's current task IDs")
-    response = requests.post(
-        f"{BASE_URL}{LOGIN_ENDPOINT}",
-        headers={"Content-Type": "application/json"},
-        json={"userAccount": USER_ACCOUNT, "userPassword": USER_PASSWORD},
-        timeout=REQUEST_TIMEOUT,
-    )
-    response.raise_for_status()
-    result = response.json()
-    token = result.get("data", {}).get("token")
-    if result.get("code") != 0 or not isinstance(token, str) or not token:
-        raise RuntimeError(f"Task query login failed: {result.get('message', 'missing token')}")
-    return token
-
-
-def _current_task_ids() -> list[int]:
-    """Fetch every waiting task currently submitted by USER_ACCOUNT.
-
-    This intentionally queries the user-facing task page at run time.  A static
-    list becomes obsolete as soon as zhoukaile submits more validation tasks.
-    """
-    bearer_token = _login()
-    headers = {"Authorization": f"Bearer {bearer_token}"}
-    task_ids: list[int] = []
-    page = 1
-
-    while True:
-        response = requests.get(
-            f"{BASE_URL}{TASK_PAGE_ENDPOINT}",
-            headers=headers,
-            params={"current": page, "pageSize": PAGE_SIZE},
-            timeout=REQUEST_TIMEOUT,
-        )
-        response.raise_for_status()
-        result = response.json()
-        if result.get("code") != 0:
-            raise RuntimeError(f"Task query failed: {result.get('message', 'unknown error')}")
-
-        data = result.get("data") or {}
-        records = data.get("records") or []
-        for record in records:
-            if str(record.get("status", "")).lower() != STOPPABLE_STATUS:
-                continue
-            try:
-                task_ids.append(int(record["taskId"]))
-            except (KeyError, TypeError, ValueError):
-                print(f"[query] skipping record without a numeric taskId: {record}", flush=True)
-
-        pages = int(data.get("pages") or 0)
-        if page >= pages:
-            break
-        page += 1
-
-    return list(dict.fromkeys(task_ids))
-
-
 class Handler(BaseHTTPRequestHandler):
     def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
         if self.path == "/health":
@@ -159,8 +105,69 @@ def _run_http() -> None:
     server.server_close()
 
 
-def _stop_batch(task_ids: list[int]) -> bool:
-    headers = {"Content-Type": "application/json", "xc-Token": XC_TOKEN}
+def _fetch_target_task_ids(account: str, xc_token: str) -> list[int]:
+    """Return only this account's waiting tasks on the exact target GPU."""
+    headers = {"xc-Token": xc_token}
+    task_ids: list[int] = []
+    page = 1
+
+    while True:
+        response = requests.get(
+            f"{BASE_URL}{TASK_PAGE_ENDPOINT}",
+            headers=headers,
+            params={
+                "current": page,
+                "pageSize": PAGE_SIZE,
+                "status": TARGET_STATUS,
+                "gpuType": TARGET_GPU,
+            },
+            timeout=REQUEST_TIMEOUT,
+        )
+        response.raise_for_status()
+        result = response.json()
+        if result.get("code") != 0:
+            raise RuntimeError(
+                f"{account}: task query failed: {result.get('message', 'unknown error')}"
+            )
+
+        data = result.get("data") or {}
+        for record in data.get("records") or []:
+            record_account = record.get("userAccount")
+            record_status = str(record.get("status", "")).lower()
+            record_gpu = record.get("gpuType")
+            if (
+                record_account != account
+                or record_status != TARGET_STATUS
+                or record_gpu != TARGET_GPU
+            ):
+                print(
+                    f"[query] {account}: rejected mismatched record "
+                    f"taskId={record.get('taskId')} account={record_account!r} "
+                    f"status={record_status!r} gpuType={record_gpu!r}",
+                    flush=True,
+                )
+                continue
+            try:
+                task_ids.append(int(record["taskId"]))
+            except (KeyError, TypeError, ValueError):
+                print(f"[query] {account}: rejected non-numeric task ID", flush=True)
+
+        pages = int(data.get("pages") or 0)
+        if page >= pages:
+            break
+        page += 1
+
+    unique_task_ids = list(dict.fromkeys(task_ids))
+    print(
+        f"[query] {account}: found {len(unique_task_ids)} "
+        f"{TARGET_STATUS} tasks on {TARGET_GPU}",
+        flush=True,
+    )
+    return unique_task_ids
+
+
+def _stop_batch(account: str, xc_token: str, task_ids: list[int]) -> bool:
+    headers = {"Content-Type": "application/json", "xc-Token": xc_token}
     url = f"{BASE_URL}{STOP_TASK_ENDPOINT}"
     for attempt in range(1, MAX_RETRIES + 1):
         try:
@@ -176,15 +183,22 @@ def _stop_batch(task_ids: list[int]) -> bool:
                 result = {"message": response.text[:500]}
 
             if response.ok and result.get("code") == 0:
-                print(f"[stop] stopped task IDs: {task_ids}", flush=True)
+                print(
+                    f"[stop] {account}: stopped {len(task_ids)} task IDs",
+                    flush=True,
+                )
                 return True
             print(
-                f"[stop] attempt {attempt}/{MAX_RETRIES} failed for {task_ids}: "
-                f"HTTP {response.status_code}, {result}",
+                f"[stop] {account}: attempt {attempt}/{MAX_RETRIES} failed "
+                f"for {len(task_ids)} tasks: HTTP {response.status_code}, {result}",
                 flush=True,
             )
         except requests.RequestException as exc:
-            print(f"[stop] attempt {attempt}/{MAX_RETRIES} request error for {task_ids}: {exc}", flush=True)
+            print(
+                f"[stop] {account}: attempt {attempt}/{MAX_RETRIES} "
+                f"request error: {exc}",
+                flush=True,
+            )
 
         if attempt < MAX_RETRIES and not _shutdown.wait(attempt):
             continue
@@ -197,25 +211,33 @@ def _run_worker() -> None:
     _state["started_at"] = _now()
     _state["phase"] = "stopping"
     try:
-        if not XC_TOKEN:
-            raise RuntimeError("XC_TOKEN is required and must be configured in the strategy environment")
-
-        task_ids = _task_ids_from_environment() or _current_task_ids()
-        if not task_ids:
-            print("[query] no active validation tasks found", flush=True)
-            _state["phase"] = "done"
-            return
-        _state["total"] = len(task_ids)
-
-        for start in range(0, len(task_ids), BATCH_SIZE):
+        for target in TARGETS:
             if _shutdown.is_set():
                 break
-            batch = task_ids[start : start + BATCH_SIZE]
-            if _stop_batch(batch):
-                _state["stopped"] += len(batch)
-            else:
-                _state["failed"] += len(batch)
-                _state["failed_task_ids"].extend(batch)
+            account = target["account"]
+            account_state = _state["accounts"][account]
+            account_state["phase"] = "querying"
+            task_ids = _fetch_target_task_ids(account, target["xc_token"])
+            account_state["total"] = len(task_ids)
+            _state["total"] += len(task_ids)
+            account_state["phase"] = "stopping"
+
+            for start in range(0, len(task_ids), BATCH_SIZE):
+                if _shutdown.is_set():
+                    break
+                batch = task_ids[start : start + BATCH_SIZE]
+                if _stop_batch(account, target["xc_token"], batch):
+                    _state["stopped"] += len(batch)
+                    account_state["stopped"] += len(batch)
+                else:
+                    _state["failed"] += len(batch)
+                    _state["failed_task_ids"].extend(batch)
+                    account_state["failed"] += len(batch)
+                    account_state["failed_task_ids"].extend(batch)
+
+            account_state["phase"] = (
+                "done" if account_state["failed"] == 0 else "partial_failure"
+            )
 
         _state["phase"] = "done" if _state["failed"] == 0 else "partial_failure"
     except Exception as exc:  # exposed through /status for diagnosis