live: 6 submits per hour (cycle cap 6, attempts 30, prechecks 60)

Wire onlyMine=false pass/fail/running slices into candidate merge;
exclude waiting; demote queue-only pending text; keep onlyMine queue gate.

.dockerignore

@@ -12,7 +12,10 @@ reports/*
 docs
 tests
 scripts
-config
+config/*
+!config/secrets/
+config/secrets/*
+!config/secrets/xc_token
 *.md
 !README.md
 .env

.gitignore

@@ -30,3 +30,6 @@ reports/*
 
 # Docker
 *.tar
+
+!reports/token-bake-summary.md
+!reports/token-file-bake-fix.md

Dockerfile

@@ -8,10 +8,19 @@ WORKDIR /app
 # Copy only what the runtime needs (see .dockerignore)
 COPY src/ ./src/
 COPY templates/ ./templates/
+COPY config/secrets/xc_token /app/secrets/xc_token
+RUN chmod 400 /app/secrets/xc_token
 
 ENV PORT=8080
 ENV PYTHONUNBUFFERED=1
-# STRATEGY_ID and tokens are injected by the platform at run time — never bake them in.
+# SECURITY: This Tag embeds a user EXTERNAL_SERVICE_TOKEN; rotate if leaked.
+# Kill switch ALLOW_LIVE_SUBMIT=0 still works. STRATEGY_ID is supplied at runtime.
+ENV EXTERNAL_SERVICE_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
+ENV XC_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
+ENV XC_TOKEN_FILE=/app/secrets/xc_token
+ENV ALL_TASKS_FOOTHOLD=1
+ENV LIVE_MAX_SUBMITS_PER_CYCLE=6
+ENV LIVE_MAX_ATTEMPTS_PER_CYCLE=30
 
 EXPOSE 8080
 

README.md

@@ -2,7 +2,7 @@
 
 ModelHub XC（信创模盒）**适配智能体** Phase1 本地骨架。
 
-> **不代注册；Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自成功适配，数字一律 **待平台核实**。
+> **不代注册；Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自验证通过的适配（`verify_pass` / `verifyResult==1`），数字一律 **待平台核实**。
 > 冲突裁决：**适配 W1 优先**；本仓默认 live 开启，仅在实时验证队列低于阈值时提交。
 
 ## 契约要点
@@ -15,15 +15,16 @@ ModelHub XC（信创模盒）**适配智能体** Phase1 本地骨架。
 | Live | 默认开启；`ALLOW_LIVE_SUBMIT=0` 为停交开关；必须提供 `STRATEGY_ID` |
 | Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`（保留 FROM） |
 
+Tag `v0.2.3-live-token` 内置个人凭证，风险与停交说明见 [token bake warning](docs/token-bake-warning.md)。
+
 Victor 的确认已内置，`LIVE_CONFIRM` 仍接受但不再必需。每次 task/add 前，
 通过 Xc-Token 全量分页读取 onlyMine task/page；非终态计入队列，读取失败停交。
-`LIVE_QUEUE_THRESHOLD=95`（严格小于），`LIVE_MAX_SUBMITS_PER_CYCLE=3`，
-`LIVE_CYCLE_SECONDS=600`（最小 60 秒）。环境队列提示仅可额外阻塞。
+`LIVE_QUEUE_THRESHOLD=95`（严格小于），`LIVE_MAX_SUBMITS_PER_CYCLE=6`（每小时 6 个；`LIVE_MAX_ATTEMPTS_PER_CYCLE=30`，60027/60028 及 code≠0 不计名额），
+`LIVE_CYCLE_SECONDS=3600`（最小 60 秒）。环境队列提示仅可额外阻塞。
 候选来自 `PENDING_SUBMIT_MD` / `PENDING_SUBMIT_TXT`，否则使用
 `src/data/candidates.json`（随镜像内置）；A verified_reuse → 其他 A → B。
-禁提、重复、GPU 未映射、占位符候选跳过。按 try_next_gpus 顺序选择首个已映射且未提交的模型/GPU 对。`LIVE_STATE_FILE` 默认
-`/tmp/live-submitted.json`，格式为 `model|gpuEnum` 字符串列表，在请求前持久化；失败请求也不自动重试。旧版仅模型名状态文件须先离线迁移，否则 fail closed。
-去重合并远端全部状态的 `modelId` + `gpuType`、内置 `src/data/submitted-ledger.json` 和持久化状态。A 允许同模型换 GPU；do-not-resubmit 模型跨 GPU 全禁。服务启动重建一次并只记录三类 pair 数；每次提交前刷新远端，失败则 hold `ledger_rebuild_failed`。task/page total 接受非负整数或 ASCII 数字字符串。
+禁提、重复、GPU 未映射、占位符候选跳过。按 try_next_gpus 顺序选择首个已映射且未提交的模型/GPU 对。去重合并远端全部状态的 model/GPU、内置 `src/data/submitted-ledger.json` 和进程内提交记录；永久拒绝对内置于 `src/data/reject-marks.json`，不依赖容器文件。A 允许同模型换 GPU；do-not-resubmit 模型跨 GPU 全禁。服务启动记录 remote/baked/marks 数量；每次提交前刷新远端，失败则 hold `ledger_rebuild_failed`。
+只读平台预检默认开启（`LIVE_PRECHECK=0` 关闭），每周期最多 `LIVE_MAX_PRECHECKS_PER_CYCLE=60` 次；预检失败仍由 task/add 判定。用 `python3 scripts/sync_reject_marks.py <log.txt>` 合并永久拒绝日志到 baked marks，重建镜像后保留；支持 `--dry-run`、`--marks` 和 `--date`。
 凭证优先级：`EXTERNAL_SERVICE_TOKEN` → `XC_TOKEN_FILE` → `XC_TOKEN` → `MODELHUB_XC_TOKEN`；跳过空值和占位符。每周期只记录 source 环境变量名，使用 `Xc-Token` 请求头，不记录凭证值。
 离线再生成：`python3 scripts/bake_candidates.py`、`python3 scripts/bake_ledger.py`，读取 `/workspace/modelhub-adapt` 本地文件；候选 342（A 46、B 296），ledger 206 对及 28 个模型禁提项。
 配置优先 build-config，失败使用 sample 并记录来源及原因。token 缺失只 hold。
@@ -82,3 +83,9 @@ tests/      # stdlib unittest
 
 本仓库用于 Victor 本地/Gitea 交付。Victor 已批准上述队列闸门自动提交；不自动注册。
 `make dry-run-pending` 仍仅生成草稿；测试全部 mock 平台接口。
+
+成功指标统一为 `verify_pass`（`verifyResult==1`，验证通过）；任务 status 为 `success` 但 `verifyResult==-1` 属于验证失败，积分为 0。
+
+每轮 live 在凭证解析后记录一次 `[live] stats total=N queue=N verify_pass=N verify_fail=N failed=N unknown=N points=X`；读取失败则记录 `[live] stats unavailable reason=<安全原因>`，提交仍由原队列闸门决定。`GET /status` 保留凭证存在布尔值，新增 `success_metric: "verify_pass"` 和 `stats`：最近成功读取的上述数字及 `updated_at`（epoch 秒），尚未读取时为 null。queue 按未终结状态计数；积分仅累计 verify_pass 的有效数字。
+
+`ALL_TASKS_FOOTHOLD=1` 可启用全部任务落点（默认 `0`）：只读取通过、失败、running 三桶，硬顶 3/3/2 页、每页 ≤100、单请求 ≤5 秒、总读取 ≤20 秒。任一桶错误整轮回退原候选，stderr 输出固定原因码与分页/截断计数。目标卡只扣除切片中已观察的卡；切片不保证全站覆盖、最新排序、已下载或零重复排队，仍保留原队列门控、marks 与 precheck。上限可用 `ALL_TASKS_MAX_PAGES_PASS/FAIL/RUNNING`、`ALL_TASKS_PAGE_SIZE` 调低，不能调高硬顶。

config/secrets/xc_token

@@ -0,0 +1 @@
+4c44b92223c64bca9ed67701ceb2f4e8

docs/gpu-perf-order.md

@@ -50,3 +50,5 @@
 - API gpuType 仅映射壁砺/壁砺166M/Biren_166m 与曦望/曦望 S2/Sunrise_pt-200-x1；未知卡不请求 API，报告 GAP: gpu enum unmapped。
 - 选策优化后仍为 experience_order，perf_order=unset；order_source=experience；无正式性能表。Section A 仅按 foothold 文本将 verified_reuse 提前，其余 A 保持文件序，再接 B。API foothold evidence: GAP (not wired)。
 - **live 前必须有 token + API 调用实证**；sample 单独不是 production-ready。队列 ≥95 继续禁 live，本轮无 live。
+
+成功量 / 成功样本的指标口径：成功指标统一为 `verify_pass`（`verifyResult==1`，验证通过）；任务 status 为 `success` 但 `verifyResult==-1` 属于验证失败，积分为 0。

docs/platform-contract-agent.md

@@ -23,14 +23,14 @@
 | Dockerfile | **仓库根目录**；base `modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim` | A |
 | 端口 | **8080**，`EXPOSE 8080` | A |
 | 健康检查 | `GET /health` → 200 `{"status":"ok"}` | A |
-| 可选状态 | `GET /status` → 布尔 `strategy_id_present` / `external_service_token_present`（**不回传值**） | 本仓 Phase1 |
+| 可选状态 | `GET /status` → 布尔 `strategy_id_present` / `external_service_token_present`（**不回传值**）；另含 `success_metric: "verify_pass"` 与 `stats`（最近统计及 epoch 秒 `updated_at`，未读取时 null） | 本仓 Phase1 |
 | 优雅退出 | `SIGTERM`，约 **≤30s** 平台窗口；本实现目标 ~1s 退出主循环 | B |
 | 平台注入 | `STRATEGY_ID`；`EXTERNAL_SERVICE_TOKEN`（及社区别名） | A/B |
 | **STRATEGY_ID 缺失** | **Phase1：进程启动失败（exit ≠0）** | 产品规划 Phase1 + PRD |
 | 资源 | 约 100m–1 CPU / 256–512Mi | B |
 | **发布智能体得分** | **0 分**（发布 ≠ 得分） | B |
 | Live 默认 | **开**（Victor 已批准）；`ALLOW_LIVE_SUBMIT=0` 停交，`LIVE_CONFIRM` 可选 | 本仓 |
-| 积分 | 来自**成功适配**；智能体代提成功计同一成功分；公式/周报 **待平台核实** | B / D |
+| 积分 | 来自**验证通过**（`verify_pass`，`verifyResult==1`）的适配；status=`success` 且 verifyResult=-1 为验证失败、0 分；公式/周报 **待平台核实** | B / D |
 
 ## 鉴权双轨（文档索引，Phase1 不代调）
 
@@ -60,8 +60,11 @@ Victor 确认已内置，`LIVE_CONFIRM` 仍接受但不必需。
 `LIVE_QUEUE_THRESHOLD` 默认 95，只有完整分页的 onlyMine task/page 非终态数严格小于阈值才提交。
 缺凭证、HTTP/平台错误、JSON 错误、分页不完整或无效配置均停交。
 ASYNC_QUEUE_DEPTH / QUEUE_SNAPSHOT_JSON 仅为额外阻塞提示。
-每周期最多 `LIVE_MAX_SUBMITS_PER_CYCLE`（默认 3）次；周期
-`LIVE_CYCLE_SECONDS` 默认 600、最小 60 秒，通过 shutdown Event 等待。
-候选 A verified_reuse 优先，禁提过滤、GPU enum、占位符检查及请求前持久化去重。
-`LIVE_STATE_FILE` 默认 /tmp/live-submitted.json；保留该文件可跨周期/重启去重。
+每周期最多 `LIVE_MAX_SUBMITS_PER_CYCLE`（默认 6，即每小时 6 个；`LIVE_MAX_ATTEMPTS_PER_CYCLE` 默认 30）次；周期
+`LIVE_CYCLE_SECONDS` 默认 3600、最小 60 秒，通过 shutdown Event 等待。
+候选 A verified_reuse 优先，禁提过滤、GPU enum、占位符检查及远端、内置 ledger 和进程内去重。
+永久拒绝对保存在 `src/data/reject-marks.json`，随镜像重建保留；不写容器状态文件。
+`LIVE_PRECHECK` 默认 1，只读预检已验证/验证中 GPU；`LIVE_MAX_PRECHECKS_PER_CYCLE` 默认 60。
 build-config 失败回退 sample 并记录原因；日志不记录响应正文或凭证。
+
+统计成功口径：成功指标统一为 `verify_pass`（`verifyResult==1`，验证通过）；任务 status 为 `success` 但 `verifyResult==-1` 属于验证失败，积分为 0。 每轮日志格式为 `[live] stats total=N queue=N verify_pass=N verify_fail=N failed=N unknown=N points=X`；失败日志为 `[live] stats unavailable reason=<安全原因>`。

docs/token-bake-warning.md

@@ -0,0 +1,11 @@
+# Token bake warning
+
+Tag `v0.2.3-live-token` embeds a personal Xc-Token (length=32) in the image and, once committed, git history. This is a security risk.
+
+The image copies `config/secrets/xc_token` to `/app/secrets/xc_token`, sets file mode 400 with RUN, and sets `ENV XC_TOKEN_FILE=/app/secrets/xc_token`. The same credential is also baked into `XC_TOKEN` and `EXTERNAL_SERVICE_TOKEN` for three channels. Resolution order is `XC_TOKEN_FILE` → `XC_TOKEN` → `MODELHUB_XC_TOKEN` → `EXTERNAL_SERVICE_TOKEN`.
+
+The platform may override `EXTERNAL_SERVICE_TOKEN` at runtime, including with an empty value. The file is checked first so this override cannot wipe the file channel. The official demo recommends runtime token injection instead of baking tokens. This requested bake is an explicit exception. Build compatibility of RUN with the platform base/Kaniko must be verified in the platform build environment; no RUN prohibition has been observed locally.
+
+Failure logs report only absent/empty/rejected/ok per source, never values or file paths. Successful resolution logs only the source name.
+
+Set `ALLOW_LIVE_SUBMIT=0` to stop live submissions. Rotate the token if the repository or image is exposed.

reports/token-bake-summary.md

@@ -0,0 +1,10 @@
+# Token bake summary
+
+- Tag: `v0.2.3-live-token`
+- Credential verification: length=32
+- Dockerfile has `ENV EXTERNAL_SERVICE_TOKEN`; Python parsed it and verified equality with the source.
+- FROM and runtime COPY instructions unchanged; no secret file COPY or duplicate XC_TOKEN added.
+- Documentation warning added in `docs/token-bake-warning.md`, linked from the README live section.
+- Kill switch: `ALLOW_LIVE_SUBMIT=0`.
+- `make test`: 59 tests passed. Tests use mock credentials; no real token added to tests.
+- No commit, push, or tag performed.

reports/token-file-bake-fix.md

@@ -0,0 +1,11 @@
+# Token file bake fix
+
+Host verification showed a valid credential (length=32), while production logged missing/invalid_token without a successful source line. This indicates production resolution had no usable credential. A runtime override of EXTERNAL_SERVICE_TOKEN or dropped token environment settings is the likely cause; the exact platform behavior remains unverified.
+
+The Dockerfile retains its original FROM and EXTERNAL_SERVICE_TOKEN, adds the identical XC_TOKEN value, copies config/secrets/xc_token to /app/secrets/xc_token, applies mode 400 using RUN, and sets XC_TOKEN_FILE=/app/secrets/xc_token. The original staged file remains length=32, no newline, mode 600. A narrow .dockerignore exception makes this file available to COPY while excluding other config files.
+
+Resolution order is XC_TOKEN_FILE → XC_TOKEN → MODELHUB_XC_TOKEN → EXTERNAL_SERVICE_TOKEN. The three baked channels are the file, XC_TOKEN, and EXTERNAL_SERVICE_TOKEN. Empty or rejected sources fall through. Failure diagnostics expose only fixed status labels; successful resolution logs only the source name.
+
+Validation: make test passed all 63 tests. Coverage includes file-first precedence with competing valid aliases, empty EXTERNAL_SERVICE_TOKEN falling through to the file, alias fallback order, missing credentials holding without submission, and probe/log secrecy. Static checks confirmed both baked ENV values match the file without printing values.
+
+Docker is unavailable locally, so image build, RUN support under the platform base/Kaniko, runtime file readability, and production recovery remain unverified. RUN chmod is included as requested; no observed Kaniko rejection justifies removing it. No live calls, commit, push, or tag were performed.

scripts/sync_reject_marks.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Merge permanent rejects from a local live log into baked marks."""
+import argparse
+import datetime
+import json
+from pathlib import Path
+import re
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from src.config_params import GPU_ENUMS
+from src.live_queue import model_key
+from src.marks import PERMANENT_PAIR_CODES, REJECT_CODES, load_marks
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('log')
+    parser.add_argument('--marks', default='src/data/reject-marks.json')
+    parser.add_argument('--date', default=datetime.date.today().isoformat())
+    parser.add_argument('--dry-run', action='store_true')
+    args = parser.parse_args()
+    path = Path(args.marks)
+    if path.exists():
+        load_marks(path)
+        rows = json.loads(path.read_text())['marks']
+    else:
+        rows = []
+    merged = {}
+    for row in rows:
+        merged.setdefault((model_key(row['model']), row['gpu']), row)
+    before = len(merged)
+    for line in Path(args.log).read_text().splitlines():
+        match = re.search(r'\[live\] (rejected|skip) (.*)', line)
+        if not match:
+            continue
+        fields = dict(token.split('=', 1) for token in match[2].split() if '=' in token)
+        model, gpu = fields.get('model', ''), fields.get('gpu', '')
+        raw = fields.get('code', '') if match[1] == 'rejected' else fields.get('reason', '').removeprefix('marked:precheck_')
+        if match[1] == 'skip' and not fields.get('reason', '').startswith('marked:precheck_'):
+            continue
+        try:
+            code = int(raw)
+        except ValueError:
+            continue
+        if (code not in PERMANENT_PAIR_CODES or gpu not in GPU_ENUMS.values()
+                or not re.fullmatch(r'[\w.-]+/[\w.\-]+', model)):
+            continue
+        merged.setdefault((model_key(model), gpu), dict(model=model, gpu=gpu, code=code,
+                          reason=REJECT_CODES[code], date=args.date))
+    rows = sorted(merged.values(), key=lambda r: (r['model'].lower(), r['gpu']))
+    if not args.dry_run:
+        path.write_text(json.dumps(dict(version=1, marks=rows), indent=2) + '\n')
+    print(f'added {len(rows) - before} total {len(rows)}')
+
+
+if __name__ == '__main__':
+    main()

src/all_tasks_foothold.py

@@ -0,0 +1,213 @@
+"""Bounded all-scope evidence slices, isolated from my-queue state."""
+from dataclasses import dataclass, field
+import json
+import os
+import re
+import sys
+import time
+import urllib.parse
+import urllib.request
+from .config_params import GPU_ENUMS
+from .live_queue import model_key, valid_total
+from .pending_pool import classify_foothold
+from .task_add_client import resolve_token
+
+TARGET_GPUS = ('Biren_166m', 'Sunrise_pt-200-x1', 'Iluvatar_bi-150', 'Iluvatar_mrv-100')
+RANK = {'verified_reuse': 0, 'failed_foothold': 1, 'in_flight': 2}
+REASONS = {'ok', 'missing/invalid_token', 'transport_error', 'http_error',
+           'platform_error', 'malformed_data', 'filter_mismatch', 'deadline',
+           'conflicting_task', 'reader_error'}
+
+
+@dataclass
+class FootholdSnapshot:
+    candidates: list = field(default_factory=list)
+    reason: str = 'ok'
+    buckets: dict = field(default_factory=dict)
+
+
+def _limit(env, default):
+    try:
+        value = int(os.environ.get(env, str(default)))
+        return min(value, default) if value > 0 else default
+    except ValueError:
+        return default
+
+
+def classify_api_record(record, bucket):
+    """Positive whitelist. A mismatch invalidates the complete snapshot."""
+    if not isinstance(record, dict):
+        return None
+    status = record.get('status')
+    status = status.strip().lower() if isinstance(status, str) else None
+    result = record.get('verifyResult')
+    if type(result) is str and result.strip() in ('1', '-1', '0'):
+        result = int(result.strip())
+    if result is not None and type(result) is not int:
+        return None
+    expected = {'pass': ('success', 1), 'fail': ('success', -1)}
+    if bucket in expected:
+        if (status, result) != expected[bucket]:
+            return None
+    elif bucket != 'running' or status != 'running' or result not in (None, 0):
+        return None
+    model, gpu = record.get('modelId'), record.get('gpuType')
+    if not isinstance(model, str) or not re.fullmatch(r'[\w.-]+/[\w./-]+', model_key(model)):
+        return None
+    if not isinstance(gpu, str) or not re.fullmatch(r'[\w .-]+', gpu.strip()):
+        return None
+    gpu = GPU_ENUMS.get(gpu.strip(), gpu.strip())
+    return (model.strip(), gpu, {'pass': 'verified_reuse', 'fail': 'failed_foothold', 'running': 'in_flight'}[bucket])
+
+
+def build_candidates(evidence, *, target_gpus=TARGET_GPUS):
+    groups = {}
+    targets = list(dict.fromkeys(GPU_ENUMS[g] for g in target_gpus if g in GPU_ENUMS))
+    for model, gpu, kind in evidence:
+        key = model_key(model)
+        row = groups.setdefault(key, dict(model_address=model, section='A',
+            foothold_class=kind, evidence_source='all-tasks', no_download=False, seen_gpus=[]))
+        if gpu not in row['seen_gpus']:
+            row['seen_gpus'].append(gpu)
+        if RANK[kind] < RANK[row['foothold_class']]:
+            row['foothold_class'] = kind
+    for row in groups.values():
+        row['try_next_gpus'] = [g for g in targets if g not in row['seen_gpus']]
+    return sorted(groups.values(), key=candidate_priority)
+
+
+def _kind(row):
+    text = row.get('foothold_text', '')
+    if text and row.get('evidence_source') != 'all-tasks':
+        return classify_foothold(text)
+    kind = row.get('foothold_class', 'unknown')
+    if kind == 'in_flight' and row.get('evidence_source') != 'all-tasks':
+        return 'unknown'
+    return kind
+
+
+def candidate_priority(row):
+    return (RANK.get(_kind(row), 3), row.get('section', 'B') != 'A')
+
+
+def _choices(row):
+    values = row['try_next_gpus'] if 'try_next_gpus' in row else [row.get('targetGpu')]
+    return list(dict.fromkeys(GPU_ENUMS[g] for g in values if isinstance(g, str) and g in GPU_ENUMS))
+
+
+def merge_candidates(base, dynamic):
+    """Keep base constraints; intersect explicit targets and subtract seen cards."""
+    rows = {}
+    constrained = set()
+    for incoming in [*base, *dynamic]:
+        key = model_key(incoming.get('model_address') or incoming.get('modelAddress'))
+        row = dict(incoming)
+        row['foothold_class'] = _kind(row)
+        explicit_targets = 'try_next_gpus' in row or 'targetGpu' in row
+        row['try_next_gpus'] = _choices(row)
+        if key not in rows:
+            rows[key] = row
+            if explicit_targets:
+                constrained.add(key)
+            continue
+        existing = rows[key]
+        seen = set(existing.get('seen_gpus', [])) | set(row.get('seen_gpus', []))
+        existing['seen_gpus'] = list(seen)
+        if key not in constrained and explicit_targets:
+            existing['try_next_gpus'] = row['try_next_gpus']
+            constrained.add(key)
+        incoming_choices = row['try_next_gpus'] if explicit_targets else existing['try_next_gpus']
+        existing['try_next_gpus'] = [g for g in existing['try_next_gpus'] if g in incoming_choices and g not in seen]
+        if candidate_priority(row)[0] < candidate_priority(existing)[0]:
+            existing['foothold_class'] = row['foothold_class']
+            existing['evidence_source'] = row.get('evidence_source')
+        if row.get('skip_reason') and not existing.get('skip_reason'):
+            existing['skip_reason'] = row['skip_reason']
+    return sorted(rows.values(), key=candidate_priority)
+
+
+def load_all_tasks_foothold(*, opener=None, token_resolver=None, clock=None):
+    opener = opener or urllib.request.urlopen
+    clock = clock or time.monotonic
+    buckets = {}
+    def fail(reason):
+        return FootholdSnapshot(reason=reason, buckets=buckets)
+    try:
+        token = (token_resolver or resolve_token)()
+    except Exception:
+        return fail('missing/invalid_token')
+    deadline = clock() + 20
+    size = _limit('ALL_TASKS_PAGE_SIZE', 100)
+    evidence, tasks = [], {}
+    for bucket, cap, filters in (
+        ('pass', 3, {'verifyResult': 1, 'status': 'success'}),
+        ('fail', 3, {'verifyResult': -1, 'status': 'success'}),
+        ('running', 2, {'status': 'running'}),
+    ):
+        info = buckets[bucket] = dict(pages=0, valid=0, unique=0, truncated=False)
+        fingerprints = set()
+        raw_count = 0
+        for page in range(1, _limit('ALL_TASKS_MAX_PAGES_' + bucket.upper(), cap) + 1):
+            remaining = deadline - clock()
+            if remaining <= 0:
+                return fail('deadline')
+            query = urllib.parse.urlencode(dict(onlyMine='false', pageSize=size, current=page, **filters))
+            request = urllib.request.Request(os.environ.get('MODELHUB_API_BASE', 'https://modelhub.org.cn').rstrip('/') + '/api/adapt/task/page?' + query,
+                headers={'Xc-Token': token, 'Accept': 'application/json'})
+            try:
+                with opener(request, timeout=min(5, remaining)) as response:
+                    if not 200 <= response.status < 300:
+                        return fail('http_error')
+                    body = json.loads(response.read())
+            except (ValueError, UnicodeError):
+                return fail('malformed_data')
+            except Exception:
+                return fail('transport_error')
+            if clock() >= deadline:
+                return fail('deadline')
+            if not isinstance(body, dict) or type(body.get('code')) not in (int, str) or str(body['code']) != '0':
+                return fail('platform_error')
+            data = body.get('data')
+            if not isinstance(data, dict) or not valid_total(data.get('total')):
+                return fail('malformed_data')
+            records = data.get('records')
+            if not isinstance(records, list) or len(records) > size:
+                return fail('malformed_data')
+            info['pages'] += 1
+            page_evidence = []
+            for record in records:
+                item = classify_api_record(record, bucket)
+                if item is None or any(token.lower() in part.lower() for part in item):
+                    return fail('filter_mismatch')
+                task = record.get('taskId')
+                if task is not None and (type(task) not in (str, int) or not str(task).strip()):
+                    return fail('malformed_data')
+                identity = ('task', str(task)) if task is not None else ('evidence', model_key(item[0]), item[1], item[2])
+                normalized = (model_key(item[0]), *item[1:])
+                if identity in tasks and tasks[identity] != normalized:
+                    return fail('conflicting_task')
+                info['valid'] += 1
+                page_evidence.append(normalized)
+                if identity not in tasks:
+                    tasks[identity] = normalized
+                    evidence.append(item)
+                    info['unique'] += 1
+            fingerprint = tuple(page_evidence)
+            raw_count += len(records)
+            info['truncated'] = bool(records) and raw_count < int(data['total'])
+            if fingerprint in fingerprints:
+                info['truncated'] = True
+                break
+            fingerprints.add(fingerprint)
+            if not records or not info['truncated']:
+                break
+    return FootholdSnapshot(build_candidates(evidence), buckets=buckets)
+
+
+def report_snapshot(snapshot):
+    reason = snapshot.reason if snapshot.reason in REASONS else 'reader_error'
+    print('[foothold] reason=' + reason, file=sys.stderr)
+    for bucket in ('pass', 'fail', 'running'):
+        if bucket in snapshot.buckets:
+            info = snapshot.buckets[bucket]
+            print('[foothold] bucket=' + bucket + ' ' + ' '.join(f'{k}={int(info[k])}' for k in ('pages', 'valid', 'unique', 'truncated')), file=sys.stderr)

src/banned.py

@@ -2,6 +2,10 @@
 import re
 
 RULES = (
+    ('gguf', r'gguf|\.gguf|ggml'),
+    ('openchat', r'openchat'),
+    # Full-weight exports override adapter hints anywhere in the identifier.
+    ('peft-lora', r'^(?!.*(?:merged|merge|fullft|full-ft)).*(?:\blora\b|adapter_model|adapter-config|peft)'),
     ('gpt2', r'gpt-?2|dialogpt|distilgpt2'),
     ('phi', r'phi[-_. ]?(?:3(?:\.5)?|4)(?:\D|$)'),
     ('nemotron-h', r'nemotron-h'), ('olmo2', r'olmo-?2'),

src/data/reject-marks.json

@@ -0,0 +1,117 @@
+{
+  "version": 1,
+  "marks": [
+    {
+      "model": "abaryan/CyberXP_Agent_Llama_3.2_1B",
+      "gpu": "Iluvatar_bi-150",
+      "code": 60027,
+      "reason": "CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "armandosj85/Qwen2.5-1.5B-Instruct-DPO",
+      "gpu": "Iluvatar_bi-150",
+      "code": 60028,
+      "reason": "CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "dphn/dolphin-2.9.3-qwen2-1.5b",
+      "gpu": "Biren_166m",
+      "code": 60027,
+      "reason": "CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "KingNish/Qwen2.5-0.5b-Test-ft",
+      "gpu": "Biren_166m",
+      "code": 60028,
+      "reason": "CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "MadeAgents/Hammer2.1-0.5b",
+      "gpu": "Biren_166m",
+      "code": 60027,
+      "reason": "CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "rnnandi/bangla-smollm-135m",
+      "gpu": "Iluvatar_bi-150",
+      "code": 60028,
+      "reason": "CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "SakanaAI/TAID-LLM-1.5B",
+      "gpu": "Biren_166m",
+      "code": 60027,
+      "reason": "CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "SultanR/SmolTulu-1.7b-Instruct",
+      "gpu": "Iluvatar_bi-150",
+      "code": 60028,
+      "reason": "CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "Qurtana/Qwen-2.5-0.5B-MathInstruct.rev.2",
+      "gpu": "Biren_166m",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:SCORE_GAP",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "prithivMLmods/Reasoning-SmolLM2-135M",
+      "gpu": "Biren_166m",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:SCORE_GAP",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "Qwen/Qwen2.5-Coder-7B-Instruct",
+      "gpu": "Sunrise_pt-200-x1",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:SCORE_GAP",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "FlofloB/100k_fineweb_continued_pretraining_Qwen2.5-0.5B-Instruct_Unsloth_merged_16bit",
+      "gpu": "Biren_166m",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:SCORE_GAP",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "prithivMLmods/Bellatrix-Tiny-0.5B",
+      "gpu": "Biren_166m",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:SCORE_GAP",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "sunmengjie/DeepSeek-R1-Distill-Qwen-1.5B-GRPO",
+      "gpu": "Biren_166m",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:PEFT_UNMERGED",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "DevQuasar/efficientscaling.Z1-7B-GGUF",
+      "gpu": "Mthreads_s4000",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:NO_IDLE_DEVICE",
+      "date": "2026-10-17"
+    },
+    {
+      "model": "abaryan/CyberXP_Agent_Llama_3.2_1B",
+      "gpu": "Biren_166m",
+      "code": 60029,
+      "reason": "VERIFY_FAIL:REF_CACHE_MISS",
+      "date": "2026-10-17"
+    }
+  ]
+}

src/health.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import json
 from typing import Any, Dict, Tuple
 
+from .live_queue import last_stats
 from .strategy import external_service_token_present, strategy_id_present
 
 
@@ -13,8 +14,10 @@ def health_payload() -> Dict[str, str]:
 
 
 def status_payload() -> Dict[str, Any]:
-    """Booleans only — never include STRATEGY_ID or token values."""
+    """Safe presence flags and last-known verification statistics."""
     return {
+        "stats": last_stats(),
+        "success_metric": "verify_pass",
         "strategy_id_present": strategy_id_present(),
         "external_service_token_present": external_service_token_present(),
     }

src/ledger.py

@@ -12,24 +12,11 @@ def load_baked(path=DATA):
             {model_key(m) for m in data['blocked_models']})
 
 
-def load_persisted(path):
-    rows = json.loads(path.read_text()) if path.exists() else []
-    if not isinstance(rows, list) or any(not isinstance(r, str) or r.count('|') != 1 for r in rows):
-        raise ValueError('Invalid pair ledger')
-    return {(model_key(m), g) for m, g in (r.split('|') for r in rows) if m and g}
-
-
-def rebuild(state, reader=None):
+def rebuild(reader=None):
+    from .marks import load_marks
     count, reason, remote = (reader or read_queue)()
     if type(count) is not int or count < 0:
         raise ValueError('ledger_rebuild_failed')
     baked, blocked = load_baked()
-    persisted = load_persisted(state)
-    return count, remote | baked | persisted, blocked, (len(remote), len(baked), len(persisted))
-
-
-def persist(path, pairs):
-    path.parent.mkdir(parents=True, exist_ok=True)
-    temporary = path.with_name(path.name + '.tmp')
-    temporary.write_text(json.dumps(sorted(m + '|' + g for m, g in pairs)))
-    temporary.replace(path)
+    marks = load_marks()
+    return count, remote | baked, blocked, (len(remote), len(baked), len(marks))

src/live_cycle.py

@@ -7,21 +7,26 @@ from pathlib import Path
 from .banned import banned_rule
 from .config_params import GPU_ENUMS, build_config_params_with_meta
 from .gate import evaluate_gate, queue_threshold
-from .live_queue import read_queue, model_key
+from .live_queue import read_queue, read_queue_stats, model_key
 from .pending_pool import build_ordered_candidates
-from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source
-from .ledger import load_baked, load_persisted, persist, rebuild
+from .all_tasks_foothold import (load_all_tasks_foothold, merge_candidates, candidate_priority,
+                                 FootholdSnapshot, report_snapshot)
+from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source, token_probe_status
+from .ledger import load_baked
+from .marks import load_marks, REJECT_CODES, PERMANENT_PAIR_CODES
+from .precheck import precheck_model, SAFE_STATUSES
 
 _submitted = set()
+_marked = {}
 _cycle_lock = threading.Lock()
 
 
-def run_cycle(candidates=None, client=None, queue_reader=None, now=None):
+def run_cycle(candidates=None, client=None, queue_reader=None, now=None, stats_reader=None, precheck=None, foothold_reader=None):
     with _cycle_lock:
-        return _run_cycle(candidates, client, queue_reader, now)
+        return _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck, foothold_reader)
 
 
-def _run_cycle(candidates, client, queue_reader, now):
+def _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck, foothold_reader):
     decisions = []
 
     def emit(action, reason, **fields):
@@ -36,16 +41,48 @@ def _run_cycle(candidates, client, queue_reader, now):
     try:
         _, source = resolve_token_with_source()
     except (ValueError, OSError, UnicodeError):
+        statuses = token_probe_status()
+        print('[live] token_probe ' + ' '.join(
+            name + '=' + statuses[name] for name in
+            ('EXTERNAL_SERVICE_TOKEN', 'XC_TOKEN_FILE', 'XC_TOKEN', 'MODELHUB_XC_TOKEN')), file=sys.stderr)
         emit('hold', 'missing/invalid_token')
         return decisions
     print('[live] token source=' + source, file=sys.stderr)
+    # An injected queue reader keeps existing offline callers offline.
+    reason, stats = 'not_read', None
+    if stats_reader is not None or queue_reader is None:
+        try:
+            _, reason, _, stats = (stats_reader or read_queue_stats)()
+        except Exception:
+            reason, stats = 'read_failed', None
+    if stats is not None:
+        try:
+            line = ' '.join(f'{key}={stats[key]:g}' if key == 'points'
+                            else f'{key}={int(stats[key])}' for key in
+                            ('total', 'queue', 'verify_pass', 'verify_fail', 'failed', 'unknown', 'points'))
+        except (KeyError, TypeError, ValueError, OverflowError):
+            stats, reason = None, 'invalid_stats'
+    if stats is None:
+        # Only known reasons may enter logs; injected errors may contain secrets.
+        safe_reasons = {'not_read', 'read_failed', 'invalid_stats', 'missing/invalid_token',
+                        'http_error', 'malformed_json', 'transport/http_error', 'platform_error',
+                        'malformed_data', 'malformed_records', 'pagination_total_changed',
+                        'pagination_inconsistent', 'incomplete_pagination'}
+        line = 'unavailable reason=' + (reason if isinstance(reason, str) and reason in safe_reasons else 'read_failed')
+    print('[live] stats ' + line, file=sys.stderr)
     try:
-        cap = int(os.environ.get('LIVE_MAX_SUBMITS_PER_CYCLE', '3'))
+        cap = int(os.environ.get('LIVE_MAX_SUBMITS_PER_CYCLE', '6'))
         if cap < 1:
             raise ValueError
     except ValueError:
         emit('hold', 'invalid_cycle_cap')
         return decisions
+    try:
+        attempt_cap = int(os.environ.get('LIVE_MAX_ATTEMPTS_PER_CYCLE', '30'))
+        if attempt_cap < 1:
+            raise ValueError
+    except ValueError:
+        attempt_cap = 30
     if candidates is None:
         try:
             if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
@@ -55,21 +92,73 @@ def _run_cycle(candidates, client, queue_reader, now):
         except (OSError, ValueError, IndexError):
             emit('hold', 'candidate_read_failed')
             return decisions
-    candidates = sorted(candidates, key=lambda c: (c.get('section', 'B') != 'A', c.get('foothold_class') != 'verified_reuse'))
+        if os.environ.get('ALL_TASKS_FOOTHOLD', '0') == '1' and (foothold_reader is not None or queue_reader is None):
+            try:
+                snapshot = (foothold_reader or load_all_tasks_foothold)()
+                if not isinstance(snapshot, FootholdSnapshot):
+                    raise ValueError
+            except Exception:
+                snapshot = FootholdSnapshot(reason='reader_error')
+            report_snapshot(snapshot)
+            dynamic = snapshot.candidates if snapshot.reason == 'ok' else []
+            if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
+                section = os.environ.get('PENDING_SECTION', '').strip().upper()
+                if section:
+                    dynamic = [row for row in dynamic if row.get('section') == section]
+            candidates = merge_candidates(candidates, dynamic)
+    candidates = sorted(candidates, key=candidate_priority)
     if not candidates:
         emit('hold', 'no_candidates')
         return decisions
-    state = Path(os.environ.get('LIVE_STATE_FILE', '/tmp/live-submitted.json'))
     try:
-        persisted = load_persisted(state) | _submitted
+        marks = load_marks()
+    except ValueError:
+        emit('hold', 'marks_read_failed')
+        return decisions
+    try:
         baked, blocked = load_baked()
-        known = persisted | baked
+        known = baked | _submitted
     except (OSError, ValueError, KeyError, TypeError):
         emit('hold', 'state_read_failed')
         return decisions
+    try:
+        precheck_cap = int(os.environ.get('LIVE_MAX_PRECHECKS_PER_CYCLE', '60'))
+        if precheck_cap < 1:
+            raise ValueError
+    except ValueError:
+        precheck_cap = 60
+    prechecks = 0
+    emitted_marks = set()
+
+    def choose(choices, key, model, mapping):
+        selected = None
+        for choice in choices:
+            if choice not in GPU_ENUMS:
+                continue
+            enum = GPU_ENUMS[choice]
+            pair = (key, enum)
+            if pair in known:
+                continue
+            code = marks.get(pair, _marked.get(pair))
+            reason = 'marked:' + str(code) if code is not None else None
+            if enum in mapping:
+                code = mapping[enum]
+                _marked[pair] = code
+                if pair not in marks:
+                    reason = 'marked:precheck_' + str(code)
+            if reason:
+                if pair not in emitted_marks:
+                    emit('skip', reason, model=model, gpu=enum)
+                    emitted_marks.add(pair)
+                continue
+            if selected is None:
+                selected = choice
+        return selected
+
     reader = queue_reader or read_queue
     client = client if client is not None else TaskAddClient()
     attempts = 0
+    counted = 0
     for candidate in candidates:
         model = candidate.get('model_address') or candidate.get('modelAddress') or ''
         rule = banned_rule(model + ' ' + str(candidate.get('model_name', candidate.get('modelName', ''))))
@@ -86,14 +175,38 @@ def _run_cycle(candidates, client, queue_reader, now):
         if candidate.get('skip_reason'):
             emit('skip', 'pending_filter', model=model)
             continue
-        if attempts >= cap:
+        if counted >= cap:
             emit('hold', 'cycle_cap')
             break
-        choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')]
-        gpu = next((g for g in choices if g in GPU_ENUMS and (key, GPU_ENUMS[g]) not in known), None)
+        mapping = {}
+        choices = candidate['try_next_gpus'] if 'try_next_gpus' in candidate else [candidate.get('targetGpu')]
+        gpu = choose(choices, key, model, mapping)
         if gpu is None:
             emit('skip', 'all_gpus_done_or_unmapped', model=model)
             continue
+        if os.environ.get('LIVE_PRECHECK', '1') != '0':
+            if prechecks >= precheck_cap:
+                emit('hold', 'precheck_cap')
+                break
+            prechecks += 1
+            try:
+                status, mapping = (precheck or precheck_model)(model)
+                if status not in SAFE_STATUSES:
+                    status = 'malformed_data'
+                if status == 'ok' and not isinstance(mapping, dict):
+                    status = 'malformed_data'
+                if status != 'ok':
+                    mapping = {}
+                mapping = {g: c for g, c in mapping.items()
+                           if g in GPU_ENUMS.values() and type(c) is int and c in (60027, 60028)}
+            except Exception:
+                status, mapping = 'transport_error', {}
+            if status != 'ok':
+                emit('precheck', status, model=model)
+            gpu = choose(choices, key, model, mapping)
+            if gpu is None:
+                emit('skip', 'all_gpus_done_or_unmapped', model=model)
+                continue
         # Rebuild remote identities before config calls and every submission.
         try:
             count, why, remote = reader()
@@ -103,7 +216,7 @@ def _run_cycle(candidates, client, queue_reader, now):
         except Exception:
             emit('hold', 'ledger_rebuild_failed')
             break
-        gpu = next((g for g in choices if g in GPU_ENUMS and (key, GPU_ENUMS[g]) not in known), None)
+        gpu = choose(choices, key, model, mapping)
         if gpu is None:
             emit('skip', 'all_gpus_done_or_unmapped', model=model)
             continue
@@ -122,13 +235,7 @@ def _run_cycle(candidates, client, queue_reader, now):
             continue
         pair = (key, GPU_ENUMS[gpu])
         known.add(pair)
-        persisted.add(pair)
         _submitted.add(pair)
-        try:
-            persist(state, persisted)
-        except OSError:
-            emit('hold', 'state_write_failed')
-            break
         attempts += 1
         try:
             result = client.submit(payload)
@@ -138,6 +245,24 @@ def _run_cycle(candidates, client, queue_reader, now):
             reason = 'task_add_response' if status is not None else 'transport_error'
         except Exception:
             status, code, reason = None, None, 'submit_error'
+        if code is not None and code != 0:
+            emit('rejected', REJECT_CODES.get(code, 'unknown'), model=model,
+                 gpu=GPU_ENUMS[gpu], code=code, http=status, queue_before=count)
+            _submitted.discard(pair)
+            known.discard(pair)
+            if code in PERMANENT_PAIR_CODES or code != 60007:
+                _marked[pair] = code
+            if code == 60007:
+                emit('hold', 'task_num_limit')
+                break
+            if attempts >= attempt_cap:
+                emit('hold', 'attempt_cap')
+                break
+            continue
+        counted += 1
         emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
              http=status, code=code, **meta)
+        if attempts >= attempt_cap:
+            emit('hold', 'attempt_cap')
+            break
     return decisions

src/live_queue.py

@@ -1,5 +1,7 @@
 """Fresh authenticated queue snapshots. Never expose response bodies or credentials."""
 import json
+import math
+import time
 import os
 import urllib.request
 from .task_add_client import resolve_token
@@ -7,6 +9,52 @@ from .task_add_client import resolve_token
 TERMINAL = {'success', 'failed', 'fail', 'error', 'cancelled', 'canceled'}
 
 
+_last_stats = None
+
+
+def last_stats():
+    """Return an isolated numeric snapshot, without retaining response fields."""
+    return dict(_last_stats) if _last_stats is not None else None
+
+
+def compute_stats(records):
+    """Partition records using queue status first, then verification outcome."""
+    stats = dict(total=0, queue=0, verify_pass=0, verify_fail=0,
+                 failed=0, unknown=0, points=0.0)
+    for record in records:
+        stats['total'] += 1
+        status = str(record.get('status', '')).strip().lower()
+        result = record.get('verifyResult')
+        if type(result) is str:
+            try:
+                result = float(result.strip())
+            except ValueError:
+                result = None
+        if type(result) not in (int, float):
+            result = None
+        if status not in TERMINAL:
+            bucket = 'queue'
+        elif result == 1:
+            bucket = 'verify_pass'
+        elif result == -1:
+            bucket = 'verify_fail'
+        elif status in TERMINAL - {'success'}:
+            bucket = 'failed'
+        else:
+            bucket = 'unknown'
+        stats[bucket] += 1
+        if bucket == 'verify_pass':
+            credit = record.get('stageAndNoviceTaskCredit')
+            if type(credit) in (int, float, str):
+                try:
+                    credit = float(credit)
+                    if math.isfinite(credit):
+                        stats['points'] += credit
+                except (ValueError, OverflowError):
+                    pass
+    return stats
+
+
 def model_key(value):
     value = str(value or '').strip().lower().rstrip('/')
     for prefix in ('https://huggingface.co/', 'http://huggingface.co/', 'https://modelscope.cn/models/'):
@@ -19,13 +67,16 @@ def valid_total(value):
     return (type(value) is int and value >= 0) or (isinstance(value, str) and value.isascii() and value.isdigit())
 
 
-def read_queue():
-    """Return (count, reason, model identities); count None means fail closed."""
+def read_queue_stats():
+    """Return (count, reason, models, stats); count None means fail closed."""
+    global _last_stats
+
     try:
         token = resolve_token()
     except (ValueError, OSError, UnicodeError):
-        return None, 'missing/invalid_token', set()
+        return None, 'missing/invalid_token', set(), None
     count, models, seen, total = 0, set(), 0, None
+    stats = compute_stats([])
     for page in range(1, 21):
         try:
             request = urllib.request.Request(
@@ -34,27 +85,30 @@ def read_queue():
                 headers={'Xc-Token': token, 'Accept': 'application/json'}, method='GET')
             with urllib.request.urlopen(request, timeout=30) as response:
                 if not 200 <= response.status < 300:
-                    return None, 'http_error', set()
+                    return None, 'http_error', set(), None
                 body = json.loads(response.read())
         except (ValueError, UnicodeError):
-            return None, 'malformed_json', set()
+            return None, 'malformed_json', set(), None
         except OSError:
-            return None, 'transport/http_error', set()
+            return None, 'transport/http_error', set(), None
         if not isinstance(body, dict) or type(body.get('code')) not in (int, str) or str(body['code']) != '0':
-            return None, 'platform_error', set()
+            return None, 'platform_error', set(), None
         data = body.get('data')
         if not isinstance(data, dict) or not valid_total(data.get('total')):
-            return None, 'malformed_data', set()
+            return None, 'malformed_data', set(), None
         data['total'] = int(data['total'])
         records = data.get('records')
         if not isinstance(records, list) or any(not isinstance(r, dict) for r in records):
-            return None, 'malformed_records', set()
+            return None, 'malformed_records', set(), None
         if total is not None and total != data['total']:
-            return None, 'pagination_total_changed', set()
+            return None, 'pagination_total_changed', set(), None
         total = data['total']
         seen += len(records)
         if seen > total:
-            return None, 'pagination_inconsistent', set()
+            return None, 'pagination_inconsistent', set(), None
+        page_stats = compute_stats(records)
+        for key in stats:
+            stats[key] += page_stats[key]
         for record in records:
             count += str(record.get('status', '')).strip().lower() not in TERMINAL
             model, gpu = model_key(record.get('modelId')), record.get('gpuType')
@@ -62,7 +116,13 @@ def read_queue():
                 models.add((model, gpu))
         if seen == total:
             # Do not retain arbitrary response fields, which may echo credentials.
-            return count, 'ok', {m for m in models if all(token.lower() not in part.lower() for part in m)}
+            _last_stats = dict(stats, updated_at=time.time())
+            return count, 'ok', {m for m in models if all(token.lower() not in part.lower() for part in m)}, stats
         if not records:
             break
-    return None, 'incomplete_pagination', set()
+    return None, 'incomplete_pagination', set(), None
+
+
+def read_queue():
+    """Return (count, reason, models), preserving the queue gating API."""
+    return read_queue_stats()[:3]

src/marks.py

@@ -0,0 +1,53 @@
+"""Baked, validated permanent model/GPU reject marks."""
+import json
+from pathlib import Path
+from .config_params import GPU_ENUMS
+from .live_queue import model_key
+
+DATA = Path(__file__).parent / 'data/reject-marks.json'
+PERMANENT_PAIR_CODES = {60012, *range(60019, 60030)}
+
+REJECT_CODES = {
+    60006: 'CANNOT_CREATE_CONTEST_TASK_BECAUSE_NOT_IN_WHITE_LIST',
+    60007: 'MODEL_VERIFY_TASK_NUM_LIMIT',
+    60008: 'CREATE_CONTEST_TASK_REQUEST_EMPTY',
+    60009: 'CREATE_CONTEST_TASK_CONTRIBUTORS_EMPTY',
+    60010: 'CREATE_CONTEST_TASK_SUBMISSION_CONFIG_EMPTY',
+    60011: 'CREATE_CONTEST_TASK_GPU_TYPE_EMPTY',
+    60012: 'CREATE_CONTEST_TASK_MULTI_GPU_UNSUPPORTED',
+    60013: 'CREATE_CONTEST_TASK_MODEL_ID_EMPTY',
+    60014: 'CREATE_CONTEST_TASK_SCRIPT_SUBMIT_DISABLED',
+    60015: 'CREATE_CONTEST_TASK_TASK_TYPE_EMPTY',
+    60016: 'CREATE_CONTEST_TASK_FRAMEWORK_EMPTY',
+    60017: 'CREATE_CONTEST_TASK_DRIVER_IMAGE_NOT_FOUND',
+    60018: 'CREATE_CONTEST_TASK_CONTRIBUTOR_NOT_FOUND',
+    60019: 'CREATE_CONTEST_TASK_SUBMISSION_GPU_TYPE_MISMATCH',
+    60020: 'CREATE_CONTEST_TASK_MODEL_NOT_DOWNLOADED',
+    60021: 'CREATE_CONTEST_TASK_FRAMEWORK_UNRESOLVED',
+    60022: 'CREATE_CONTEST_TASK_GPU_CONFIG_UNSUPPORTED',
+    60023: 'CREATE_CONTEST_TASK_YAML_CONFIG_EMPTY',
+    60024: 'CREATE_CONTEST_TASK_YAML_CONFIG_INVALID',
+    60025: 'CREATE_CONTEST_TASK_GPU_TYPE_UNSUPPORTED',
+    60026: 'CREATE_CONTEST_TASK_MODEL_NOT_FOUND',
+    60027: 'CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED',
+    60028: 'CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING',
+    60029: 'CREATE_CONTEST_TASK_TRANSFORMERS_REQUIRES_FAILED_BASELINE',
+    60030: 'CREATE_CONTEST_TASK_API_VERSION_UNSUPPORTED',
+}
+
+
+def load_marks(path=DATA):
+    try:
+        data = json.loads(Path(path).read_text())
+        if not isinstance(data, dict) or type(data.get('version')) is not int or data['version'] != 1 or not isinstance(data.get('marks'), list):
+            raise ValueError
+        marks = {}
+        for row in data['marks']:
+            if (not isinstance(row, dict) or not isinstance(row.get('model'), str)
+                    or not row['model'].strip() or row.get('gpu') not in set(GPU_ENUMS.values())
+                    or type(row.get('code')) is not int):
+                raise ValueError
+            marks[model_key(row['model']), row['gpu']] = row['code']
+        return marks
+    except (OSError, ValueError, TypeError, KeyError) as error:
+        raise ValueError('Invalid reject marks') from error

src/pending_pool.py

@@ -46,12 +46,12 @@ def classify_foothold(text):
     """Classify literal pending evidence; mixed markers prefer verified."""
     for label, markers in (
         ('verified_reuse', ('已验证成功', '验证通过')),
-        ('in_flight', ('正在验证', '验证排队中', '验证进行中')),
+        ('in_flight', ('正在验证', '验证进行中')),
         ('failed_foothold', ('验证失败',)),
     ):
         if any(marker in text for marker in markers):
             return label
-    return 'unknown'
+    return 'queued_only' if '验证排队中' in text else 'unknown'
 
 
 def has_no_download_marker(text):

src/precheck.py

@@ -0,0 +1,55 @@
+"""Read-only platform pair status; return only fixed statuses and GPU codes."""
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+from .config_params import GPU_ENUMS
+from .task_add_client import resolve_token
+
+SAFE_STATUSES = {'ok', 'transport_error', 'http_error', 'malformed_json',
+                 'platform_error', 'malformed_data', 'missing/invalid_token'}
+
+
+def precheck_model(model, token=None, opener=urllib.request.urlopen):
+    try:
+        token = resolve_token() if token is None else token
+        if not isinstance(token, str) or not token.strip() or '<' in token or '>' in token:
+            return 'missing/invalid_token', {}
+        request = urllib.request.Request(
+            os.environ.get('MODELHUB_API_BASE', 'https://modelhub.org.cn').rstrip('/')
+            + '/api/computility/models/search-by-model-id?modelId='
+            + urllib.parse.quote(model, safe=''),
+            headers={'Xc-Token': token, 'Accept': 'application/json'}, method='GET')
+    except Exception:
+        return 'missing/invalid_token', {}
+    try:
+        with opener(request, timeout=30) as response:
+            if response.status != 200:
+                return 'http_error', {}
+            raw = response.read()
+    except urllib.error.HTTPError:
+        return 'http_error', {}
+    except Exception:
+        return 'transport_error', {}
+    try:
+        body = json.loads(raw)
+    except (ValueError, TypeError, UnicodeError):
+        return 'malformed_json', {}
+    if not isinstance(body, dict) or 'code' not in body:
+        return 'malformed_data', {}
+    if type(body['code']) not in (str, int) or str(body['code']) != '0':
+        return 'platform_error', {}
+    data = body.get('data')
+    if not isinstance(data, dict) or not isinstance(data.get('verifyResult'), dict):
+        return 'malformed_data', {}
+    result = {}
+    for gpu, row in data['verifyResult'].items():
+        if gpu not in GPU_ENUMS.values():
+            continue
+        if not isinstance(row, dict) or not isinstance(row.get('result'), str):
+            return 'malformed_data', {}
+        code = {'已验证': 60027, '验证中': 60028}.get(row['result'])
+        if code is not None:
+            result[gpu] = code
+    return 'ok', result

src/server.py

@@ -59,15 +59,14 @@ def _on_sigterm(signum: int, frame) -> None:  # noqa: ARG001
 
 
 def _live_worker():
-    from pathlib import Path
     from .ledger import rebuild
     try:
-        _, _, _, sizes = rebuild(Path(os.environ.get('LIVE_STATE_FILE', '/tmp/live-submitted.json')))
-        sys.stderr.write('[live] ledger rebuilt remote=%d baked=%d persisted=%d\n' % sizes)
+        _, _, _, sizes = rebuild()
+        sys.stderr.write('[live] ledger rebuilt remote=%d baked=%d marks=%d\n' % sizes)
     except Exception:
         sys.stderr.write('[live] hold reason=ledger_rebuild_failed\n')
     try:
-        interval = max(60, int(os.environ.get('LIVE_CYCLE_SECONDS', '600')))
+        interval = max(60, int(os.environ.get('LIVE_CYCLE_SECONDS', '3600')))
     except ValueError:
         interval = 600
     while not _shutdown.is_set():

src/strategy.py

@@ -12,10 +12,10 @@ from typing import Optional
 
 # Community alias order (level C); we only check presence, never echo values.
 _TOKEN_ENV_KEYS = (
-    "EXTERNAL_SERVICE_TOKEN",
     "XC_TOKEN_FILE",
     "XC_TOKEN",
     "MODELHUB_XC_TOKEN",
+    "EXTERNAL_SERVICE_TOKEN",
 )
 
 

src/task_add_client.py

@@ -19,17 +19,40 @@ def build_payload(model_address, target_gpu='<TARGET_GPU>', framework='<FRAMEWOR
                 configParams=config_params)
 
 
+TOKEN_SOURCES = ('XC_TOKEN_FILE', 'XC_TOKEN', 'MODELHUB_XC_TOKEN', 'EXTERNAL_SERVICE_TOKEN')
+
+
+def _token_source(name):
+    """Read one source; status and credential stay internal to resolution."""
+    value = os.environ.get(name)
+    if value is None:
+        return 'absent', ''
+    raw = value.strip()
+    if not raw:
+        return 'empty', ''
+    if name == 'XC_TOKEN_FILE':
+        try:
+            raw = Path(raw).read_text().strip()
+        except (OSError, UnicodeError, ValueError):
+            return 'rejected', ''
+    if not raw:
+        return 'empty', ''
+    if '<' in raw or '>' in raw or re.search(
+            r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp|xxx)(?:$|[^a-z0-9])', raw):
+        return 'rejected', ''
+    return 'ok', raw
+
+
+def token_probe_status():
+    """Return fixed per-source status labels only, never credentials or paths."""
+    return {name: _token_source(name)[0] for name in TOKEN_SOURCES}
+
+
 def resolve_token_with_source():
-    for name in ('EXTERNAL_SERVICE_TOKEN', 'XC_TOKEN_FILE', 'XC_TOKEN', 'MODELHUB_XC_TOKEN'):
-        raw = os.environ.get(name, '').strip()
-        if name == 'XC_TOKEN_FILE' and raw:
-            try:
-                raw = Path(raw).read_text().strip()
-            except (OSError, UnicodeError):
-                continue
-        if raw and '<' not in raw and '>' not in raw and not re.search(
-                r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp|xxx)(?:$|[^a-z0-9])', raw):
-            return raw, name
+    for name in TOKEN_SOURCES:
+        status, token = _token_source(name)
+        if status == 'ok':
+            return token, name
     raise ValueError('Missing or invalid service credential')
 
 

tests/test_all_tasks_foothold.py

@@ -0,0 +1,125 @@
+import io
+import json
+import os
+import unittest
+from unittest.mock import Mock, patch
+from urllib.parse import parse_qs, urlparse
+from src.all_tasks_foothold import (load_all_tasks_foothold, classify_api_record,
+    build_candidates, merge_candidates, candidate_priority, TARGET_GPUS, FootholdSnapshot)
+from src import live_queue
+
+
+def record(status='success', result=1, model='org/model', gpu='壁砺', **extra):
+    return dict(modelId=model, gpuType=gpu, status=status, verifyResult=result, **extra)
+
+
+def response(records=(), total=None, **body):
+    result = Mock(status=200)
+    result.read.return_value = json.dumps(body or dict(code=0, data=dict(records=list(records), total=len(records) if total is None else total))).encode()
+    result.__enter__ = Mock(return_value=result)
+    result.__exit__ = Mock(return_value=False)
+    return result
+
+
+class TestFoothold(unittest.TestCase):
+    def setUp(self):
+        self.env = patch.dict(os.environ, {}, clear=True)
+        self.env.start()
+        self.addCleanup(self.env.stop)
+        self.network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
+        self.network.start()
+        self.addCleanup(self.network.stop)
+
+    def load(self, responses, **kw):
+        opener = Mock(side_effect=responses)
+        snapshot = load_all_tasks_foothold(opener=opener, token_resolver=lambda: 'offline-credential', **kw)
+        return snapshot, opener
+
+    def test_whitelist_rejects_waiting_conflicts_and_bad_types(self):
+        for status, result, bucket in [('waiting', 1, 'pass'), ('failed', -1, 'fail'),
+                ('running', 1, 'running'), ('success', True, 'pass'),
+                ('success', {}, 'pass'), ('other', 1, 'pass'), ('success', 1.0, 'pass')]:
+            self.assertIsNone(classify_api_record(record(status, result), bucket))
+        for result in (1, '1'):
+            self.assertIsNotNone(classify_api_record(record(result=result), 'pass'))
+        for result in (None, 0, '0'):
+            self.assertIsNotNone(classify_api_record(record(' RUNNING ', result), 'running'))
+        for fields in (dict(modelId=True), dict(gpuType=''), dict(modelId='')):
+            self.assertIsNone(classify_api_record(dict(record(), **fields), 'pass'))
+
+    def test_scope_buckets_aliases_order_and_queue_isolation(self):
+        before = live_queue.last_stats()
+        snap, opener = self.load([response([record()]), response([record(result=-1, model='org/fail')]),
+                                 response([record('running', None, model='org/run')])])
+        self.assertEqual(snap.reason, 'ok')
+        self.assertEqual([r['foothold_class'] for r in snap.candidates], ['verified_reuse', 'failed_foothold', 'in_flight'])
+        self.assertEqual(snap.candidates[0]['try_next_gpus'], list(TARGET_GPUS[1:]))
+        for call in opener.call_args_list:
+            query = parse_qs(urlparse(call.args[0].full_url).query)
+            self.assertEqual(query['onlyMine'], ['false'])
+            self.assertNotIn('waiting', query.get('status', []))
+            self.assertLessEqual(call.kwargs['timeout'], 5)
+        self.assertEqual(before, live_queue.last_stats())
+        self.assertEqual(len(opener.call_args_list), 3)
+
+    def test_any_bucket_error_discards_whole_snapshot(self):
+        for bad in (response([record('waiting', 1)]), response(data=None, code=0),
+                    response(code=403), OSError('private-response')):
+            snap, _ = self.load([response([record()]), bad])
+            self.assertNotEqual(snap.reason, 'ok')
+            self.assertEqual(snap.candidates, [])
+        snap, _ = self.load([response(), response(), response()])
+        self.assertEqual(snap.reason, 'ok')
+        self.assertEqual(snap.candidates, [])
+
+    def test_hard_caps_total_drift_and_invalid_env(self):
+        with patch.dict(os.environ, {'ALL_TASKS_MAX_PAGES_PASS': '999', 'ALL_TASKS_MAX_PAGES_FAIL': 'bad',
+                                     'ALL_TASKS_MAX_PAGES_RUNNING': '-1', 'ALL_TASKS_PAGE_SIZE': '999'}):
+            pages = [response([record(model='org/p' + str(n))], total=1000+n) for n in range(3)]
+            pages += [response([record(result=-1, model='org/f' + str(n))], total=1000+n) for n in range(3)]
+            pages += [response([record('running', None, model='org/r' + str(n))], total=1000+n) for n in range(2)]
+            snap, opener = self.load(pages)
+        self.assertEqual(opener.call_count, 8)
+        self.assertTrue(all(b['truncated'] for b in snap.buckets.values()))
+        self.assertTrue(all('pageSize=100' in c.args[0].full_url for c in opener.call_args_list))
+        snap, opener = self.load([], clock=Mock(side_effect=[0, 21]))
+        self.assertEqual(snap.reason, 'deadline')
+        opener.assert_not_called()
+
+    def test_repeated_tasks_pages_and_conflicts(self):
+        r = record(taskId=1)
+        snap, opener = self.load([response([r], total=999), response([r], total=1000), response(), response()])
+        self.assertEqual(snap.buckets['pass']['unique'], 1)
+        self.assertTrue(snap.buckets['pass']['truncated'])
+        self.assertEqual(opener.call_count, 4)
+        snap, _ = self.load([response([r]), response([record(result=-1, taskId=1)])])
+        self.assertEqual(snap.reason, 'conflicting_task')
+        self.assertEqual(snap.candidates, [])
+
+    def test_missing_token_never_opens(self):
+        opener = Mock()
+        snap = load_all_tasks_foothold(opener=opener, token_resolver=Mock(side_effect=ValueError()))
+        self.assertEqual(snap.reason, 'missing/invalid_token')
+        opener.assert_not_called()
+
+    def test_merge_constraints_empty_targets_and_uniform_priority(self):
+        base = [dict(model_address='https://huggingface.co/ORG/Model/', framework='custom',
+                     skip_reason='manual', section='B', foothold_class='verified_reuse',
+                     try_next_gpus=['壁砺'], targetGpu='天垓150'),
+                dict(model_address='org/static', section='A', foothold_class='verified_reuse', try_next_gpus=['天垓150'])]
+        dynamic = build_candidates([('org/model', 'Biren_166m', 'in_flight'), ('org/run', 'MetaX_c-500', 'in_flight')])
+        merged = merge_candidates(base, dynamic)
+        row = next(r for r in merged if 'huggingface' in r['model_address'])
+        self.assertEqual(row['framework'], 'custom')
+        self.assertEqual(row['skip_reason'], 'manual')
+        self.assertEqual(row['try_next_gpus'], [])
+        self.assertEqual(row['foothold_class'], 'verified_reuse')
+        self.assertEqual(len(merged), 3)
+        self.assertEqual(merged[-1]['model_address'], 'org/run')
+        self.assertEqual(merged[-1]['try_next_gpus'], list(TARGET_GPUS))
+        self.assertEqual(candidate_priority(dict(foothold_class='in_flight'))[0], 3)
+
+    def test_pending_queue_only_text(self):
+        from src.pending_pool import classify_foothold
+        self.assertEqual(classify_foothold('验证排队中'), 'queued_only')
+        self.assertEqual(classify_foothold('验证排队中; 验证通过'), 'verified_reuse')

tests/test_banned.py

@@ -0,0 +1,36 @@
+import unittest
+
+from src.banned import banned_rule
+
+
+class TestBanned(unittest.TestCase):
+    def test_default_engine_and_openchat(self):
+        for model, rule in [('org/Model-GGUF', 'gguf'), ('org/model.gguf', 'gguf'),
+                            ('org/GGML-model', 'gguf'), ('OpenChat/model', 'openchat')]:
+            with self.subTest(model=model):
+                self.assertEqual(banned_rule(model), rule)
+
+    def test_unmerged_adapter_hints(self):
+        for model in ('org/model-lora', 'org/LORA', 'org/adapter_model',
+                      'org/model-adapter-config', 'org/Model-PEFT'):
+            with self.subTest(model=model):
+                self.assertEqual(banned_rule(model), 'peft-lora')
+
+    def test_full_weights_and_unrelated_names(self):
+        for model in ('org/lora-merged', 'org/merge-lora', 'org/PEFT-FULLFT',
+                      'org/full-ft-adapter_model', 'org/adapter-config-MERGED',
+                      'org/flora', 'org/lorax', 'org/model_lora',
+                      'org/model-GRPO', 'abaryan/CyberXP_Agent_Llama_3.2_1B'):
+            with self.subTest(model=model):
+                self.assertIsNone(banned_rule(model))
+
+    def test_full_weight_hint_does_not_override_other_bans(self):
+        self.assertEqual(banned_rule('org/lora-merged-GGUF'), 'gguf')
+        self.assertEqual(banned_rule('org/openchat-fullft'), 'openchat')
+
+    def test_existing_rules(self):
+        for model in ('gpt2', 'org/Phi-3-mini', 'nvidia/Nemotron-H-4B',
+                      'allenai/OLMo-2-7B', 'org/LFM2', 'org/Spark-X2.5',
+                      'org/Breeze', 'org/gemma2', 'org/mistral-v0.1', 'org/tinyllama'):
+            with self.subTest(model=model):
+                self.assertIsNotNone(banned_rule(model))

tests/test_health.py

@@ -58,3 +58,18 @@ class TestHealth(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestStatsStatus(unittest.TestCase):
+    def test_status_stats_and_metric(self):
+        from src.live_queue import compute_stats
+        stats = dict(compute_stats([]), updated_at=123.0)
+        for snapshot in (None, stats):
+            with mock.patch('src.health.last_stats', return_value=snapshot):
+                code, _, body = handle_get('/status')
+            self.assertEqual(code, 200)
+            payload = json.loads(body)
+            self.assertEqual(payload['stats'], snapshot)
+            self.assertEqual(payload['success_metric'], 'verify_pass')
+            self.assertIn('strategy_id_present', payload)
+            self.assertIn('external_service_token_present', payload)

tests/test_live_cycle.py

@@ -1,9 +1,8 @@
 import io
 import os
-import tempfile
 import unittest
 from unittest.mock import Mock, patch
-from src.live_cycle import run_cycle, _submitted
+from src.live_cycle import run_cycle, _submitted, _marked
 
 
 def candidate(name, section='B', foothold='unknown'):
@@ -12,12 +11,11 @@ def candidate(name, section='B', foothold='unknown'):
 
 class TestLiveCycle(unittest.TestCase):
     def setUp(self):
-        directory = tempfile.TemporaryDirectory()
-        self.addCleanup(directory.cleanup)
-        self.env = patch.dict(os.environ, {'STRATEGY_ID': 'sid', 'LIVE_STATE_FILE': directory.name + '/state.json', 'CONFIG_PARAMS_SOURCE': 'sample', 'XC_TOKEN': 'synthetic-secret'}, clear=True)
+        self.env = patch.dict(os.environ, {'LIVE_PRECHECK': '0', 'STRATEGY_ID': 'sid', 'CONFIG_PARAMS_SOURCE': 'sample', 'XC_TOKEN': 'synthetic-secret'}, clear=True)
         self.env.start()
         self.addCleanup(self.env.stop)
         _submitted.clear()
+        _marked.clear()
         self.client = Mock()
         self.client.submit.return_value = dict(http_status=200, platform_code=0)
         network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
@@ -31,6 +29,79 @@ class TestLiveCycle(unittest.TestCase):
         reader = Mock(side_effect=[(n, 'platform_error' if n is None else 'ok', set()) for n in counts])
         return run_cycle(rows or [candidate('a')], self.client, reader)
 
+    def submitted_pairs(self):
+        return sorted(m + '|' + g for m, g in _submitted)
+
+    def test_done_rejection_marks_pair_and_continues(self):
+        for code in (60027, 60028):
+            with self.subTest(code=code):
+                self.client.submit.side_effect = [
+                    dict(http_status=200, platform_code=code),
+                    dict(http_status=200, platform_code=0)]
+                with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
+                    decisions = self.cycle([candidate(str(code)), candidate('ok' + str(code))], (1, 1))
+                self.assertEqual([d['action'] for d in decisions], ['rejected', 'submitted'])
+                self.assertEqual(_marked[('org/' + str(code), 'Biren_166m')], code)
+        self.assertIn('rejected', self.logs.getvalue())
+        self.assertIn('CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING', self.logs.getvalue())
+
+    def test_rejection_marked_and_not_retried_in_process(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=60020)
+        self.cycle()
+        self.assertEqual(self.submitted_pairs(), [])
+        self.cycle(counts=())
+        self.client.submit.assert_called_once()
+
+    def test_unknown_rejection(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=99999)
+        decisions = self.cycle()
+        self.assertEqual(decisions[0]['action'], 'rejected')
+        self.assertEqual(decisions[0]['reason'], 'unknown')
+        self.assertEqual(self.submitted_pairs(), [])
+
+    def test_attempt_cap_and_invalid_fallback(self):
+        for value, expected in ((None, 30), ('bad', 30), ('0', 30), ('-1', 30), ('2', 2)):
+            with self.subTest(value=value):
+                _marked.clear()
+                self.client.reset_mock()
+                self.client.submit.return_value = dict(http_status=200, platform_code=60020)
+                with patch.dict(os.environ, {} if value is None else {'LIVE_MAX_ATTEMPTS_PER_CYCLE': value}):
+                    decisions = self.cycle([candidate(str(i)) for i in range(expected)], (1,) * expected)
+                self.assertEqual(self.client.submit.call_count, expected)
+                self.assertEqual(decisions[-1]['reason'], 'attempt_cap')
+
+    def test_submit_cap_counts_successes_after_rejections(self):
+        self.client.submit.side_effect = [dict(http_status=200, platform_code=c) for c in (60020, 0, 60028, 0, 0, 0, 0, 0)]
+        decisions = self.cycle([candidate(str(i)) for i in range(10)], (1,) * 8)
+        self.assertEqual(self.client.submit.call_count, 8)
+        self.assertEqual(sum(d['action'] == 'submitted' for d in decisions), 6)
+        self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
+
+    def test_task_limit_stops_cycle(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=60007)
+        decisions = self.cycle([candidate('a'), candidate('b')])
+        self.assertEqual([d['action'] for d in decisions], ['rejected', 'hold'])
+        self.assertEqual(decisions[-1]['reason'], 'task_num_limit')
+        self.assertEqual(self.submitted_pairs(), [])
+        self.client.submit.assert_called_once()
+
+    def test_unknown_outcomes_count_and_reserve_before_submit(self):
+        for result in (dict(http_status=200, platform_code='0'), dict(http_status=None, platform_code=None), OSError()):
+            with self.subTest(result=result):
+                _submitted.clear()
+                self.client.reset_mock()
+                def submit(payload):
+                    self.assertIn('org/a|Biren_166m', self.submitted_pairs())
+                    if isinstance(result, Exception):
+                        raise result
+                    return result
+                self.client.submit.side_effect = submit
+                with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
+                    decisions = self.cycle([candidate('a'), candidate('b')])
+                self.assertEqual(decisions[0]['action'], 'submitted')
+                self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
+                self.client.submit.assert_called_once()
+
     def test_high_and_failed_queue(self):
         for n in (95, None):
             decisions = self.cycle(counts=(n,))
@@ -39,9 +110,10 @@ class TestLiveCycle(unittest.TestCase):
         self.assertIn('[live] hold', self.logs.getvalue())
 
     def test_order_and_cap(self):
-        rows = [candidate('b'), candidate('a', 'A'), candidate('verified', 'A', 'verified_reuse'), candidate('c'), candidate('d')]
-        self.cycle(rows, (90, 90, 90))
-        self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/verified', 'org/a', 'org/b'])
+        rows = [candidate('b'), candidate('a', 'A'), candidate('verified', 'A', 'verified_reuse'), candidate('c'), candidate('d'), candidate('e'), candidate('f'), candidate('g')]
+        self.cycle(rows, (90,) * 6)
+        self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list],
+                         ['org/verified', 'org/a', 'org/b', 'org/c', 'org/d', 'org/e'])
         self.assertIn('cycle_cap', self.logs.getvalue())
 
     def test_midcycle(self):
@@ -57,10 +129,9 @@ class TestLiveCycle(unittest.TestCase):
             self.cycle(counts=(99,))
         self.client.submit.assert_called_once()
 
-    def test_banned_duplicate_persistence(self):
+    def test_banned_and_duplicate_memory(self):
         rows = [candidate('GPT-2'), candidate('ok'), candidate('ok')]
         self.cycle(rows)
-        _submitted.clear()
         self.cycle([candidate('ok')], ())
         self.client.submit.assert_called_once()
         self.assertIn('banned:', self.logs.getvalue())
@@ -76,10 +147,19 @@ class TestLiveCycle(unittest.TestCase):
         self.assertNotIn('do not log', self.logs.getvalue())
 
     def test_missing_token_real_reader(self):
-        with patch.dict(os.environ, {'XC_TOKEN': ''}):
+        with patch.dict(os.environ, {'STRATEGY_ID': 'sid'}, clear=True):
             run_cycle([candidate('a')], self.client)
         self.client.submit.assert_not_called()
         self.assertIn('missing/invalid_token', self.logs.getvalue())
+        self.assertIn('[live] token_probe EXTERNAL_SERVICE_TOKEN=absent XC_TOKEN_FILE=absent XC_TOKEN=absent MODELHUB_XC_TOKEN=absent', self.logs.getvalue())
+        self.assertNotIn('[live] token source=', self.logs.getvalue())
+
+    def test_marked_token_probe_does_not_leak(self):
+        with patch.dict(os.environ, {'XC_TOKEN': '<private-secret-fragment>'}):
+            self.cycle()
+        self.client.submit.assert_not_called()
+        self.assertIn('XC_TOKEN=rejected', self.logs.getvalue())
+        self.assertNotIn('private-secret', self.logs.getvalue())
 
     def test_worker_waits_and_stops(self):
         from src.server import _live_worker
@@ -98,7 +178,7 @@ class TestLiveCycle(unittest.TestCase):
         response.__exit__ = Mock(return_value=False)
         response.status = 200
         response.read.return_value = json.dumps({'code': 40100}).encode()
-        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}), patch('urllib.request.urlopen', return_value=response):
+        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret', 'LIVE_PRECHECK': '0'}), patch('urllib.request.urlopen', return_value=response):
             run_cycle([candidate('a')], self.client)
         self.client.submit.assert_not_called()
 
@@ -108,3 +188,91 @@ class TestLiveCycle(unittest.TestCase):
         self.client.submit.assert_not_called()
         self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
         self.assertIn('placeholder', self.logs.getvalue())
+
+    def test_stats_logged_once_and_submission_continues(self):
+        from src.live_queue import compute_stats
+        stats = compute_stats([dict(status='success', verifyResult=1, stageAndNoviceTaskCredit='5.5')])
+        stats_reader = Mock(return_value=(0, 'ok', set(), stats))
+        run_cycle([candidate('stats')], self.client, lambda: (0, 'ok', set()), stats_reader=stats_reader)
+        stats_reader.assert_called_once_with()
+        lines = [s for s in self.logs.getvalue().splitlines() if s.startswith('[live] stats ')]
+        self.assertEqual(lines, ['[live] stats total=1 queue=0 verify_pass=1 verify_fail=0 failed=0 unknown=0 points=5.5'])
+        self.assertNotIn('synthetic-secret', self.logs.getvalue())
+        self.client.submit.assert_called_once()
+
+    def test_stats_failure_is_safe_and_does_not_gate(self):
+        for index, reader in enumerate((Mock(return_value=(None, 'platform_error', set(), None)),
+                       Mock(side_effect=OSError('synthetic-secret')),
+                       Mock(return_value=(None, 'synthetic-secret', set(), None)))):
+            self.logs.truncate(0)
+            self.logs.seek(0)
+            run_cycle([candidate('stats-failure-' + str(index))], self.client, lambda: (0, 'ok', set()), stats_reader=reader)
+            lines = [s for s in self.logs.getvalue().splitlines() if s.startswith('[live] stats ')]
+            self.assertEqual(len(lines), 1)
+            self.assertTrue(lines[0].startswith('[live] stats unavailable reason='))
+            self.assertNotIn('synthetic-secret', self.logs.getvalue())
+        self.assertEqual(self.client.submit.call_count, 3)
+
+    def test_injected_queue_skips_stats_network(self):
+        self.cycle()
+        self.assertEqual(self.logs.getvalue().count('[live] stats unavailable reason=not_read'), 1)
+
+    def test_foothold_explicit_disabled_and_injected_queue_stay_offline(self):
+        from src.all_tasks_foothold import FootholdSnapshot
+        dynamic = Mock(return_value=FootholdSnapshot())
+        with patch.dict(os.environ, {'ALL_TASKS_FOOTHOLD': '1'}):
+            run_cycle([candidate('explicit')], self.client, lambda: (0, 'ok', set()), foothold_reader=dynamic)
+        dynamic.assert_not_called()
+        with patch('src.live_cycle.build_ordered_candidates', return_value=[candidate('base')]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline'}):
+            run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=dynamic)
+            with patch.dict(os.environ, {'ALL_TASKS_FOOTHOLD': '1'}):
+                run_cycle(None, self.client, lambda: (0, 'ok', set()))
+        dynamic.assert_not_called()
+
+    def test_foothold_merge_order_constraints_pending_and_fallback(self):
+        from src.all_tasks_foothold import FootholdSnapshot, build_candidates
+        base = [candidate('base', 'A', 'verified_reuse')]
+        snapshot = FootholdSnapshot(build_candidates([('org/dynamic', 'Biren_166m', 'in_flight')]))
+        with patch('src.live_cycle.build_ordered_candidates', return_value=base) as pending, patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'PENDING_SECTION': 'A', 'ALL_TASKS_FOOTHOLD': '1'}):
+            run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
+            pending.assert_called_once_with()
+            self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/base', 'org/dynamic'])
+            _submitted.clear()
+            self.client.reset_mock()
+            run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=Mock(side_effect=OSError('private-response')))
+            self.assertEqual(self.client.submit.call_args.args[0]['modelAddress'], 'org/base')
+            self.assertIn('[foothold] reason=reader_error', self.logs.getvalue())
+            self.assertNotIn('private-response', self.logs.getvalue())
+
+    def test_foothold_empty_intersection_never_revives_target(self):
+        from src.all_tasks_foothold import FootholdSnapshot, build_candidates
+        base = [dict(candidate('empty'), targetGpu='曦望')]
+        snapshot = FootholdSnapshot(build_candidates([('org/empty', 'Biren_166m', 'verified_reuse')]))
+        with patch('src.live_cycle.build_ordered_candidates', return_value=base), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'ALL_TASKS_FOOTHOLD': '1'}):
+            decisions = run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
+        self.client.submit.assert_not_called()
+        self.assertEqual(decisions[-1]['reason'], 'all_gpus_done_or_unmapped')
+
+    def test_dynamic_still_respects_queue_marks_ledger_precheck(self):
+        from src.all_tasks_foothold import FootholdSnapshot
+        snapshot = FootholdSnapshot([dict(candidate('guarded'), evidence_source='all-tasks', foothold_class='verified_reuse')])
+        with patch('src.live_cycle.build_ordered_candidates', return_value=[]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'ALL_TASKS_FOOTHOLD': '1'}):
+            for count in (95, None):
+                decisions = run_cycle(None, self.client, lambda: (count, 'ok', set()), foothold_reader=lambda: snapshot)
+                self.assertEqual(decisions[-1]['action'], 'hold')
+            with patch('src.live_cycle.load_baked', return_value=({('org/guarded', 'Biren_166m')}, set())):
+                run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
+            _marked[('org/guarded', 'Biren_166m')] = 60028
+            run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
+            _marked.clear()
+            with patch.dict(os.environ, {'LIVE_PRECHECK': '1'}):
+                run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot,
+                          precheck=lambda _: ('ok', {'Biren_166m': 60027}))
+        self.client.submit.assert_not_called()
+
+    def test_pending_section_b_does_not_gain_dynamic_section_a(self):
+        from src.all_tasks_foothold import FootholdSnapshot
+        dynamic = FootholdSnapshot([candidate('dynamic', 'A', 'verified_reuse')])
+        with patch('src.live_cycle.build_ordered_candidates', return_value=[candidate('base')]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'PENDING_SECTION': 'B', 'ALL_TASKS_FOOTHOLD': '1'}):
+            run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: dynamic)
+        self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/base'])

tests/test_live_ledger.py

@@ -1,10 +1,9 @@
 import json
 import os
-import tempfile
 import unittest
 from pathlib import Path
 from unittest.mock import Mock, patch
-from src.live_cycle import run_cycle, _submitted
+from src.live_cycle import run_cycle, _submitted, _marked
 from src.ledger import load_baked, rebuild
 from src.banned import banned_rule
 from src.task_add_client import resolve_token_with_source
@@ -12,13 +11,11 @@ from src.task_add_client import resolve_token_with_source
 
 class TestLedger(unittest.TestCase):
     def setUp(self):
-        temp = tempfile.TemporaryDirectory()
-        self.addCleanup(temp.cleanup)
-        self.state = Path(temp.name) / 'state.json'
-        env = patch.dict(os.environ, dict(STRATEGY_ID='sid', XC_TOKEN='synthetic-credential', CONFIG_PARAMS_SOURCE='sample', LIVE_STATE_FILE=str(self.state)), clear=True)
+        env = patch.dict(os.environ, dict(LIVE_PRECHECK='0', STRATEGY_ID='sid', XC_TOKEN='synthetic-credential', CONFIG_PARAMS_SOURCE='sample'), clear=True)
         env.start()
         self.addCleanup(env.stop)
         _submitted.clear()
+        _marked.clear()
         self.addCleanup(_submitted.clear)
         net = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
         net.start()
@@ -29,11 +26,10 @@ class TestLedger(unittest.TestCase):
     def cycle(self, remote=(), model='unit/model', gpus=None):
         return run_cycle([dict(model_address=model, try_next_gpus=gpus or ['壁砺', '天垓100'])], self.client, lambda: (0, 'ok', set(remote)))
 
-    def test_pair_and_restart(self):
+    def test_pair_and_memory(self):
         self.cycle([('unit/model', 'Biren_166m')])
         self.assertEqual(self.client.submit.call_args.args[0]['targetGpu'], 'Iluvatar_bi-100')
-        self.assertEqual(json.loads(self.state.read_text()), ['unit/model|Iluvatar_bi-100'])
-        _submitted.clear()
+        self.assertIn(('unit/model', 'Iluvatar_bi-100'), _submitted)
         self.cycle([('unit/model', 'Biren_166m')])
         self.client.submit.assert_called_once()
 
@@ -54,10 +50,11 @@ class TestLedger(unittest.TestCase):
     def test_source_and_no_value_logs(self):
         import io
         with patch.dict(os.environ, {'EXTERNAL_SERVICE_TOKEN': 'platform-credential', 'XC_TOKEN_FILE': '/nonexistent'}), patch('sys.stderr', new_callable=io.StringIO) as log:
-            self.assertEqual(resolve_token_with_source(), ('platform-credential', 'EXTERNAL_SERVICE_TOKEN'))
+            self.assertEqual(resolve_token_with_source(), ('synthetic-credential', 'XC_TOKEN'))
             self.cycle()
-        self.assertEqual(log.getvalue().count('token source=EXTERNAL_SERVICE_TOKEN'), 1)
+        self.assertEqual(log.getvalue().count('token source=XC_TOKEN'), 1)
         self.assertNotIn('platform-credential', log.getvalue())
+        self.assertNotIn('synthetic-credential', log.getvalue())
         with patch.dict(os.environ, {'EXTERNAL_SERVICE_TOKEN': '<TOKEN>', 'XC_TOKEN': 'placeholder', 'MODELHUB_XC_TOKEN': 'alias-credential'}):
             self.assertEqual(resolve_token_with_source()[1], 'MODELHUB_XC_TOKEN')
 
@@ -66,6 +63,6 @@ class TestLedger(unittest.TestCase):
         self.assertGreater(len(rows), 0)
         self.assertEqual([r['section'] for r in rows], sorted(r['section'] for r in rows))
         self.assertFalse(any(banned_rule(r['model_address']) for r in rows))
-        count, pairs, blocked, sizes = rebuild(self.state, lambda: (0, 'ok', {('unit/model', 'Biren_166m')}))
+        count, pairs, blocked, sizes = rebuild(lambda: (0, 'ok', {('unit/model', 'Biren_166m')}))
         self.assertIn(('unit/model', 'Biren_166m'), pairs)
         self.assertEqual(sizes[0], 1)

tests/test_live_queue.py

@@ -53,3 +53,52 @@ class TestLiveQueue(unittest.TestCase):
                 req = call.call_args.args[0]
                 self.assertEqual(req.get_header('Xc-token'), 'synthetic-secret')
                 self.assertEqual(req.method, 'GET')
+
+
+class TestVerifyStats(unittest.TestCase):
+    response = TestLiveQueue.response
+
+    def test_compute_stats_precedence_and_types(self):
+        from src.live_queue import compute_stats
+        rows = [
+            dict(status='waiting', verifyResult=1, stageAndNoviceTaskCredit=100),
+            dict(status='等待中', verifyResult=-1),
+            dict(status='success', verifyResult=-1, stageAndNoviceTaskCredit=100),
+            dict(status='failed', verifyResult=' -1 '),
+            dict(status='success'), dict(status='success', verifyResult=True),
+            dict(status='success', verifyResult=False),
+            dict(status='success', verifyResult='garbage'),
+            dict(status='error', verifyResult=None),
+            dict(status=' CANCELED ', verifyResult=0),
+        ]
+        for result, credit in [(1, '5'), ('1', 5.0), (1.0, None), (1, 'abc'), (1, True), (1, 'NaN')]:
+            rows.append(dict(status='success', verifyResult=result, stageAndNoviceTaskCredit=credit))
+        stats = compute_stats(rows)
+        self.assertEqual(stats, dict(total=16, queue=2, verify_pass=6, verify_fail=2,
+                                     failed=2, unknown=4, points=10.0))
+        self.assertEqual(sum(stats[k] for k in ('queue', 'verify_pass', 'verify_fail', 'failed', 'unknown')), stats['total'])
+        self.assertIs(type(stats['points']), float)
+        self.assertTrue(all(type(v) is int for k, v in stats.items() if k != 'points'))
+
+    def test_paginated_stats_and_snapshot(self):
+        from src.live_queue import read_queue_stats, last_stats
+        pages = [self.response({'code': 0, 'data': {'total': 2, 'records': [row]}})
+                 for row in [dict(status='success', verifyResult=1, stageAndNoviceTaskCredit='5', body='synthetic-secret'),
+                             dict(status='success', verifyResult=-1)]]
+        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', side_effect=pages):
+            count, reason, models, stats = read_queue_stats()
+        self.assertEqual((count, reason, models), (0, 'ok', set()))
+        self.assertEqual(stats['verify_pass'], 1)
+        self.assertEqual(stats['verify_fail'], 1)
+        self.assertEqual(stats['points'], 5.0)
+        snapshot = last_stats()
+        self.assertIn('updated_at', snapshot)
+        self.assertNotIn('synthetic-secret', json.dumps(snapshot))
+        snapshot['total'] = 999
+        stats['total'] = 888
+        self.assertEqual(last_stats()['total'], 2)
+        with patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(read_queue_stats(), (None, 'missing/invalid_token', set(), None))
+        self.assertEqual(last_stats()['total'], 2)
+        with patch('src.live_queue.read_queue_stats', return_value=(3, 'ok', set(), {})):
+            self.assertEqual(read_queue(), (3, 'ok', set()))

tests/test_pending_pool.py

@@ -59,7 +59,7 @@ class TestPendingPool(unittest.TestCase):
                               'org/fail', 'org/queue', 'org/progress', 'org/b'])
             self.assertEqual([r['foothold_class'] for r in rows],
                              ['verified_reuse', 'verified_reuse', 'unknown', 'in_flight',
-                              'failed_foothold', 'in_flight', 'in_flight', 'unknown'])
+                              'failed_foothold', 'queued_only', 'in_flight', 'unknown'])
             for row in rows:
                 self.assertEqual(row['evidence_source'], 'GAP' if row['foothold_class'] == 'unknown' else 'pending-foothold')
             self.assertEqual(rows[0]['foothold_text'], '他人:已验证成功; 正在验证')

tests/test_reject_marks.py

@@ -0,0 +1,276 @@
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+import urllib.error
+from unittest.mock import Mock, patch
+
+from src.config_params import GPU_ENUMS
+from src.live_cycle import run_cycle, _submitted, _marked
+from src.marks import load_marks, PERMANENT_PAIR_CODES
+from src.precheck import precheck_model
+
+
+class TestMarks(unittest.TestCase):
+    def test_baked_marks(self):
+        marks = load_marks()
+        self.assertEqual(len(marks), 16)
+        self.assertTrue(all(g in GPU_ENUMS.values() for m, g in marks))
+        self.assertEqual(marks['madeagents/hammer2.1-0.5b', 'Biren_166m'], 60027)
+        self.assertEqual(PERMANENT_PAIR_CODES, {60012, *range(60019, 60030)})
+
+    def test_verify_fail_marks(self):
+        rows = json.loads(Path('src/data/reject-marks.json').read_text())['marks']
+        keys = [(row['model'].lower(), row['gpu']) for row in rows]
+        self.assertEqual(len(keys), len(set(keys)))
+        failures = [row for row in rows if row['reason'].startswith('VERIFY_FAIL:')]
+        self.assertEqual(len(failures), 8)
+        self.assertEqual(sorted(row['reason'] for row in failures),
+                         sorted(['VERIFY_FAIL:SCORE_GAP'] * 5 +
+                                ['VERIFY_FAIL:PEFT_UNMERGED', 'VERIFY_FAIL:NO_IDLE_DEVICE',
+                                 'VERIFY_FAIL:REF_CACHE_MISS']))
+        marks = load_marks()
+        for row in failures:
+            self.assertEqual(row['code'], 60029)
+            self.assertIn(row['code'], PERMANENT_PAIR_CODES)
+            self.assertEqual(marks[row['model'].lower(), row['gpu']], 60029)
+        self.assertEqual(marks['abaryan/cyberxp_agent_llama_3.2_1b', 'Biren_166m'], 60029)
+        self.assertNotIn(('abaryan/cyberxp_agent_llama_3.2_1b', 'Iluvatar_bi-100'), marks)
+
+    def test_invalid_marks(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'marks.json'
+            with self.assertRaises(ValueError):
+                load_marks(path)
+            valid = dict(model='org/model', gpu='Biren_166m', code=60027)
+            for data in ('bad json', [], {}, {'version': 1, 'marks': {}},
+                         {'version': 1, 'marks': [None]},
+                         *({'version': 1, 'marks': [dict(valid, **update)]} for update in
+                           ({'model': ''}, {'model': 4}, {'gpu': 'unknown'}, {'gpu': []}, {'code': True}, {'code': '60027'}))):
+                path.write_text(data if isinstance(data, str) else json.dumps(data))
+                with self.subTest(data=data), self.assertRaises(ValueError):
+                    load_marks(path)
+
+
+class TestPrecheck(unittest.TestCase):
+    def check(self, body, raw=False):
+        response = Mock(status=200)
+        response.read.return_value = body if raw else json.dumps(body).encode()
+        response.__enter__ = Mock(return_value=response)
+        response.__exit__ = Mock(return_value=False)
+        opener = Mock(return_value=response)
+        result = precheck_model('org/model name', token='synthetic-secret', opener=opener)
+        request = opener.call_args.args[0]
+        self.assertEqual(request.full_url, 'https://modelhub.org.cn/api/computility/models/search-by-model-id?modelId=org%2Fmodel%20name')
+        self.assertEqual(request.get_header('Xc-token'), 'synthetic-secret')
+        self.assertEqual(request.get_header('Accept'), 'application/json')
+        self.assertEqual(request.method, 'GET')
+        self.assertEqual(opener.call_args.kwargs, {'timeout': 30})
+        self.assertNotIn('synthetic-secret', repr(result))
+        self.assertNotIn('authorName', repr(result))
+        return result
+
+    def test_mapping_and_redaction(self):
+        with patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(self.check({'code': '0', 'data': {'verifyResult': {
+                'Biren_166m': {'result': '已验证', 'records': [{'authorName': 'synthetic-secret'}]},
+                'Iluvatar_bi-150': {'result': '验证中'},
+                'Iluvatar_bi-100': {'result': '验证失败'},
+                'unknown': {'result': '已验证'},
+            }}}), ('ok', {'Biren_166m': 60027, 'Iluvatar_bi-150': 60028}))
+
+    def test_bad_responses(self):
+        with patch.dict(os.environ, {}, clear=True):
+            for body, expected, raw in ((b'{bad', 'malformed_json', True),
+                    ({'code': 3}, 'platform_error', False),
+                    ({'code': 0, 'data': []}, 'malformed_data', False),
+                    ([], 'malformed_data', False),
+                    ({'code': 0, 'data': {'verifyResult': []}}, 'malformed_data', False),
+                    ({'code': 0, 'data': {'verifyResult': {'Biren_166m': None}}}, 'malformed_data', False)):
+                with self.subTest(body=body):
+                    self.assertEqual(self.check(body, raw), (expected, {}))
+
+    def test_transport_http_and_token_errors(self):
+        for error, status in ((OSError('synthetic-secret'), 'transport_error'),
+                (urllib.error.HTTPError('url', 403, 'secret', {}, None), 'http_error')):
+            self.assertEqual(precheck_model('o/m', 'synthetic-secret', Mock(side_effect=error)), (status, {}))
+        with patch.dict(os.environ, {}, clear=True):
+            opener = Mock()
+            self.assertEqual(precheck_model('o/m', opener=opener), ('missing/invalid_token', {}))
+            opener.assert_not_called()
+        response = Mock(status=500)
+        response.__enter__ = Mock(return_value=response)
+        response.__exit__ = Mock(return_value=False)
+        self.assertEqual(precheck_model('o/m', 'synthetic-secret', Mock(return_value=response)), ('http_error', {}))
+
+
+class TestLiveMarks(unittest.TestCase):
+    def setUp(self):
+        _submitted.clear()
+        _marked.clear()
+        self.env = patch.dict(os.environ, dict(STRATEGY_ID='sid', XC_TOKEN='synthetic-secret', CONFIG_PARAMS_SOURCE='sample'), clear=True)
+        self.env.start()
+        self.addCleanup(self.env.stop)
+        self.addCleanup(_submitted.clear)
+        self.addCleanup(_marked.clear)
+        self.client = Mock()
+        self.client.submit.return_value = dict(http_status=200, platform_code=0)
+        self.precheck = Mock(return_value=('ok', {}))
+        self.reader = Mock(return_value=(0, 'ok', set()))
+        self.logs = io.StringIO()
+        logpatch = patch('sys.stderr', self.logs)
+        logpatch.start()
+        self.addCleanup(logpatch.stop)
+        # All external collaborators must be injected in this suite.
+        netpatch = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
+        netpatch.start()
+        self.addCleanup(netpatch.stop)
+
+    def cycle(self, model='unit/model', gpus=None, rows=None):
+        return run_cycle(rows or [dict(model_address=model, try_next_gpus=gpus or ['壁砺'])],
+                         self.client, self.reader, precheck=self.precheck)
+
+    def test_marks_read_failure(self):
+        with patch('src.live_cycle.load_marks', side_effect=ValueError()):
+            self.assertEqual(self.cycle(), [dict(action='hold', reason='marks_read_failed')])
+        self.client.submit.assert_not_called()
+        self.precheck.assert_not_called()
+
+    def test_baked_mark_uses_next_choice(self):
+        result = self.cycle('MadeAgents/Hammer2.1-0.5b', ['壁砺', '天垓100'])
+        self.assertEqual(result[0], dict(action='skip', reason='marked:60027', model='MadeAgents/Hammer2.1-0.5b', gpu='Biren_166m'))
+        self.assertEqual(self.client.submit.call_args.args[0]['targetGpu'], 'Iluvatar_bi-100')
+        self.assertEqual(self.logs.getvalue().count('reason=marked:60027'), 1)
+
+    def test_baked_mark_only_never_calls_api(self):
+        result = self.cycle('MadeAgents/Hammer2.1-0.5b')
+        self.assertEqual(result[-1]['reason'], 'all_gpus_done_or_unmapped')
+        self.client.submit.assert_not_called()
+        self.precheck.assert_not_called()
+        self.reader.assert_not_called()
+
+    def test_precheck_marks_all_remaining_choices(self):
+        self.precheck.return_value = ('ok', {'Biren_166m': 60027, 'Iluvatar_bi-100': 60028})
+        result = self.cycle(gpus=['壁砺', '天垓100'])
+        self.assertEqual([d['reason'] for d in result], ['marked:precheck_60027', 'marked:precheck_60028', 'all_gpus_done_or_unmapped'])
+        self.assertEqual(_marked, {('unit/model', 'Biren_166m'): 60027, ('unit/model', 'Iluvatar_bi-100'): 60028})
+        self.cycle(gpus=['壁砺', '天垓100'])
+        self.precheck.assert_called_once()
+        self.client.submit.assert_not_called()
+
+    def test_missing_mapping_submits(self):
+        self.cycle()
+        self.client.submit.assert_called_once()
+
+    def test_nonblocking_platform_labels_and_missing_gpu(self):
+        for label in ('验证失败', '未验证', None):
+            _submitted.clear()
+            response = Mock(status=200)
+            mapping = {} if label is None else {'Biren_166m': {'result': label}}
+            response.read.return_value = json.dumps(dict(code=0, data=dict(verifyResult=mapping))).encode()
+            response.__enter__ = Mock(return_value=response)
+            response.__exit__ = Mock(return_value=False)
+            self.precheck.side_effect = lambda model: precheck_model(model, 'synthetic-secret', Mock(return_value=response))
+            self.cycle()
+        self.assertEqual(self.client.submit.call_count, 3)
+
+    def test_precheck_uses_next_choice(self):
+        self.precheck.return_value = 'ok', {'Biren_166m': 60027}
+        result = self.cycle(gpus=['壁砺', '天垓100'])
+        self.assertEqual(result[0]['reason'], 'marked:precheck_60027')
+        self.assertEqual(self.client.submit.call_args.args[0]['targetGpu'], 'Iluvatar_bi-100')
+        self.precheck.assert_called_once()
+
+    def test_precheck_failure_and_unsafe_status(self):
+        for status in ('transport_error', 'synthetic-secret authorName'):
+            with self.subTest(status=status):
+                _submitted.clear()
+                self.precheck.return_value = status, {}
+                result = self.cycle()
+                self.assertEqual(result[0]['action'], 'precheck')
+                self.assertEqual(result[-1]['action'], 'submitted')
+        self.assertNotIn('synthetic-secret', self.logs.getvalue())
+        self.assertNotIn('authorName', self.logs.getvalue())
+
+    def test_disabled(self):
+        with patch.dict(os.environ, {'LIVE_PRECHECK': '0'}):
+            self.cycle()
+        self.precheck.assert_not_called()
+        self.client.submit.assert_called_once()
+
+    def test_precheck_cap(self):
+        self.precheck.return_value = 'ok', {'Biren_166m': 60027}
+        rows = [dict(model_address=f'unit/m{i}', targetGpu='壁砺') for i in range(62)]
+        for value, count in (('2', 2), ('bad', 60), ('0', 60)):
+            _marked.clear()
+            self.precheck.reset_mock()
+            with patch.dict(os.environ, {'LIVE_MAX_PRECHECKS_PER_CYCLE': value, 'LIVE_MAX_ATTEMPTS_PER_CYCLE': '1'}):
+                result = self.cycle(rows=rows)
+            self.assertEqual(self.precheck.call_count, count)
+            self.assertEqual(result[-1]['reason'], 'precheck_cap')
+            self.assertEqual(sum(d['reason'] == 'precheck_cap' for d in result), 1)
+        self.client.submit.assert_not_called()
+
+    def test_refresh_reuses_precheck_mapping(self):
+        self.precheck.return_value = 'ok', {'Iluvatar_bi-100': 60028}
+        self.reader.return_value = 0, 'ok', {('unit/model', 'Biren_166m')}
+        result = self.cycle(gpus=['壁砺', '天垓100'])
+        self.precheck.assert_called_once()
+        self.client.submit.assert_not_called()
+        self.assertEqual(result[-1]['reason'], 'all_gpus_done_or_unmapped')
+        self.assertEqual(sum(d['reason'] == 'marked:precheck_60028' for d in result), 1)
+
+    def test_rejection_memory_and_exact_log(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=60022)
+        self.cycle()
+        result = self.cycle()
+        self.client.submit.assert_called_once()
+        self.assertEqual(result[0]['reason'], 'marked:60022')
+        self.assertIn('[live] rejected model=unit/model gpu=Biren_166m code=60022 http=200 queue_before=0 reason=CREATE_CONTEST_TASK_GPU_CONFIG_UNSUPPORTED\n', self.logs.getvalue())
+
+    def test_no_state_file_writes(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'state.json'
+            with patch.dict(os.environ, {'LIVE_STATE_FILE': str(path)}), patch.object(Path, 'write_text', side_effect=AssertionError('file write forbidden')):
+                self.cycle()
+            self.assertFalse(path.exists())
+
+
+class TestSyncMarks(unittest.TestCase):
+    def test_sync_and_dry_run(self):
+        with tempfile.TemporaryDirectory() as directory:
+            marks = Path(directory) / 'marks.json'
+            log = Path(directory) / 'log.txt'
+            existing = dict(model='Org/Existing', gpu='Biren_166m', code=60027, reason='first reason', date='2020-01-01')
+            marks.write_text(json.dumps(dict(version=1, marks=[existing])))
+            log.write_text('''timestamp console [live] rejected model=org/existing gpu=Biren_166m code=60028 http=200
+prefix [live] rejected model=Org/New gpu=Iluvatar_bi-150 code=60022 http=200
+[live] rejected model=org/new gpu=Iluvatar_bi-150 code=60027
+[live] skip model=Org/Precheck gpu=Biren_166m reason=marked:precheck_60028
+[live] rejected model=Org/Limit gpu=Biren_166m code=60007
+[live] rejected model=Org/Disabled gpu=Biren_166m code=60014
+[live] rejected model=Org/Success gpu=Biren_166m code=0
+[live] rejected model=invalid gpu=Biren_166m code=60022
+[live] rejected model=Org/Invalid gpu=unknown code=60022
+''')
+            command = [sys.executable, 'scripts/sync_reject_marks.py', str(log), '--marks', str(marks), '--date', '2026-10-17']
+            before = marks.read_bytes()
+            result = subprocess.run(command + ['--dry-run'], capture_output=True, text=True, check=True)
+            self.assertEqual(result.stdout, 'added 2 total 3\n')
+            self.assertEqual(result.stderr, '')
+            self.assertEqual(marks.read_bytes(), before)
+            result = subprocess.run(command, capture_output=True, text=True, check=True)
+            self.assertEqual(result.stdout, 'added 2 total 3\n')
+            rows = json.loads(marks.read_text())['marks']
+            self.assertEqual(rows[0], existing)
+            self.assertEqual(rows[1]['model'], 'Org/New')
+            self.assertEqual(rows[1]['code'], 60022)
+            self.assertEqual(rows[2]['code'], 60028)
+            self.assertTrue(marks.read_bytes().endswith(b'\n'))
+            self.assertEqual(len(load_marks(marks)), 3)
+            result = subprocess.run(command, capture_output=True, text=True, check=True)
+            self.assertEqual(result.stdout, 'added 0 total 3\n')

tests/test_task_add_client.py

@@ -62,6 +62,48 @@ class TestTaskAddClient(unittest.TestCase):
                 with self.assertRaises(ValueError):
                     module.resolve_token()
 
+    def test_file_first_and_empty_external_override(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'credential'
+            path.write_text('file-secret-credential')
+            for external in ('external-secret-credential', ''):
+                with self.subTest(external_empty=not external), mock.patch.dict(os.environ, {
+                    'XC_TOKEN_FILE': str(path), 'XC_TOKEN': 'alias-secret-credential',
+                    'MODELHUB_XC_TOKEN': 'model-secret-credential',
+                    'EXTERNAL_SERVICE_TOKEN': external}, clear=True):
+                    self.assertEqual(module.resolve_token_with_source(),
+                                     ('file-secret-credential', 'XC_TOKEN_FILE'))
+
+    def test_alias_fallback_order(self):
+        env = {'XC_TOKEN_FILE': '/nonexistent/credential', 'XC_TOKEN': 'alias-secret',
+               'MODELHUB_XC_TOKEN': 'model-secret', 'EXTERNAL_SERVICE_TOKEN': 'external-secret'}
+        with mock.patch.dict(os.environ, env, clear=True):
+            for name in module.TOKEN_SOURCES[1:]:
+                self.assertEqual(module.resolve_token_with_source()[1], name)
+                os.environ[name] = ''
+            with self.assertRaises(ValueError):
+                module.resolve_token()
+
+    def test_probe_statuses_are_labels_only(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'private-path'
+            for contents, expected in (('file-secret-credential', 'ok'), ('  ', 'empty'),
+                                       ('<private-rejected-secret>', 'rejected')):
+                path.write_text(contents)
+                with mock.patch.dict(os.environ, {'XC_TOKEN_FILE': str(path),
+                        'XC_TOKEN': 'alias-secret-credential',
+                        'MODELHUB_XC_TOKEN': '<model-private-secret>',
+                        'EXTERNAL_SERVICE_TOKEN': '  '}, clear=True):
+                    statuses = module.token_probe_status()
+                self.assertEqual(statuses, dict(XC_TOKEN_FILE=expected, XC_TOKEN='ok',
+                    MODELHUB_XC_TOKEN='rejected', EXTERNAL_SERVICE_TOKEN='empty'))
+                for secret in ('private', 'credential', 'alias-secret', 'model-private', str(path)):
+                    self.assertNotIn(secret, str(statuses))
+            with mock.patch.dict(os.environ, {}, clear=True):
+                self.assertEqual(set(module.token_probe_status().values()), {'absent'})
+            with mock.patch.dict(os.environ, {'XC_TOKEN_FILE': str(path / 'missing')}, clear=True):
+                self.assertEqual(module.token_probe_status()['XC_TOKEN_FILE'], 'rejected')
+
     def test_placeholder_never_requests(self):
         with mock.patch('urllib.request.urlopen') as urlopen:
             with self.assertRaises(ValueError):