live: EXTERNAL_SERVICE_TOKEN precedence, baked A/B candidates, model|gpu ledger from task/page modelId+gpuType, string total fix, Iluvatar/MTT enums (59 tests)

.gitignore

@@ -19,6 +19,7 @@ build/
 # Dry-run / local artifacts (keep directory)
 reports/*
 !reports/.gitkeep
+!reports/live-gate-summary.md
 
 # Editor / OS
 .DS_Store

Dockerfile

@@ -7,6 +7,7 @@ WORKDIR /app
 
 # Copy only what the runtime needs (see .dockerignore)
 COPY src/ ./src/
+COPY templates/ ./templates/
 
 ENV PORT=8080
 ENV PYTHONUNBUFFERED=1

Makefile

@@ -12,11 +12,11 @@ help:
 	@echo "build-config-dry generate YAML via API (requires token; never creates task)"
 	@echo "probe-readonly authenticated read-only task/page GET"
 	@echo "dry-run-pending read-only adapt pending A→B report"
-	@echo "test         stdlib unit tests + local SIGTERM integration (needs sockets)"
+	@echo "test         stdlib unit tests + offline SIGTERM integration"
 	@echo "build        Docker build using platform base (needs Docker + TCR access)"
 	@echo "smoke-health local HTTP smoke (needs sockets; default test STRATEGY_ID)"
 	@echo "dry-run      read-only pending report (example fallback)"
-	@echo "No register/submit/deploy/push — default live off."
+	@echo "No register/submit/deploy/push — live defaults on; ALLOW_LIVE_SUBMIT=0 stops live."
 
 test:
 	$(PYTHON) -m unittest discover -s tests -v

README.md

@@ -2,8 +2,8 @@
 
 ModelHub XC（信创模盒）**适配智能体** Phase1 本地骨架。
 
-> **不代注册、不代提交。** 发布智能体本身 **不得分**。积分来自成功适配，数字一律 **待平台核实**。  
+> **不代注册；Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自成功适配，数字一律 **待平台核实**。
-> 冲突裁决：**适配 W1 优先**；本仓默认 dry-run，不抢验证队列。
+> 冲突裁决：**适配 W1 优先**；本仓默认 live 开启，仅在实时验证队列低于阈值时提交。
 
 ## 契约要点
 
@@ -12,13 +12,21 @@ ModelHub XC（信创模盒）**适配智能体** Phase1 本地骨架。
 | 端口 | 8080 |
 | 健康检查 | `GET /health` → `{"status":"ok"}` |
 | 启动 | **必须**有非空 `STRATEGY_ID`，否则 exit ≠0 |
-| Live | `ALLOW_LIVE_SUBMIT=1` **且** `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT`；Phase1 仍无真实 `task/add` 客户端 |
+| Live | 默认开启；`ALLOW_LIVE_SUBMIT=0` 为停交开关；必须提供 `STRATEGY_ID` |
-| Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`（勿擅自改默认 FROM） |
+| Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`（保留 FROM） |
 
-闸门以 `src/gate.py` 实现为准：仅 `ALLOW_LIVE_SUBMIT=1` 与
+Victor 的确认已内置，`LIVE_CONFIRM` 仍接受但不再必需。每次 task/add 前，
-`LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT` 同时满足才开逻辑闸门。PRD 示例
+通过 Xc-Token 全量分页读取 onlyMine task/page；非终态计入队列，读取失败停交。
-`ALLOW_LIVE_SUBMIT=true` / `CONFIRM_LIVE=YES` 不受支持；本仓不读取确认文件。
+`LIVE_QUEUE_THRESHOLD=95`（严格小于），`LIVE_MAX_SUBMITS_PER_CYCLE=3`，
-即使开闸，无真实客户端仍返回 blocked；请保持默认关闭。
+`LIVE_CYCLE_SECONDS=600`（最小 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 数字字符串。
+凭证优先级：`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。
 
 ## 快速开始
 
@@ -34,6 +42,7 @@ make build           # 需 Docker + 可达平台 base；失败见 docs/build-not
 
 ```bash
 export STRATEGY_ID=local-dev
+export ALLOW_LIVE_SUBMIT=0  # 本地开发停交
 python3 -m src.server
 # → http://127.0.0.1:8080/health
 ```
@@ -71,6 +80,5 @@ tests/      # stdlib unittest
 
 ## 许可与边界
 
-本仓库用于 Victor 本地/Gitea 交付骨架；任何 ModelHub 写操作须 Victor 本人在控制台或显式批准后执行。
+本仓库用于 Victor 本地/Gitea 交付。Victor 已批准上述队列闸门自动提交；不自动注册。
-
+`make dry-run-pending` 仍仅生成草稿；测试全部 mock 平台接口。
-Phase2：`make dry-run-pending` 只读生成 A→B 草稿，名称硬门与禁提过滤已接入；需 Victor 人工提交/确认。task/add 客户端默认不调用，必须显式注入并通过双闸门、队列及单次预算；测试仅 mock HTTP。

docs/gpu-perf-order.md

@@ -19,8 +19,8 @@
 |-----------------|------------|------------------------------|------|
 | 1（一梯队并列） | 壁砺166M | `Biren_166m` | ≈ 曦望 S2；社区成功量壁砺更高 |
 | 1（一梯队并列） | 曦望 S2 | `Sunrise_pt-200-x1`（抓包证实） | 与壁砺并列，勿臆造先后「性能」 |
-| 2 | 天垓150 | Iluvatar 天垓族（字符串待核） | 条件回退；饱和+唯一性碰撞多 |
+| 2 | 天垓150 | Iluvatar_bi-150 | 条件回退；饱和+唯一性碰撞多 |
-| 3 | 天垓100 | （待核） | 一般不扇出 |
+| 3 | 天垓100 | Iluvatar_bi-100 | 一般不扇出 |
 | 4 | 智铠 mrv-100 | `Iluvatar_mrv-100` | 我方唯一成功样本族；不当最快卡宣传 |
 
 并列一梯队时：若 pending `try-next` 已给出相对顺序则保留；否则可按「壁砺 / 曦望」与池内缺口并列候选，报告标 `order_source=experience`。
@@ -44,7 +44,8 @@
 - 结论：**API**，未公开文档但已抓包证实，依据 [autoconfig-capture-2026-10-16.md](/workspace/modelhub-adapt/logs/autoconfig-capture-2026-10-16.md)。
 - `POST /api/adapt/task/build-config`，query 为 `gpuType` / `framework` / `taskType`，空 body；成功 JSON `code=0`，`data` 为 YAML。
 - UI 无「自动配置」标签；选择框架触发 effect 填入 configParams。模型地址不进入该请求。
-- 抓包观察 Authorization（值未记录）；容器默认 Xc-Token 社区实践，可显式切换 Authorization，尚无 token 鉴权实证。
+- 2026-10-17 主机只读探针确认 task/page 使用 Xc-Token 成功；容器继续使用 Xc-Token，Authorization Bearer 属于浏览器路径。
+- 已核对 GPU 枚举：天垓100=Iluvatar_bi-100，天垓150=Iluvatar_bi-150，智铠100=Iluvatar_mrv-100，MTT S4000/摩尔线程=Mthreads_s4000；保留壁砺及曦望映射。枚举证据不代表官方性能排序。
 - 默认 dry-run：`resolve_token()` 成功时优先 build-config；失败回退 sample，逐条报告 configParams_source、GAP 和 fallback reason。无 token → sample。CONFIG_PARAMS_SOURCE=sample 强制离线；=build-config 强制 API 路径（dry-run with_meta 仍可回退）。make build-config-dry 保持失败非零退出。
 - 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)。

docs/platform-contract-agent.md

@@ -2,7 +2,7 @@
 
 - 日期：2026-10-16 Asia/Shanghai
 - 依据：产品规划 `agent-product-plan-2026-10-16.md` §1 + PRD `02-agent-contribution-prd.md`
-- **冲突裁决：适配 W1 优先**（队列 ≥95 禁 live；本骨架默认 dry-run）
+- **冲突裁决：适配 W1 优先**（队列 ≥95 禁 live；默认 live 开启，实时队列闸门控制）
 
 ## 核实等级
 
@@ -29,7 +29,7 @@
 | **STRATEGY_ID 缺失** | **Phase1：进程启动失败（exit ≠0）** | 产品规划 Phase1 + PRD |
 | 资源 | 约 100m–1 CPU / 256–512Mi | B |
 | **发布智能体得分** | **0 分**（发布 ≠ 得分） | B |
-| Live 默认 | **关**；双重闸门 `ALLOW_LIVE_SUBMIT=1` + `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT` | 本仓 |
+| Live 默认 | **开**（Victor 已批准）；`ALLOW_LIVE_SUBMIT=0` 停交，`LIVE_CONFIRM` 可选 | 本仓 |
 | 积分 | 来自**成功适配**；智能体代提成功计同一成功分；公式/周报 **待平台核实** | B / D |
 
 ## 鉴权双轨（文档索引，Phase1 不代调）
@@ -41,7 +41,7 @@
 
 ## 禁止
 
-- 代注册、代 live 提交、Makefile 含 register/submit/deploy/push
+- 代注册、绕过实时队列闸门、Makefile 含 register/submit/deploy/push
 - 日志/响应/镜像层写入 token 或 STRATEGY_ID **值**
 - 把「智能体已发布」写入千分进度
 
@@ -51,12 +51,17 @@
 |------|------|
 | 异步验证队列 | 适配 W1 优先；≥95 全体停交（含智能体 live） |
 | Codex | hold 期扩池/归因 > 智能体脚手架 |
-| 本骨架 | 默认 dry-run；本轮无真实 `task/add` HTTP 客户端 |
+| 本骨架 | 默认 live；每次 task/add 前实时读取 onlyMine 队列 |
 
 ## 闸门配置以实现为准
 
-`src/gate.py` 只接受 `ALLOW_LIVE_SUBMIT=1` 与
+`ALLOW_LIVE_SUBMIT` 默认 1，0 为 kill switch；`STRATEGY_ID` 必需。
-`LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT`（去除首尾空白后精确匹配）。
+Victor 确认已内置，`LIVE_CONFIRM` 仍接受但不必需。
-PRD 的 `ALLOW_LIVE_SUBMIT=true` / `CONFIRM_LIVE=YES` 仅为需求示例，
+`LIVE_QUEUE_THRESHOLD` 默认 95，只有完整分页的 onlyMine task/page 非终态数严格小于阈值才提交。
-本仓不支持这些值/别名，也不读取确认文件。开闸无客户端仍 blocked。
+缺凭证、HTTP/平台错误、JSON 错误、分页不完整或无效配置均停交。
-这只是本地逻辑闸门，不代表已获 live 授权或已校验队列与候选硬门。
+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；保留该文件可跨周期/重启去重。
+build-config 失败回退 sample 并记录原因；日志不记录响应正文或凭证。

reports/live-gate-summary.md

@@ -0,0 +1,11 @@
+# Live gate update — 2026-10-17
+
+Queue snapshots accept integer or ASCII digit-string totals and derive model/GPU pairs from modelId and gpuType across all statuses. Nonterminal tasks count toward the queue gate. Invalid or incomplete snapshots stop submissions with ledger_rebuild_failed.
+
+Token resolution prefers EXTERNAL_SERVICE_TOKEN, XC_TOKEN_FILE, XC_TOKEN, then MODELHUB_XC_TOKEN; empty and placeholder credentials are skipped. Missing credentials hold even with injected submit clients. Each allowed cycle logs only the selected source name. Xc-Token remains the authentication header.
+
+Offline generators read local adapt evidence only. Input candidates: A=60, B=296; after pending and banned filters: A=46, B=296, total=342. Verified reuse A precedes other A, then B. Baked ledger: 206 pairs and 28 blocked models. Baked JSON ships through COPY src/; .dockerignore does not exclude src/data/*.json.
+
+The ledger merges remote pairs, baked pairs, and persisted model|gpu pairs. Explicit do-not-resubmit models are blocked on every GPU; other models may use the next mapped, unsubmitted GPU in candidate order. Reservations are written before task/add, including failed attempts. Legacy model-only state files fail closed and require offline migration. Startup rebuild logs source counts; each prospective submission refreshes remote identities and queue count.
+
+Validation: make test passes 59 tests using stdlib only. Added coverage for pair dedupe and restart persistence, baked blocks, next-GPU selection, missing tokens, token source logging, digit-string totals, invalid totals, and baked candidate ordering/filtering. SIGTERM uses a mocked HTTP server and real subprocess signal without network calls. Dockerfile FROM unchanged. No git commit, push, tag, live probe, build-config call, or live submission performed.

scripts/bake_candidates.py

@@ -0,0 +1,23 @@
+"""Bake the local pending pool without network access."""
+import json
+import sys
+from collections import Counter
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from src.pending_pool import build_ordered_candidates
+from src.banned import banned_rule
+
+
+def main():
+    rows = build_ordered_candidates('/workspace/modelhub-adapt/pending-submit.md', '/workspace/modelhub-adapt/pending-submit.txt')
+    print('input', dict(Counter(r['section'] for r in rows)))
+    rows = [dict(model_address=r['model_address'], section=r['section'], foothold_class=r['foothold_class'], try_next_gpus=r['try_next_gpus'], framework='vllm') for r in rows if not r['skip_reason'] and not banned_rule(r['model_address'])]
+    rows.sort(key=lambda r: (r['section'] != 'A', r['foothold_class'] != 'verified_reuse'))
+    output = Path(__file__).resolve().parents[1] / 'src/data/candidates.json'
+    output.parent.mkdir(exist_ok=True)
+    output.write_text(json.dumps(rows, ensure_ascii=False, indent=2) + '\n')
+    print('baked', dict(Counter(r['section'] for r in rows)), 'total', len(rows))
+
+
+if __name__ == '__main__':
+    main()

scripts/bake_ledger.py

@@ -0,0 +1,39 @@
+"""Extract only model/GPU identities from local evidence; never copy raw fields."""
+import json
+import re
+import sys
+from pathlib import Path
+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.pending_pool import load_do_not_resubmit
+
+
+def extract(value, source, pairs):
+    if isinstance(value, dict):
+        model = value.get('modelId') or value.get('model') or value.get('model_address')
+        gpu = value.get('gpuType') or value.get('gpu_code') or value.get('gpu') or value.get('gpuLabel')
+        if isinstance(model, str) and re.fullmatch(r'[\w.-]+/[\w.\-/]+', model) and isinstance(gpu, str) and gpu in GPU_ENUMS:
+            pairs.setdefault((model_key(model), GPU_ENUMS[gpu]), source)
+        for child in value.values():
+            extract(child, source, pairs)
+    elif isinstance(value, list):
+        for child in value:
+            extract(child, source, pairs)
+
+
+def main():
+    base = Path('/workspace/modelhub-adapt')
+    files = [base / 'logs/task-page-snapshot-2026-10-17.json', *sorted((base / 'logs').glob('submit-session-*.json')), *sorted((base / 'logs').glob('submit-batch-*.json'))]
+    pairs = {}
+    for path in files:
+        extract(json.loads(path.read_text()), path.name, pairs)
+    blocks = sorted(map(model_key, load_do_not_resubmit(base / 'do-not-resubmit.md')))
+    output = Path(__file__).resolve().parents[1] / 'src/data/submitted-ledger.json'
+    output.parent.mkdir(exist_ok=True)
+    output.write_text(json.dumps(dict(pairs=[dict(model=m, gpu=g, source=s) for (m,g),s in sorted(pairs.items())], blocked_models=blocks), indent=2) + '\n')
+    print('baked pairs', len(pairs), 'blocked models', len(blocks))
+
+
+if __name__ == '__main__':
+    main()

src/banned.py

@@ -0,0 +1,15 @@
+"""Case-insensitive live model exclusions."""
+import re
+
+RULES = (
+    ('gpt2', r'gpt-?2|dialogpt|distilgpt2'),
+    ('phi', r'phi[-_. ]?(?:3(?:\.5)?|4)(?:\D|$)'),
+    ('nemotron-h', r'nemotron-h'), ('olmo2', r'olmo-?2'),
+    ('lfm2', r'lfm2'), ('spark2.5', r'spark2_5|spark-x2\.5'),
+    ('breeze', r'breeze'), ('gemma2', r'gemma-?2'),
+    ('mistral-v0.1', r'mistral.*v0\.1'), ('tinyllama', r'tinyllama'),
+)
+
+
+def banned_rule(model):
+    return next((name for name, pattern in RULES if re.search(pattern, model, re.I)), None)

src/config_params.py

@@ -10,6 +10,11 @@ GPU_ENUMS = {
     '壁砺': 'Biren_166m', '壁砺166M': 'Biren_166m', 'Biren_166m': 'Biren_166m',
     '曦望': 'Sunrise_pt-200-x1', '曦望 S2': 'Sunrise_pt-200-x1',
     'Sunrise_pt-200-x1': 'Sunrise_pt-200-x1',
+    '天垓100': 'Iluvatar_bi-100', 'Iluvatar_bi-100': 'Iluvatar_bi-100',
+    '天垓150': 'Iluvatar_bi-150', 'Iluvatar_bi-150': 'Iluvatar_bi-150',
+    '智铠': 'Iluvatar_mrv-100', '智铠100': 'Iluvatar_mrv-100',
+    '智铠 mrv-100': 'Iluvatar_mrv-100', 'Iluvatar_mrv-100': 'Iluvatar_mrv-100',
+    'MTT S4000': 'Mthreads_s4000', '摩尔线程': 'Mthreads_s4000', 'Mthreads_s4000': 'Mthreads_s4000',
 }
 
 

src/gate.py

@@ -1,123 +1,74 @@
-"""Dual live-submit gate. Default: dry-run / blocked.
+"""Live defaults on with baked-in approval. LIVE_CONFIRM remains optional.
 
-Live is allowed only when ALL of:
+Environment hints only block; a fresh authenticated API snapshot authorizes each add.
-  - ALLOW_LIVE_SUBMIT=1
-  - LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT
-  - STRATEGY_ID present
-
-Phase2 client exists but requires explicit injection. No automatic wiring.
-Queue ≥95 blocks; at most one injected submit per process.
 """
-
-from __future__ import annotations
-
 import os
-import threading
-from .queue_depth import queue_depth
 from dataclasses import dataclass
-from typing import Any, Callable, Dict, Optional, Protocol
-
 from .strategy import strategy_id_present
 
-_submit_count = 0
+LIVE_CONFIRM_VALUE = 'I_CONFIRM_LIVE_SUBMIT'
-_submit_lock = threading.Lock()
-
-LIVE_CONFIRM_VALUE = "I_CONFIRM_LIVE_SUBMIT"
 
 
-class SubmitClient(Protocol):
+def live_enabled():
-    """Minimal interface for a submit client (injected fake in tests)."""
+    return os.environ.get('ALLOW_LIVE_SUBMIT', '1').strip() == '1'
 
-    def submit(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+
-        ...
+def queue_threshold():
+    value = int(os.environ.get('LIVE_QUEUE_THRESHOLD', '95'))
+    if value < 1:
+        raise ValueError('invalid threshold')
+    return value
 
 
 @dataclass(frozen=True)
 class GateDecision:
-    mode: str  # "dry-run" | "live" | "blocked"
+    mode: str
     allowed: bool
     reason: str
 
 
-def _allow_live_flag() -> bool:
+def evaluate_gate():
-    return os.environ.get("ALLOW_LIVE_SUBMIT", "0").strip() == "1"
+    if not live_enabled():
-
+        return GateDecision('dry-run', False, 'kill_switch')
-
-def _live_confirm_ok() -> bool:
-    return os.environ.get("LIVE_CONFIRM", "").strip() == LIVE_CONFIRM_VALUE
-
-
-def evaluate_gate() -> GateDecision:
-    """Evaluate whether live submit is permitted."""
     if not strategy_id_present():
-        return GateDecision(
+        return GateDecision('blocked', False, 'STRATEGY_ID missing or empty')
-            mode="blocked",
+    try:
-            allowed=False,
+        threshold = queue_threshold()
-            reason="STRATEGY_ID missing or empty",
+    except ValueError:
-        )
+        return GateDecision('blocked', False, 'invalid_threshold')
-    if not _allow_live_flag():
+    # Evaluate each configured hint independently; neither can override the other.
-        return GateDecision(
+    for key in ('ASYNC_QUEUE_DEPTH', 'QUEUE_SNAPSHOT_JSON'):
-            mode="dry-run",
+        if key not in os.environ:
-            allowed=False,
+            continue
-            reason="ALLOW_LIVE_SUBMIT is not 1 (default dry-run)",
+        if key == 'ASYNC_QUEUE_DEPTH':
-        )
+            try:
-    if not _live_confirm_ok():
+                depth = int(os.environ[key])
-        return GateDecision(
+            except ValueError:
-            mode="blocked",
+                depth = None
-            allowed=False,
+        else:
-            reason=f"LIVE_CONFIRM must be exactly {LIVE_CONFIRM_VALUE!r}",
+            import json
-        )
+            from pathlib import Path
-    depth = queue_depth()
+            try:
-    if depth is None and ("ASYNC_QUEUE_DEPTH" in os.environ or "QUEUE_SNAPSHOT_JSON" in os.environ):
+                data = json.loads(Path(os.environ[key]).read_text())
-        return GateDecision("blocked", False, "queue depth invalid/unknown; human confirmation required")
+                value = next(data[k] for k in ('depth', 'queue_depth', 'async_queue_depth', 'ASYNC_QUEUE_DEPTH') if k in data)
-    if depth is not None and depth >= 95:
+                depth = None if isinstance(value, bool) else int(str(value))
-        return GateDecision("blocked", False, "queue ≥95: 禁 live")
+            except (OSError, ValueError, TypeError, StopIteration):
-    return GateDecision(
+                depth = None
-        mode="live",
+        if depth is None or depth < 0:
-        allowed=True,
+            return GateDecision('blocked', False, 'queue_hint_invalid')
-        reason="dual gate open (ALLOW_LIVE_SUBMIT=1 and LIVE_CONFIRM set)",
+        if depth >= threshold:
-    )
+            return GateDecision('blocked', False, f'queue>={threshold} hint_count={depth}')
+    return GateDecision('live', True, 'live_enabled')
 
 
-def attempt_live_submit(
+def attempt_live_submit(payload, client=None, queue_reader=None):
-    payload: Dict[str, Any],
+    """Compatibility entry point; delegates to the same fresh queue cycle."""
-    client: Optional[SubmitClient] = None,
-) -> Dict[str, Any]:
-    """Attempt a live submit only if the gate opens and a client is injected.
-
-    Phase2 TaskAddClient must be explicitly injected and dual-gated.
-    Tests must use fake clients or mock HTTP only.
-    """
-    global _submit_count
-    decision = evaluate_gate()
-    if not decision.allowed:
-        return {
-            "ok": False,
-            "mode": decision.mode,
-            "reason": decision.reason,
-            "submitted": False,
-        }
     if client is None:
-        return {
+        return dict(ok=False, submitted=False, mode='blocked', reason='no real request without explicit client')
-            "ok": False,
+    from .live_cycle import run_cycle
-            "mode": "blocked",
+    decisions = run_cycle([dict(model_address=payload.get('modelAddress'), targetGpu=payload.get('targetGpu'),
-            "reason": (
+                                framework=payload.get('framework', 'vllm'), configParams=payload.get('configParams'))],
-                "Phase2 task/add client exists; explicit client injection required; "
+                          client=client, queue_reader=queue_reader)
-                "no real request without injection and dual gate"
+    result = decisions[-1]
-            ),
+    return dict(result, ok=result['action'] == 'submitted', submitted=result['action'] == 'submitted',
-            "submitted": False,
+                mode='live' if result['action'] == 'submitted' else 'blocked')
-        }
-    with _submit_lock:
-        if _submit_count >= 1:
-            return dict(ok=False, mode="blocked", submitted=False,
-                        reason="process limit: at most 1 real add")
-        # Reserve before invocation: a transport failure may still have created a task.
-        _submit_count += 1
-    result = client.submit(payload)
-    return {
-        "ok": True,
-        "mode": "live",
-        "reason": decision.reason,
-        "submitted": True,
-        "result": result,
-    }

src/ledger.py

@@ -0,0 +1,35 @@
+"""Pair reservations and explicit model blocks; no credentials or response bodies."""
+import json
+from pathlib import Path
+from .live_queue import model_key, read_queue
+
+DATA = Path(__file__).parent / 'data/submitted-ledger.json'
+
+
+def load_baked(path=DATA):
+    data = json.loads(Path(path).read_text())
+    return ({(model_key(r['model']), r['gpu']) for r in data['pairs']},
+            {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):
+    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)

src/live_cycle.py

@@ -0,0 +1,143 @@
+"""Queue-gated live cycles; all remote collaborators are injectable."""
+import json
+import os
+import sys
+import threading
+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 .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
+
+_submitted = set()
+_cycle_lock = threading.Lock()
+
+
+def run_cycle(candidates=None, client=None, queue_reader=None, now=None):
+    with _cycle_lock:
+        return _run_cycle(candidates, client, queue_reader, now)
+
+
+def _run_cycle(candidates, client, queue_reader, now):
+    decisions = []
+
+    def emit(action, reason, **fields):
+        decision = dict(action=action, reason=reason, **fields)
+        decisions.append(decision)
+        print('[live] ' + action + ' ' + ' '.join(f'{k}={str(v).replace(chr(10), " ").replace(chr(13), " ")}' for k, v in dict(fields, reason=reason).items()), file=sys.stderr)
+
+    gate = evaluate_gate()
+    if not gate.allowed:
+        emit('hold', gate.reason)
+        return decisions
+    try:
+        _, source = resolve_token_with_source()
+    except (ValueError, OSError, UnicodeError):
+        emit('hold', 'missing/invalid_token')
+        return decisions
+    print('[live] token source=' + source, file=sys.stderr)
+    try:
+        cap = int(os.environ.get('LIVE_MAX_SUBMITS_PER_CYCLE', '3'))
+        if cap < 1:
+            raise ValueError
+    except ValueError:
+        emit('hold', 'invalid_cycle_cap')
+        return decisions
+    if candidates is None:
+        try:
+            if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
+                candidates = build_ordered_candidates()
+            else:
+                candidates = json.loads((Path(__file__).parent / 'data/candidates.json').read_text())
+        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 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
+        baked, blocked = load_baked()
+        known = persisted | baked
+    except (OSError, ValueError, KeyError, TypeError):
+        emit('hold', 'state_read_failed')
+        return decisions
+    reader = queue_reader or read_queue
+    client = client if client is not None else TaskAddClient()
+    attempts = 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', ''))))
+        if rule:
+            emit('skip', 'banned:' + rule, model=model)
+            continue
+        if not model or '<' in model or '>' in model:
+            emit('skip', 'placeholder')
+            continue
+        key = model_key(model)
+        if key in blocked:
+            emit('skip', 'do-not-resubmit', model=model)
+            continue
+        if candidate.get('skip_reason'):
+            emit('skip', 'pending_filter', model=model)
+            continue
+        if attempts >= 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)
+        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()
+            if type(count) is not int or count < 0:
+                raise ValueError
+            known.update(remote)
+        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)
+        if gpu is None:
+            emit('skip', 'all_gpus_done_or_unmapped', model=model)
+            continue
+        if count >= queue_threshold():
+            emit('hold', f'queue>={queue_threshold()}', count=count)
+            break
+        framework = candidate.get('framework') or 'vllm'
+        try:
+            yaml, meta = build_config_params_with_meta(gpu, framework)
+        except Exception:
+            emit('skip', 'config_failed', model=model)
+            continue
+        payload = build_payload(model, GPU_ENUMS[gpu], framework, os.environ['STRATEGY_ID'].strip(), yaml)
+        if any(not value or '<' in str(value) or '>' in str(value) for value in payload.values()):
+            emit('skip', 'placeholder', model=model)
+            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)
+            status, code = result.get('http_status'), result.get('platform_code')
+            status = status if type(status) is int else None
+            code = code if type(code) is int else None
+            reason = 'task_add_response' if status is not None else 'transport_error'
+        except Exception:
+            status, code, reason = None, None, 'submit_error'
+        emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
+             http=status, code=code, **meta)
+    return decisions

src/live_queue.py

@@ -0,0 +1,68 @@
+"""Fresh authenticated queue snapshots. Never expose response bodies or credentials."""
+import json
+import os
+import urllib.request
+from .task_add_client import resolve_token
+
+TERMINAL = {'success', 'failed', 'fail', 'error', 'cancelled', 'canceled'}
+
+
+def model_key(value):
+    value = str(value or '').strip().lower().rstrip('/')
+    for prefix in ('https://huggingface.co/', 'http://huggingface.co/', 'https://modelscope.cn/models/'):
+        if value.startswith(prefix):
+            value = value[len(prefix):]
+    return value
+
+
+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."""
+    try:
+        token = resolve_token()
+    except (ValueError, OSError, UnicodeError):
+        return None, 'missing/invalid_token', set()
+    count, models, seen, total = 0, set(), 0, None
+    for page in range(1, 21):
+        try:
+            request = urllib.request.Request(
+                os.environ.get('MODELHUB_API_BASE', 'https://modelhub.org.cn').rstrip('/')
+                + f'/api/adapt/task/page?onlyMine=true&pageSize=100&current={page}',
+                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()
+                body = json.loads(response.read())
+        except (ValueError, UnicodeError):
+            return None, 'malformed_json', set()
+        except OSError:
+            return None, 'transport/http_error', set()
+        if not isinstance(body, dict) or type(body.get('code')) not in (int, str) or str(body['code']) != '0':
+            return None, 'platform_error', set()
+        data = body.get('data')
+        if not isinstance(data, dict) or not valid_total(data.get('total')):
+            return None, 'malformed_data', set()
+        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()
+        if total is not None and total != data['total']:
+            return None, 'pagination_total_changed', set()
+        total = data['total']
+        seen += len(records)
+        if seen > total:
+            return None, 'pagination_inconsistent', set()
+        for record in records:
+            count += str(record.get('status', '')).strip().lower() not in TERMINAL
+            model, gpu = model_key(record.get('modelId')), record.get('gpuType')
+            if model and isinstance(gpu, str) and gpu:
+                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)}
+        if not records:
+            break
+    return None, 'incomplete_pagination', set()

src/server.py

@@ -17,6 +17,8 @@ from typing import Optional
 
 from .health import handle_get
 from .strategy import require_strategy_id
+from .gate import live_enabled
+from .live_cycle import run_cycle
 
 _shutdown = threading.Event()
 _httpd: Optional[ThreadingHTTPServer] = None
@@ -56,6 +58,27 @@ def _on_sigterm(signum: int, frame) -> None:  # noqa: ARG001
         threading.Thread(target=httpd.shutdown, daemon=True).start()
 
 
+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)
+    except Exception:
+        sys.stderr.write('[live] hold reason=ledger_rebuild_failed\n')
+    try:
+        interval = max(60, int(os.environ.get('LIVE_CYCLE_SECONDS', '600')))
+    except ValueError:
+        interval = 600
+    while not _shutdown.is_set():
+        try:
+            run_cycle()
+        except Exception:
+            sys.stderr.write('[live] hold reason=cycle_error\n')
+        if _shutdown.wait(interval):
+            break
+
+
 def main() -> None:
     global _httpd
 
@@ -71,6 +94,9 @@ def main() -> None:
     httpd.daemon_threads = True
     _httpd = httpd
 
+    if live_enabled():
+        threading.Thread(target=_live_worker, daemon=True).start()
+
     sys.stderr.write(f"Listening on {host}:{port} (GET /health, GET /status)\n")
     try:
         httpd.serve_forever(poll_interval=0.5)

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",
 )
 
 
@@ -62,20 +62,9 @@ def _token_from_file(path: str) -> bool:
 
 def external_service_token_present() -> bool:
     """Whether any known token env / file alias is present (boolean only)."""
-    for key in _TOKEN_ENV_KEYS:
+    from .task_add_client import resolve_token_with_source
-        raw = os.environ.get(key)
+    try:
-        if raw is None:
+        resolve_token_with_source()
-            continue
-        value = raw.strip()
-        if not value:
-            continue
-        if key == "XC_TOKEN_FILE":
-            if _token_from_file(value):
-                return True
-            continue
-        # Reject obvious placeholders (community practice, level C)
-        lowered = value.lower()
-        if lowered in ("tmp", "placeholder", "changeme", "xxx"):
-            continue
         return True
-    return False
+    except (ValueError, OSError, UnicodeError):
+        return False

src/task_add_client.py

@@ -19,13 +19,22 @@ def build_payload(model_address, target_gpu='<TARGET_GPU>', framework='<FRAMEWOR
                 configParams=config_params)
 
 
+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
+    raise ValueError('Missing or invalid service credential')
+
+
 def resolve_token():
-    file = os.environ.get('XC_TOKEN_FILE')
+    return resolve_token_with_source()[0]
-    token = Path(file).read_text().strip() if file else next(
-        (os.environ[k].strip() for k in ('XC_TOKEN', 'MODELHUB_XC_TOKEN', 'EXTERNAL_SERVICE_TOKEN') if os.environ.get(k)), '')
-    if not token or re.search(r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp)(?:$|[^a-z0-9])', token):
-        raise ValueError('Missing or invalid service credential')
-    return token
 
 
 class TaskAddClient:

tests/test_config_params.py

@@ -45,7 +45,7 @@ class TestConfigParams(unittest.TestCase):
             yaml, meta = build_config_params_with_meta('壁砺')
             self.assertEqual(meta['configParams_fallback_reason'], 'GAP: transport/timeout')
         # Unmapped GPU → no API call
-        for gpu in ('天垓150', '智铠 mrv-100', 'MetaX_c-500', '<GPU>', None):
+        for gpu in ('MetaX', 'MetaX_c-500', '<GPU>', None):
             with patch('src.config_params.resolve_token', return_value='synthetic'), patch('src.config_params.build_config') as client:
                 _, meta = build_config_params_with_meta(gpu)
                 self.assertIn('GAP: gpu enum unmapped', meta['configParams_fallback_reason'])

tests/test_gate.py

@@ -1,159 +1,44 @@
-"""Gate defaults to dry-run / blocked; live only with dual confirm + STRATEGY_ID."""
+"""Default live approval, kill switch, strategy and conservative hint gates."""
-
-from __future__ import annotations
-
 import os
+import tempfile
 import unittest
-from unittest import mock
+from pathlib import Path
-
+from unittest.mock import patch
-from src.gate import LIVE_CONFIRM_VALUE, attempt_live_submit, evaluate_gate
+from src.gate import evaluate_gate, attempt_live_submit, LIVE_CONFIRM_VALUE
-
-
-class FakeClient:
-    def __init__(self) -> None:
-        self.calls = []
-
-    def submit(self, payload):
-        self.calls.append(payload)
-        return {"task_id": "fake-1"}
 
 
 class TestGate(unittest.TestCase):
     def setUp(self):
-        self.counter = mock.patch('src.gate._submit_count', 0)
+        env = patch.dict(os.environ, {'STRATEGY_ID': 'sid'}, clear=True)
-        self.counter.start()
+        env.start()
-        self.addCleanup(self.counter.stop)
+        self.addCleanup(env.stop)
-        self.env = mock.patch.dict(os.environ, {}, clear=True)
-        self.env.start()
-        self.addCleanup(self.env.stop)
 
-    def test_queue_and_process_limit(self):
+    def test_default_live_and_optional_confirmation(self):
-        client = FakeClient()
+        self.assertTrue(evaluate_gate().allowed)
-        with mock.patch.dict(os.environ, {
+        for value in ('', 'YES', LIVE_CONFIRM_VALUE):
-            'STRATEGY_ID': 'sid', 'ALLOW_LIVE_SUBMIT': '1',
+            with patch.dict(os.environ, {'LIVE_CONFIRM': value}):
-            'LIVE_CONFIRM': LIVE_CONFIRM_VALUE, 'ASYNC_QUEUE_DEPTH': '95'
+                self.assertTrue(evaluate_gate().allowed)
-        }):
-            self.assertIn('≥95', attempt_live_submit({}, client)['reason'])
-            self.assertEqual(client.calls, [])
-            os.environ['ASYNC_QUEUE_DEPTH'] = '94'
-            self.assertTrue(attempt_live_submit({}, client)['submitted'])
-            self.assertFalse(attempt_live_submit({}, client)['submitted'])
-            self.assertEqual(len(client.calls), 1)
 
-    def _clear_gate_env(self):
+    def test_kill_switch_and_missing_strategy(self):
-        return mock.patch.dict(
+        for env in ({'ALLOW_LIVE_SUBMIT': '0'}, {'ALLOW_LIVE_SUBMIT': 'true'}, {'STRATEGY_ID': '   '}):
-            os.environ,
+            with patch.dict(os.environ, env):
-            {
+                self.assertFalse(evaluate_gate().allowed)
-                "STRATEGY_ID": "sid-1",
-                "ALLOW_LIVE_SUBMIT": "0",
-                "LIVE_CONFIRM": "",
-            },
-            clear=False,
-        )
 
-    def test_default_dry_run(self) -> None:
+    def test_hints_and_threshold(self):
-        with self._clear_gate_env():
+        for value in ('95', 'invalid', '-1'):
-            d = evaluate_gate()
+            with patch.dict(os.environ, {'ASYNC_QUEUE_DEPTH': value}):
-        self.assertFalse(d.allowed)
+                self.assertFalse(evaluate_gate().allowed)
-        self.assertEqual(d.mode, "dry-run")
+        with patch.dict(os.environ, {'ASYNC_QUEUE_DEPTH': '94'}):
+            self.assertTrue(evaluate_gate().allowed)
+        with patch.dict(os.environ, {'LIVE_QUEUE_THRESHOLD': 'bad'}):
+            self.assertFalse(evaluate_gate().allowed)
+        with tempfile.TemporaryDirectory() as d:
+            p = Path(d) / 'hint.json'
+            p.write_text('{"depth":95}')
+            with patch.dict(os.environ, {'QUEUE_SNAPSHOT_JSON': str(p), 'ASYNC_QUEUE_DEPTH': '1'}):
+                self.assertFalse(evaluate_gate().allowed)
 
-    def test_missing_strategy_blocked(self) -> None:
+    def test_no_client_no_network(self):
-        env = {
+        with patch('urllib.request.urlopen') as network:
-            "ALLOW_LIVE_SUBMIT": "1",
+            self.assertFalse(attempt_live_submit({})['submitted'])
-            "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
+            network.assert_not_called()
-        }
-        # Remove STRATEGY_ID
-        base = {k: v for k, v in os.environ.items() if k != "STRATEGY_ID"}
-        base.update(env)
-        with mock.patch.dict(os.environ, base, clear=True):
-            d = evaluate_gate()
-        self.assertFalse(d.allowed)
-        self.assertEqual(d.mode, "blocked")
-
-    def test_allow_without_confirm_blocked(self) -> None:
-        with mock.patch.dict(
-            os.environ,
-            {
-                "STRATEGY_ID": "sid-1",
-                "ALLOW_LIVE_SUBMIT": "1",
-                "LIVE_CONFIRM": "YES",
-            },
-            clear=False,
-        ):
-            d = evaluate_gate()
-        self.assertFalse(d.allowed)
-        self.assertEqual(d.mode, "blocked")
-
-    def test_dual_gate_opens(self) -> None:
-        with mock.patch.dict(
-            os.environ,
-            {
-                "STRATEGY_ID": "sid-1",
-                "ALLOW_LIVE_SUBMIT": "1",
-                "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
-            },
-            clear=False,
-        ):
-            d = evaluate_gate()
-        self.assertTrue(d.allowed)
-        self.assertEqual(d.mode, "live")
-
-    def test_open_gate_without_client_still_blocked(self) -> None:
-        """Phase2: an open gate never constructs a network client."""
-        with mock.patch.dict(
-            os.environ,
-            {
-                "STRATEGY_ID": "sid-1",
-                "ALLOW_LIVE_SUBMIT": "1",
-                "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
-            },
-            clear=False,
-        ):
-            result = attempt_live_submit({"modelAddress": "x/y"})
-        self.assertFalse(result["submitted"])
-        self.assertIn("no real", result["reason"].lower())
-        self.assertEqual(result["mode"], "blocked")
-        self.assertFalse(result["ok"])
-
-    def test_prd_example_and_whitespace_id_never_submit(self) -> None:
-        cases = [
-            {"ALLOW_LIVE_SUBMIT": "true", "LIVE_CONFIRM": LIVE_CONFIRM_VALUE},
-            {"ALLOW_LIVE_SUBMIT": "1", "LIVE_CONFIRM": "", "CONFIRM_LIVE": "YES"},
-            {"ALLOW_LIVE_SUBMIT": "1", "LIVE_CONFIRM": LIVE_CONFIRM_VALUE, "STRATEGY_ID": "   "},
-        ]
-        for overrides in cases:
-            with self.subTest(overrides=overrides):
-                client = FakeClient()
-                env = {"STRATEGY_ID": "sid-test", **overrides}
-                with mock.patch.dict(os.environ, env, clear=True):
-                    result = attempt_live_submit({}, client=client)
-                self.assertFalse(result["submitted"])
-                self.assertFalse(result["ok"])
-                self.assertEqual(client.calls, [])
-                self.assertTrue(result["reason"])
-
-    def test_injected_fake_client(self) -> None:
-        client = FakeClient()
-        with mock.patch.dict(
-            os.environ,
-            {
-                "STRATEGY_ID": "sid-1",
-                "ALLOW_LIVE_SUBMIT": "1",
-                "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
-            },
-            clear=False,
-        ):
-            result = attempt_live_submit({"modelAddress": "x/y"}, client=client)
-        self.assertTrue(result["submitted"])
-        self.assertEqual(len(client.calls), 1)
-
-    def test_default_attempt_rejects(self) -> None:
-        client = FakeClient()
-        with self._clear_gate_env():
-            result = attempt_live_submit({"modelAddress": "x/y"}, client=client)
-        self.assertFalse(result["submitted"])
-        self.assertEqual(len(client.calls), 0)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_live_cycle.py

@@ -0,0 +1,110 @@
+import io
+import os
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+from src.live_cycle import run_cycle, _submitted
+
+
+def candidate(name, section='B', foothold='unknown'):
+    return dict(model_address='org/' + name, section=section, foothold_class=foothold, try_next_gpus=['壁砺'])
+
+
+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.start()
+        self.addCleanup(self.env.stop)
+        _submitted.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'))
+        network.start()
+        self.addCleanup(network.stop)
+        logs = patch('sys.stderr', new_callable=io.StringIO)
+        self.logs = logs.start()
+        self.addCleanup(logs.stop)
+
+    def cycle(self, rows=None, counts=(94,)):
+        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 test_high_and_failed_queue(self):
+        for n in (95, None):
+            decisions = self.cycle(counts=(n,))
+            self.assertEqual(decisions[-1]['action'], 'hold')
+        self.client.submit.assert_not_called()
+        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'])
+        self.assertIn('cycle_cap', self.logs.getvalue())
+
+    def test_midcycle(self):
+        self.cycle([candidate(str(i)) for i in range(5)], (93, 94, 95))
+        self.assertEqual(self.client.submit.call_count, 2)
+
+    def test_kill_and_threshold(self):
+        for env in ({'ALLOW_LIVE_SUBMIT': '0'}, {'LIVE_QUEUE_THRESHOLD': 'bad'}, {'LIVE_QUEUE_THRESHOLD': '20'}):
+            with patch.dict(os.environ, env):
+                self.cycle(counts=(20,))
+        self.client.submit.assert_not_called()
+        with patch.dict(os.environ, {'LIVE_QUEUE_THRESHOLD': '100'}):
+            self.cycle(counts=(99,))
+        self.client.submit.assert_called_once()
+
+    def test_banned_duplicate_persistence(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())
+        self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
+
+    def test_remote_duplicate_and_transport_reservation(self):
+        run_cycle([candidate('remote')], self.client, lambda: (1, 'ok', {('org/remote', 'Biren_166m')}))
+        self.client.submit.assert_not_called()
+        self.client.submit.side_effect = OSError('do not log')
+        self.cycle()
+        self.cycle(counts=())
+        self.client.submit.assert_called_once()
+        self.assertNotIn('do not log', self.logs.getvalue())
+
+    def test_missing_token_real_reader(self):
+        with patch.dict(os.environ, {'XC_TOKEN': ''}):
+            run_cycle([candidate('a')], self.client)
+        self.client.submit.assert_not_called()
+        self.assertIn('missing/invalid_token', self.logs.getvalue())
+
+    def test_worker_waits_and_stops(self):
+        from src.server import _live_worker
+        with patch('src.server._shutdown') as shutdown, patch('src.server.run_cycle') as cycle:
+            shutdown.is_set.return_value = False
+            shutdown.wait.return_value = True
+            with patch.dict(os.environ, {'LIVE_CYCLE_SECONDS': '1'}):
+                _live_worker()
+            cycle.assert_called_once()
+            shutdown.wait.assert_called_once_with(60)
+
+    def test_unauthenticated_platform_zero_submits(self):
+        import json
+        response = Mock()
+        response.__enter__ = Mock(return_value=response)
+        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):
+            run_cycle([candidate('a')], self.client)
+        self.client.submit.assert_not_called()
+
+    def test_unmapped_and_placeholder(self):
+        rows = [dict(model_address='org/unmapped', targetGpu='unknown'), candidate('<MODEL>')]
+        self.cycle(rows, ())
+        self.client.submit.assert_not_called()
+        self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
+        self.assertIn('placeholder', self.logs.getvalue())

tests/test_live_ledger.py

@@ -0,0 +1,71 @@
+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.ledger import load_baked, rebuild
+from src.banned import banned_rule
+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.start()
+        self.addCleanup(env.stop)
+        _submitted.clear()
+        self.addCleanup(_submitted.clear)
+        net = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
+        net.start()
+        self.addCleanup(net.stop)
+        self.client = Mock()
+        self.client.submit.return_value = dict(http_status=200, platform_code=0)
+
+    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):
+        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.cycle([('unit/model', 'Biren_166m')])
+        self.client.submit.assert_called_once()
+
+    def test_baked_and_blocked(self):
+        pairs, blocked = load_baked()
+        model, gpu = next(iter(pairs))
+        self.cycle(model=model, gpus=[gpu])
+        self.cycle(model=next(iter(blocked)))
+        self.client.submit.assert_not_called()
+
+    def test_failure_and_missing_token(self):
+        result = run_cycle([dict(model_address='unit/model', try_next_gpus=['壁砺'])], self.client, lambda: (None, 'failed', set()))
+        self.assertEqual(result[-1]['reason'], 'ledger_rebuild_failed')
+        with patch.dict(os.environ, {'XC_TOKEN': ''}):
+            self.cycle()
+        self.client.submit.assert_not_called()
+
+    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.cycle()
+        self.assertEqual(log.getvalue().count('token source=EXTERNAL_SERVICE_TOKEN'), 1)
+        self.assertNotIn('platform-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')
+
+    def test_candidates_and_rebuild(self):
+        rows = json.loads(Path('src/data/candidates.json').read_text())
+        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')}))
+        self.assertIn(('unit/model', 'Biren_166m'), pairs)
+        self.assertEqual(sizes[0], 1)

tests/test_live_queue.py

@@ -0,0 +1,55 @@
+import json
+import os
+import unittest
+from unittest.mock import patch, MagicMock
+from src.live_queue import read_queue
+
+
+class TestLiveQueue(unittest.TestCase):
+    def response(self, body):
+        r = MagicMock()
+        r.__enter__.return_value = r
+        r.status = 200
+        r.read.return_value = json.dumps(body).encode()
+        return r
+
+    def test_pagination_and_status(self):
+        pages = [self.response({'code': 0, 'data': {'total': '3', 'records': [
+            {'status': 'waiting', 'modelId': 'org/a', 'gpuType': 'Biren_166m'}, {'status': 'success', 'verifyResult': -1}]}}),
+            self.response({'code': 0, 'data': {'total': '3', 'records': [{'status': '等待中'}]}})]
+        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', side_effect=pages) as call:
+            count, why, models = read_queue()
+        self.assertEqual(count, 2)
+        self.assertIn(('org/a', 'Biren_166m'), models)
+        self.assertIn('current=2', call.call_args.args[0].full_url)
+
+    def test_fail_closed(self):
+        for body in ({'code': 0, 'data': {'total': 'bad', 'records': []}}, {'code': 40100}, {'code': 40101}, {'code': 0, 'data': {'total': 1, 'records': []}}, []):
+            with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', return_value=self.response(body)):
+                self.assertIsNone(read_queue()[0])
+        with patch.dict(os.environ, {}, clear=True), patch('urllib.request.urlopen') as call:
+            self.assertIsNone(read_queue()[0])
+            call.assert_not_called()
+
+    def test_page_limit_and_transport(self):
+        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True):
+            with patch('urllib.request.urlopen', return_value=self.response({'code': 0, 'data': {'total': 21, 'records': [{'status': 'waiting'}]}})) as call:
+                self.assertIsNone(read_queue()[0])
+                self.assertEqual(call.call_count, 20)
+            with patch('urllib.request.urlopen', side_effect=OSError('synthetic-secret')):
+                self.assertNotIn('synthetic-secret', str(read_queue()))
+
+    def test_invalid_json_http_and_auth_header(self):
+        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True):
+            r = self.response({})
+            r.read.return_value = b'invalid synthetic-secret'
+            with patch('urllib.request.urlopen', return_value=r):
+                self.assertEqual(read_queue()[1], 'malformed_json')
+            r.status = 503
+            with patch('urllib.request.urlopen', return_value=r):
+                self.assertIsNone(read_queue()[0])
+            with patch('urllib.request.urlopen', return_value=self.response({'code': 0, 'data': {'total': 0, 'records': []}})) as call:
+                self.assertEqual(read_queue()[0], 0)
+                req = call.call_args.args[0]
+                self.assertEqual(req.get_header('Xc-token'), 'synthetic-secret')
+                self.assertEqual(req.method, 'GET')

tests/test_sigterm.py

@@ -9,7 +9,6 @@ import subprocess
 import sys
 import time
 import unittest
-import urllib.request
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -17,35 +16,36 @@ ROOT = Path(__file__).resolve().parents[1]
 
 class TestSigterm(unittest.TestCase):
     def test_sigterm_exits_quickly(self) -> None:
-        port = "19090"
         env = os.environ.copy()
         env["STRATEGY_ID"] = "sigterm-test"
-        env["PORT"] = port
+        env["ALLOW_LIVE_SUBMIT"] = "0"
         env["HOST"] = "127.0.0.1"
 
         proc = subprocess.Popen(
-            [sys.executable, "-m", "src.server"],
+            [sys.executable, '-c', """
+from unittest.mock import patch
+import src.server as server
+class OfflineServer:
+    def __init__(self, *args): pass
+    def serve_forever(self, **kwargs): server._shutdown.wait(10)
+    def shutdown(self): pass
+    def server_close(self): pass
+with patch.object(server, 'ThreadingHTTPServer', OfflineServer):
+    server.main()
+"""],
             cwd=str(ROOT),
             env=env,
             stdout=subprocess.DEVNULL,
             stderr=subprocess.PIPE,
         )
         try:
-            # wait until health is up
+            # Read readiness from stderr; no sockets or network calls.
-            url = f"http://127.0.0.1:{port}/health"
+            import selectors
-            deadline = time.time() + 5
+            selector = selectors.DefaultSelector()
-            while time.time() < deadline:
+            self.addCleanup(selector.close)
-                try:
+            selector.register(proc.stderr, selectors.EVENT_READ)
-                    with urllib.request.urlopen(url, timeout=0.5) as resp:
+            self.assertTrue(selector.select(timeout=5), 'server readiness timeout')
-                        data = json.loads(resp.read().decode("utf-8"))
+            self.assertIn(b'Listening on', proc.stderr.readline())
-                        if resp.status == 200 and data.get("status") == "ok":
-                            break
-                except Exception:
-                    time.sleep(0.1)
-            else:
-                if proc.poll() is not None:
-                    self.fail(f"server exited before health check: {proc.stderr.read().decode(errors='replace')}")
-                self.fail("server did not become healthy in time")
 
             t0 = time.time()
             proc.send_signal(signal.SIGTERM)