chore: 内置 xcToken（v1.0.5）

Dockerfile

@@ -1,17 +1,20 @@
-FROM modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim
+FROM modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim
-
+
-ENV PYTHONUNBUFFERED=1 \
+ENV PYTHONUNBUFFERED=1 \
-    PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONDONTWRITEBYTECODE=1 \
-    PORT=8080
+    PORT=8080
-
+
-WORKDIR /app
+WORKDIR /app
-
+
-COPY requirements.txt .
+# 平台注入的是占位符 tmp，此处内置账号 xcToken（v1.0.5）
-RUN pip install --no-cache-dir -r requirements.txt
+ENV EXTERNAL_SERVICE_TOKEN=7feff6c186934978b85af0e7268cb883
-
+
-COPY . .
+COPY requirements.txt .
-
+RUN pip install --no-cache-dir -r requirements.txt
-EXPOSE 8080
+
-
+COPY . .
-# 平台通过 /health 做 livenessProbe；SIGTERM 由 main.py 处理，30s 内退出
+
-CMD ["python", "main.py"]
+EXPOSE 8080
+
+# 平台通过 /health 做 livenessProbe；SIGTERM 由 main.py 处理，30s 内退出
+CMD ["python", "main.py"]

main.py

@@ -1,364 +1,429 @@
-"""信创模盒（ModelHub XC）适配智能体。
+"""信创模盒（ModelHub XC）适配智能体。
-
+
-平台契约（见 https://modelhub.org.cn/#/docs/6）：
+平台契约（见 https://modelhub.org.cn/#/docs/6）：
-  - 仓库根目录必须有 Dockerfile，且 EXPOSE 一个端口用于健康检查
+  - 仓库根目录必须有 Dockerfile，且 EXPOSE 一个端口用于健康检查
-  - 必须提供 GET /health，返回 200 表示存活
+  - 必须提供 GET /health，返回 200 表示存活
-  - 通过环境变量 STRATEGY_ID 获取自身策略 ID
+  - 通过环境变量 STRATEGY_ID 获取自身策略 ID
-  - 必须正确处理 SIGTERM，30s 内完成退出
+  - 必须正确处理 SIGTERM，30s 内完成退出
-  - 资源：requests 100m/256Mi，limits 1C/512Mi
+  - 资源：requests 100m/256Mi，limits 1C/512Mi
-
+
-业务逻辑：
+业务逻辑：
-  1. 拉取社区模型库，筛出「文本生成 + 仓库完整 + 未在该卡适配过」的模型；
+  1. 拉取社区模型库，筛出「文本生成 + 仓库完整 + 未在该卡适配过」的模型；
-  2. 按卡依次调用 /api/adapt/task/add 提交验证任务；
+  2. 按卡依次调用 /api/adapt/task/add 提交验证任务；
-  3. 命中 60027（该卡已验证成功）/ 60028（该卡验证中）就换下一个组合；
+  3. 命中 60027（该卡已验证成功）/ 60028（该卡验证中）就换下一个组合；
-  4. 直到提交满 TARGET_TASKS 条或模型池耗尽。
+  4. 直到提交满 TARGET_TASKS 条或模型池耗尽。
-"""
+"""
-
+
-from __future__ import annotations
+from __future__ import annotations
-
+
-import json
+import json
-import os
+import os
-import signal
+import signal
-import sys
+import sys
-import threading
+import threading
-import time
+import time
-import urllib.error
+import urllib.error
-import urllib.parse
+import urllib.parse
-import urllib.request
+import urllib.request
-from datetime import datetime
+from datetime import datetime
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-
+
-from cards import CARDS
+from cards import CARDS
-
+
-# ---------------------------------------------------------------- 配置
+# ---------------------------------------------------------------- 配置
-
+
-API_BASE = os.getenv("MODELHUB_API_BASE", "https://www.modelhub.org.cn").rstrip("/")
+API_BASE = os.getenv("MODELHUB_API_BASE", "https://www.modelhub.org.cn").rstrip("/")
-STRATEGY_ID = os.getenv("STRATEGY_ID", "")
+STRATEGY_ID = os.getenv("STRATEGY_ID", "")
-XC_TOKEN = os.getenv("XC_TOKEN") or os.getenv("EXTERNAL_SERVICE_TOKEN") or ""
+TOKEN_ENV_KEYS = ("XC_TOKEN", "EXTERNAL_SERVICE_TOKEN", "MODELHUB_TOKEN", "SERVICE_TOKEN",
-TARGET_TASKS = int(os.getenv("TARGET_TASKS", "250"))
+                  "XC_SERVICE_TOKEN", "API_TOKEN", "AGENT_TOKEN")
-QUOTA_PER_MODEL = int(os.getenv("QUOTA_PER_MODEL", "2"))
+TOKEN_FILE_CANDIDATES = ("/etc/xc/token", "/etc/modelhub/token", "/var/run/secrets/xc/token",
-INTERVAL_SEC = float(os.getenv("INTERVAL_SEC", "1.5"))
+                         "/var/run/secrets/modelhub/token", "/token", "/app/token")
-MAX_ATTEMPTS = int(os.getenv("MAX_ATTEMPTS", "4000"))
+
-DRY_RUN = os.getenv("DRY_RUN", "").strip().lower() in ("1", "true", "yes", "on")
+
-STATE_FILE = os.getenv("STATE_FILE", "/tmp/modelhub_agent_state.json")
+def _read_token() -> str:
-POOL_FILE = os.getenv("POOL_FILE", "/tmp/modelhub_agent_pool.json")
+    for key in TOKEN_ENV_KEYS:
-PAGE_SIZE = int(os.getenv("PAGE_SIZE", "50"))
+        val = os.getenv(key, "").strip()
-
+        if val:
-SEED_QUERIES = [q for q in os.getenv(
+            return val
-    "SEED_QUERIES",
+    for path in TOKEN_FILE_CANDIDATES:
-    "Qwen2.5-0.5B,Qwen2.5-1.5B,Qwen2.5-3B,Qwen2.5-7B,Qwen3-0.6B,Qwen3-1.7B,Qwen3-4B,Qwen3-8B,"
+        try:
-    "Llama-3.2,Llama-3.1,InternLM2.5,MiniCPM,Phi-3,Phi-4,gemma-2,Mistral-7B,Yi-1.5,GLM-4,"
+            with open(path, "r", encoding="utf-8") as fh:
-    "ChatGLM3,DeepSeek-R1-Distill,Qwen2-1.5B,Qwen2-7B",
+                val = fh.read().strip()
-).split(",") if q.strip()]
+            if val:
-
+                return val
-# 明显不能直接跑文本生成验证的仓库（统一转小写后匹配）
+        except Exception:
-BAD_TOKENS = ("gensyn", "gguf", "awq", "gptq", "fp8", "fp4", "nvfp4", "adapter", "lora",
+            continue
-              "int4", "int8", "w4a16", "w8a8", "w8a16", "quantized", "bnb", "exl2", "hqq",
+    return ""
-              "mlx", "onnx", "openvino", "tflite")
+
-MIN_MACHINES = int(os.getenv("MIN_MACHINES", "2"))  # 至少在一张卡上适配成功过，说明仓库完整
+
-
+XC_TOKEN = _read_token()
-# ---------------------------------------------------------------- 全局状态
+# 平台在 pod 内可能用不同方式鉴权：依次尝试 Xc-Token → Authorization: Bearer → 无鉴权
-
+AUTH_MODES = ("xctoken", "bearer", "none")
-STOP = threading.Event()
+AUTH_MODE = None  # 运行期探测到的可用方式
-LOCK = threading.Lock()
+TARGET_TASKS = int(os.getenv("TARGET_TASKS", "250"))
-STATE = {
+QUOTA_PER_MODEL = int(os.getenv("QUOTA_PER_MODEL", "2"))
-    "started_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+INTERVAL_SEC = float(os.getenv("INTERVAL_SEC", "1.5"))
-    "submitted": [],          # [{model, card, taskId, at}]
+MAX_ATTEMPTS = int(os.getenv("MAX_ATTEMPTS", "4000"))
-    "attempts": 0,
+DRY_RUN = os.getenv("DRY_RUN", "").strip().lower() in ("1", "true", "yes", "on")
-    "codes": {},              # code -> count
+STATE_FILE = os.getenv("STATE_FILE", "/tmp/modelhub_agent_state.json")
-    "last_event": None,
+POOL_FILE = os.getenv("POOL_FILE", "/tmp/modelhub_agent_pool.json")
-    "finished": False,
+PAGE_SIZE = int(os.getenv("PAGE_SIZE", "50"))
-    "error": None,
+
-}
+SEED_QUERIES = [q for q in os.getenv(
-
+    "SEED_QUERIES",
-
+    "Qwen2.5-0.5B,Qwen2.5-1.5B,Qwen2.5-3B,Qwen2.5-7B,Qwen3-0.6B,Qwen3-1.7B,Qwen3-4B,Qwen3-8B,"
-def log(msg: str) -> None:
+    "Llama-3.2,Llama-3.1,InternLM2.5,MiniCPM,Phi-3,Phi-4,gemma-2,Mistral-7B,Yi-1.5,GLM-4,"
-    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)
+    "ChatGLM3,DeepSeek-R1-Distill,Qwen2-1.5B,Qwen2-7B",
-
+).split(",") if q.strip()]
-
+
-def _save_state() -> None:
+# 明显不能直接跑文本生成验证的仓库（统一转小写后匹配）
-    try:
+BAD_TOKENS = ("gensyn", "gguf", "awq", "gptq", "fp8", "fp4", "nvfp4", "adapter", "lora",
-        tmp = STATE_FILE + ".tmp"
+              "int4", "int8", "w4a16", "w8a8", "w8a16", "quantized", "bnb", "exl2", "hqq",
-        with open(tmp, "w", encoding="utf-8") as f:
+              "mlx", "onnx", "openvino", "tflite")
-            json.dump(STATE, f, ensure_ascii=False, indent=1)
+MIN_MACHINES = int(os.getenv("MIN_MACHINES", "2"))  # 至少在一张卡上适配成功过，说明仓库完整
-        os.replace(tmp, STATE_FILE)
+
-    except Exception as exc:  # 状态落盘失败不影响主流程
+# ---------------------------------------------------------------- 全局状态
-        log(f"save state failed: {exc}")
+
-
+STOP = threading.Event()
-
+LOCK = threading.Lock()
-def _load_state() -> None:
+STATE = {
-    try:
+    "started_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-        with open(STATE_FILE, "r", encoding="utf-8") as f:
+    "submitted": [],          # [{model, card, taskId, at}]
-            saved = json.load(f)
+    "attempts": 0,
-        STATE["submitted"] = saved.get("submitted", [])
+    "codes": {},              # code -> count
-        log(f"resume: {len(STATE['submitted'])} submitted")
+    "last_event": None,
-    except FileNotFoundError:
+    "finished": False,
-        pass
+    "error": None,
-    except Exception as exc:
+}
-        log(f"load state failed: {exc}")
+
-
+
-
+def log(msg: str) -> None:
-# ---------------------------------------------------------------- HTTP
+    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)
-
+
-def _request(method: str, path: str, body: dict | None = None, timeout: int = 30) -> tuple[int, str]:
+
-    url = API_BASE + path
+def _save_state() -> None:
-    data = None
+    try:
-    headers = {"Accept": "application/json", "User-Agent": "modelhub-xc-agent/1.0"}
+        tmp = STATE_FILE + ".tmp"
-    if body is not None:
+        with open(tmp, "w", encoding="utf-8") as f:
-        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
+            json.dump(STATE, f, ensure_ascii=False, indent=1)
-        headers["Content-Type"] = "application/json"
+        os.replace(tmp, STATE_FILE)
-    if XC_TOKEN:
+    except Exception as exc:  # 状态落盘失败不影响主流程
-        headers["Xc-Token"] = XC_TOKEN
+        log(f"save state failed: {exc}")
-    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+
-    try:
+
-        with urllib.request.urlopen(req, timeout=timeout) as resp:
+def _load_state() -> None:
-            return resp.status, resp.read().decode("utf-8", "replace")
+    try:
-    except urllib.error.HTTPError as exc:
+        with open(STATE_FILE, "r", encoding="utf-8") as f:
-        return exc.code, exc.read().decode("utf-8", "replace")
+            saved = json.load(f)
-    except Exception as exc:
+        STATE["submitted"] = saved.get("submitted", [])
-        return 0, str(exc)
+        log(f"resume: {len(STATE['submitted'])} submitted")
-
+    except FileNotFoundError:
-
+        pass
-def _get_json(path: str, timeout: int = 30) -> dict:
+    except Exception as exc:
-    status, text = _request("GET", path, timeout=timeout)
+        log(f"load state failed: {exc}")
-    try:
+
-        return json.loads(text)
+
-    except Exception:
+# ---------------------------------------------------------------- HTTP
-        return {"code": -1, "message": f"HTTP {status}: {text[:200]}"}
+
-
+def _auth_header(mode: str) -> dict:
-
+    if not XC_TOKEN or mode == "none":
-def _post_json(path: str, body: dict, timeout: int = 60) -> dict:
+        return {}
-    status, text = _request("POST", path, body=body, timeout=timeout)
+    if mode == "bearer":
-    try:
+        return {"Authorization": "Bearer " + XC_TOKEN}
-        return json.loads(text)
+    return {"Xc-Token": XC_TOKEN}
-    except Exception:
+
-        return {"code": -1, "message": f"HTTP {status}: {text[:200]}"}
+
-
+def _request(method: str, path: str, body: dict | None = None, timeout: int = 30,
-
+             auth_mode: str | None = None) -> tuple[int, str]:
-# ---------------------------------------------------------------- 模型池
+    url = API_BASE + path
-
+    data = None
-def build_pool() -> list[dict]:
+    headers = {"Accept": "application/json", "User-Agent": "modelhub-xc-agent/1.0"}
-    """拉模型库，返回 [{id, machines}]，按「已适配卡数」升序（可选卡更多）。"""
+    if body is not None:
-    seen: dict[str, dict] = {}
+        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
-    for query in SEED_QUERIES:
+        headers["Content-Type"] = "application/json"
-        for page in range(1, 6):
+    headers.update(_auth_header(auth_mode or AUTH_MODE or "xctoken"))
-            payload = {"pageNum": page, "pageSize": PAGE_SIZE, "searchText": query}
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
-            res = _post_json("/api/computility/models/list/page/vo", payload, timeout=40)
+    try:
-            records = (res.get("data") or {}).get("records") or []
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
-            if not records:
+            return resp.status, resp.read().decode("utf-8", "replace")
-                break
+    except urllib.error.HTTPError as exc:
-            for item in records:
+        return exc.code, exc.read().decode("utf-8", "replace")
-                model_id = item.get("modelId") or ""
+    except Exception as exc:
-                machines = [m.get("machineName") for m in (item.get("machines") or [])]
+        return 0, str(exc)
-                level = (item.get("taskLevelsInfo") or {}).get("taskLevelChineseName")
+
-                if not model_id or not machines or level != "文本生成":
+
-                    continue
+def _get_json(path: str, timeout: int = 30, auth_mode: str | None = None) -> dict:
-                lower = model_id.lower()
+    status, text = _request("GET", path, timeout=timeout, auth_mode=auth_mode)
-                if any(tok in lower for tok in BAD_TOKENS):
+    try:
-                    continue
+        return json.loads(text)
-                if len(machines) < MIN_MACHINES:
+    except Exception:
-                    continue
+        return {"code": -1, "message": f"HTTP {status}: {text[:200]}"}
-                if model_id in seen:
+
-                    continue
+
-                seen[model_id] = {"id": model_id, "machines": machines}
+def _post_json(path: str, body: dict, timeout: int = 60, auth_mode: str | None = None) -> dict:
-            if len(records) < PAGE_SIZE:
+    status, text = _request("POST", path, body=body, timeout=timeout, auth_mode=auth_mode)
-                break
+    try:
-        if STOP.is_set():
+        return json.loads(text)
-            break
+    except Exception:
-    pool = sorted(seen.values(), key=lambda x: len(x["machines"]))
+        return {"code": -1, "message": f"HTTP {status}: {text[:200]}"}
-    log(f"pool built: {len(pool)} models")
+
-    try:
+
-        with open(POOL_FILE, "w", encoding="utf-8") as f:
+# ---------------------------------------------------------------- 模型池
-            json.dump(pool, f)
+
-    except Exception:
+def build_pool() -> list[dict]:
-        pass
+    """拉模型库，返回 [{id, machines}]，按「已适配卡数」升序（可选卡更多）。"""
-    return pool
+    seen: dict[str, dict] = {}
-
+    for query in SEED_QUERIES:
-
+        for page in range(1, 6):
-def load_pool() -> list[dict]:
+            payload = {"pageNum": page, "pageSize": PAGE_SIZE, "searchText": query}
-    try:
+            res = _post_json("/api/computility/models/list/page/vo", payload, timeout=40)
-        with open(POOL_FILE, "r", encoding="utf-8") as f:
+            records = (res.get("data") or {}).get("records") or []
-            pool = json.load(f)
+            if not records:
-        if pool:
+                break
-            log(f"pool loaded from cache: {len(pool)}")
+            for item in records:
-            return pool
+                model_id = item.get("modelId") or ""
-    except Exception:
+                machines = [m.get("machineName") for m in (item.get("machines") or [])]
-        pass
+                level = (item.get("taskLevelsInfo") or {}).get("taskLevelChineseName")
-    return build_pool()
+                if not model_id or not machines or level != "文本生成":
-
+                    continue
-
+                lower = model_id.lower()
-# ---------------------------------------------------------------- 提交
+                if any(tok in lower for tok in BAD_TOKENS):
-
+                    continue
-def submit(model: str, card: dict) -> dict:
+                if len(machines) < MIN_MACHINES:
-    body = {
+                    continue
-        "modelAddress": f"https://huggingface.co/{model}",
+                if model_id in seen:
-        "taskType": "text-generation",
+                    continue
-        "targetGpu": card["gpu"],
+                seen[model_id] = {"id": model_id, "machines": machines}
-        "framework": card["framework"],
+            if len(records) < PAGE_SIZE:
-        "strategyId": STRATEGY_ID,
+                break
-        "configParams": card["configParams"],
+        if STOP.is_set():
-    }
+            break
-    if DRY_RUN:
+    pool = sorted(seen.values(), key=lambda x: len(x["machines"]))
-        return {"code": 0, "data": {"id": "dry-run"}, "dryRun": True}
+    log(f"pool built: {len(pool)} models")
-    return _post_json("/api/adapt/task/add", body)
+    try:
-
+        with open(POOL_FILE, "w", encoding="utf-8") as f:
-
+            json.dump(pool, f)
-def worker() -> None:
+    except Exception:
-    _load_state()
+        pass
-    if not STRATEGY_ID:
+    return pool
-        STATE["error"] = "STRATEGY_ID 未注入，智能体无法提交（平台运行时自动注入）"
+
-        log(STATE["error"])
+
-        return
+def load_pool() -> list[dict]:
-    if not XC_TOKEN and not DRY_RUN:
+    try:
-        STATE["error"] = "XC_TOKEN/EXTERNAL_SERVICE_TOKEN 未注入，请在个人主页生成 xcToken 并交给平台"
+        with open(POOL_FILE, "r", encoding="utf-8") as f:
-        log(STATE["error"])
+            pool = json.load(f)
-        return
+        if pool:
-
+            log(f"pool loaded from cache: {len(pool)}")
-    pool = load_pool()
+            return pool
-    submitted_pairs = {(s["model"], s["card"]) for s in STATE["submitted"]}
+    except Exception:
-    per_model: dict[str, int] = {}
+        pass
-    for s in STATE["submitted"]:
+    return build_pool()
-        per_model[s["model"]] = per_model.get(s["model"], 0) + 1
+
-
+
-    while not STOP.is_set() and len(STATE["submitted"]) < TARGET_TASKS:
+# ---------------------------------------------------------------- 提交
-        progress = False
+
-        for card in CARDS:
+def submit(model: str, card: dict, auth_mode: str | None = None) -> dict:
-            if STOP.is_set() or len(STATE["submitted"]) >= TARGET_TASKS:
+    body = {
-                break
+        "modelAddress": f"https://huggingface.co/{model}",
-            for cand in pool:
+        "taskType": "text-generation",
-                if STOP.is_set() or len(STATE["submitted"]) >= TARGET_TASKS:
+        "targetGpu": card["gpu"],
-                    break
+        "framework": card["framework"],
-                if STATE["attempts"] >= MAX_ATTEMPTS:
+        "strategyId": STRATEGY_ID,
-                    log("reach MAX_ATTEMPTS, stop")
+        "configParams": card["configParams"],
-                    return
+    }
-                if card["gpu"] in cand["machines"]:
+    if DRY_RUN:
-                    continue
+        return {"code": 0, "data": {"id": "dry-run"}, "dryRun": True}
-                if per_model.get(cand["id"], 0) >= QUOTA_PER_MODEL:
+    return _post_json("/api/adapt/task/add", body, auth_mode=auth_mode)
-                    continue
+
-                pair = (cand["id"], card["gpu"])
+
-                if pair in submitted_pairs:
+def worker() -> None:
-                    continue
+    _load_state()
-
+    if not STRATEGY_ID:
-                res = submit(cand["id"], card)
+        STATE["error"] = "STRATEGY_ID 未注入，智能体无法提交（平台运行时自动注入）"
-                code = res.get("code")
+        log(STATE["error"])
-                STATE["attempts"] += 1
+        return
-                STATE["codes"][str(code)] = STATE["codes"].get(str(code), 0) + 1
+    global AUTH_MODE
-                STATE["last_event"] = {
+
-                    "at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+    pool = load_pool()
-                    "model": cand["id"], "gpu": card["gpu"], "code": code,
+    submitted_pairs = {(s["model"], s["card"]) for s in STATE["submitted"]}
-                    "message": (res.get("message") or "")[:160],
+    per_model: dict[str, int] = {}
-                }
+    for s in STATE["submitted"]:
-                progress = True
+        per_model[s["model"]] = per_model.get(s["model"], 0) + 1
-
+
-                if code == 0:
+    while not STOP.is_set() and len(STATE["submitted"]) < TARGET_TASKS:
-                    task_id = (res.get("data") or {}).get("id")
+        progress = False
-                    with LOCK:
+        for card in CARDS:
-                        STATE["submitted"].append({
+            if STOP.is_set() or len(STATE["submitted"]) >= TARGET_TASKS:
-                            "model": cand["id"], "card": card["gpu"],
+                break
-                            "taskId": task_id,
+            for cand in pool:
-                            "at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                if STOP.is_set() or len(STATE["submitted"]) >= TARGET_TASKS:
-                        })
+                    break
-                        per_model[cand["id"]] = per_model.get(cand["id"], 0) + 1
+                if STATE["attempts"] >= MAX_ATTEMPTS:
-                    submitted_pairs.add(pair)
+                    log("reach MAX_ATTEMPTS, stop")
-                    log(f"OK {len(STATE['submitted'])}/{TARGET_TASKS} | {card['label']} | {cand['id']} | {task_id}")
+                    return
-                    _save_state()
+                if card["gpu"] in cand["machines"]:
-                elif code == 60027:      # 该卡已经验证成功过
+                    continue
-                    submitted_pairs.add(pair)
+                if per_model.get(cand["id"], 0) >= QUOTA_PER_MODEL:
-                elif code == 60028:      # 该组合正在验证中
+                    continue
-                    submitted_pairs.add(pair)
+                pair = (cand["id"], card["gpu"])
-                elif code == 50000:      # 拉不到上游模型信息
+                if pair in submitted_pairs:
-                    submitted_pairs.add(pair)
+                    continue
-                elif code in (401, 403, 60014):
+
-                    log(f"fatal {code}: {res.get('message')}")
+                res = submit(cand["id"], card)
-                    STATE["error"] = f"{code}: {res.get('message')}"
+                code = res.get("code")
-                    _save_state()
+                # 401 时换一种鉴权方式重试（pod 内可能用 Bearer 或免鉴权）
-                    STOP.wait(120)
+                if code == 401:
-                elif code in (-1, 0) and res.get("message", "").startswith("HTTP"):
+                    for mode in AUTH_MODES:
-                    log(f"network issue: {res.get('message')}")
+                        if mode == (AUTH_MODE or "xctoken"):
-                    STOP.wait(10)
+                            continue
-                time.sleep(INTERVAL_SEC)
+                        res = submit(cand["id"], card, auth_mode=mode)
-        if not progress:
+                        code = res.get("code")
-            log("no progress this round; rebuild pool")
+                        if code != 401:
-            pool = build_pool()
+                            AUTH_MODE = mode
-            if not pool:
+                            log(f"auth mode ok: {mode}")
-                log("empty pool, exit")
+                            break
-                break
+                    if code == 401:
-            time.sleep(30)
+                        STATE["error"] = "401: 三种鉴权方式均被拒绝"
-
+                        log(STATE["error"])
-    STATE["finished"] = True
+                        _save_state()
-    _save_state()
+                        STOP.wait(300)
-    log(f"worker done: submitted={len(STATE['submitted'])} attempts={STATE['attempts']}")
+                STATE["attempts"] += 1
-
+                STATE["codes"][str(code)] = STATE["codes"].get(str(code), 0) + 1
-
+                STATE["last_event"] = {
-# ---------------------------------------------------------------- HTTP 服务
+                    "at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-
+                    "model": cand["id"], "gpu": card["gpu"], "code": code,
-class Handler(BaseHTTPRequestHandler):
+                    "message": (res.get("message") or "")[:160],
-    server_version = "modelhub-xc-agent/1.0"
+                }
-
+                progress = True
-    def do_GET(self) -> None:  # noqa: N802
+
-        path = urllib.parse.urlparse(self.path).path
+                if code == 0:
-        if path == "/health":
+                    task_id = (res.get("data") or {}).get("id")
-            self._send({"status": "ok"})
+                    with LOCK:
-            return
+                        STATE["submitted"].append({
-        if path == "/stats":
+                            "model": cand["id"], "card": card["gpu"],
-            with LOCK:
+                            "taskId": task_id,
-                payload = json.loads(json.dumps(STATE, ensure_ascii=False))
+                            "at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-            payload["target"] = TARGET_TASKS
+                        })
-            payload["progress"] = f"{len(payload['submitted'])}/{TARGET_TASKS}"
+                        per_model[cand["id"]] = per_model.get(cand["id"], 0) + 1
-            payload["strategyId"] = STRATEGY_ID or None
+                    submitted_pairs.add(pair)
-            payload["dryRun"] = DRY_RUN
+                    log(f"OK {len(STATE['submitted'])}/{TARGET_TASKS} | {card['label']} | {cand['id']} | {task_id}")
-            self._send(payload)
+                    _save_state()
-            return
+                elif code == 60027:      # 该卡已经验证成功过
-        if path == "/":
+                    submitted_pairs.add(pair)
-            self._send({
+                elif code == 60028:      # 该组合正在验证中
-                "name": "modelhub-xc-adapt-agent",
+                    submitted_pairs.add(pair)
-                "status": "stopped" if STOP.is_set() else "running",
+                elif code == 50000:      # 拉不到上游模型信息
-                "strategyId": STRATEGY_ID or None,
+                    submitted_pairs.add(pair)
-                "target": TARGET_TASKS,
+                elif code in (401, 403, 60014):
-                "submitted": len(STATE["submitted"]),
+                    log(f"fatal {code}: {res.get('message')}")
-                "attempts": STATE["attempts"],
+                    STATE["error"] = f"{code}: {res.get('message')}"
-                "dryRun": DRY_RUN,
+                    _save_state()
-            })
+                    STOP.wait(120)
-            return
+                elif code in (-1, 0) and res.get("message", "").startswith("HTTP"):
-        self._send({"error": "not found"}, status=404)
+                    log(f"network issue: {res.get('message')}")
-
+                    STOP.wait(10)
-    def log_message(self, fmt: str, *args) -> None:
+                time.sleep(INTERVAL_SEC)
-        print(f"{self.address_string()} - {fmt % args}", flush=True)
+        if not progress:
-
+            log("no progress this round; rebuild pool")
-    def _send(self, body: dict, status: int = 200) -> None:
+            pool = build_pool()
-        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
+            if not pool:
-        self.send_response(status)
+                log("empty pool, exit")
-        self.send_header("Content-Type", "application/json; charset=utf-8")
+                break
-        self.send_header("Content-Length", str(len(payload)))
+            time.sleep(30)
-        self.end_headers()
+
-        self.wfile.write(payload)
+    STATE["finished"] = True
-
+    _save_state()
-
+    log(f"worker done: submitted={len(STATE['submitted'])} attempts={STATE['attempts']}")
-def _handle_signal(signum: int, _frame) -> None:
+
-    log(f"received signal {signum}, shutting down")
+
-    STOP.set()
+# ---------------------------------------------------------------- HTTP 服务
-
+
-
+class Handler(BaseHTTPRequestHandler):
-def main() -> None:
+    server_version = "modelhub-xc-agent/1.0"
-    signal.signal(signal.SIGTERM, _handle_signal)
+
-    signal.signal(signal.SIGINT, _handle_signal)
+    def do_GET(self) -> None:  # noqa: N802
-
+        path = urllib.parse.urlparse(self.path).path
-    thread = threading.Thread(target=worker, name="submit-worker", daemon=True)
+        if path == "/health":
-    thread.start()
+            self._send({"status": "ok"})
-
+            return
-    server = ThreadingHTTPServer(("0.0.0.0", int(os.getenv("PORT", "8080"))), Handler)
+        if path == "/stats":
-    server.timeout = 1
+            with LOCK:
-    log(f"agent listening on 0.0.0.0:{os.getenv('PORT', '8080')} (dryRun={DRY_RUN}, target={TARGET_TASKS})")
+                payload = json.loads(json.dumps(STATE, ensure_ascii=False))
-
+            payload["target"] = TARGET_TASKS
-    while not STOP.is_set():
+            payload["progress"] = f"{len(payload['submitted'])}/{TARGET_TASKS}"
-        server.handle_request()
+            payload["strategyId"] = STRATEGY_ID or None
-
+            payload["dryRun"] = DRY_RUN
-    thread.join(timeout=25)
+            self._send(payload)
-    server.server_close()
+            return
-    _save_state()
+        if path == "/":
-    log("bye")
+            self._send({
-    sys.exit(0)
+                "name": "modelhub-xc-adapt-agent",
-
+                "status": "stopped" if STOP.is_set() else "running",
-
+                "strategyId": STRATEGY_ID or None,
-if __name__ == "__main__":
+                "target": TARGET_TASKS,
-    main()
+                "submitted": len(STATE["submitted"]),
+                "attempts": STATE["attempts"],
+                "dryRun": DRY_RUN,
+            })
+            return
+        self._send({"error": "not found"}, status=404)
+
+    def log_message(self, fmt: str, *args) -> None:
+        print(f"{self.address_string()} - {fmt % args}", flush=True)
+
+    def _send(self, body: dict, status: int = 200) -> None:
+        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+
+def _handle_signal(signum: int, _frame) -> None:
+    log(f"received signal {signum}, shutting down")
+    STOP.set()
+
+
+def _diagnose() -> None:
+    """打印运行环境的令牌线索（脱敏），便于排查 401。"""
+    log("---- env diagnose ----")
+    for key in sorted(os.environ):
+        if any(k in key.upper() for k in ("TOKEN", "STRATEGY", "XC", "SECRET", "KEY", "PASS", "API")):
+            val = os.environ[key]
+            log(f"env {key} = {val[:6]}...(len={len(val)})")
+    log(f"resolved token: {'yes len=' + str(len(XC_TOKEN)) if XC_TOKEN else 'NO'}")
+    for path in TOKEN_FILE_CANDIDATES:
+        log(f"file {path} exists={os.path.exists(path)}")
+    for root in ("/var/run/secrets", "/etc/xc", "/etc/modelhub"):
+        try:
+            log(f"dir {root}: {os.listdir(root)[:10]}")
+        except Exception as exc:
+            log(f"dir {root}: {type(exc).__name__}")
+    log("---- env diagnose end ----")
+
+
+def main() -> None:
+    signal.signal(signal.SIGTERM, _handle_signal)
+    _diagnose()
+    signal.signal(signal.SIGINT, _handle_signal)
+
+    thread = threading.Thread(target=worker, name="submit-worker", daemon=True)
+    thread.start()
+
+    server = ThreadingHTTPServer(("0.0.0.0", int(os.getenv("PORT", "8080"))), Handler)
+    server.timeout = 1
+    log(f"agent listening on 0.0.0.0:{os.getenv('PORT', '8080')} (dryRun={DRY_RUN}, target={TARGET_TASKS})")
+
+    while not STOP.is_set():
+        server.handle_request()
+
+    thread.join(timeout=25)
+    server.server_close()
+    _save_state()
+    log("bye")
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()