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)

This commit is contained in:
Victor Cao
2026-10-17 06:53:56 +08:00
parent 3adf13ad22
commit 101be63660
20 changed files with 5029 additions and 129 deletions

View File

@@ -12,7 +12,7 @@ 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)"

View File

@@ -20,9 +20,12 @@ Victor 的确认已内置,`LIVE_CONFIRM` 仍接受但不再必需。每次 tas
`LIVE_QUEUE_THRESHOLD=95`(严格小于),`LIVE_MAX_SUBMITS_PER_CYCLE=3`,
`LIVE_CYCLE_SECONDS=600`(最小 60 秒)。环境队列提示仅可额外阻塞。
候选来自 `PENDING_SUBMIT_MD` / `PENDING_SUBMIT_TXT`,否则使用
`src/data/pending-submit.md`(若存在);A verified_reuse → 其他 A → B。
禁提、重复、GPU 未映射、占位符候选跳过。`LIVE_STATE_FILE` 默认
`/tmp/live-submitted.json`,在请求前持久化去重;失败请求也不自动重试。
`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。
## 快速开始

View File

@@ -19,8 +19,8 @@
|-----------------|------------|------------------------------|------|
| 1(一梯队并列) | 壁砺166M | `Biren_166m` | ≈ 曦望 S2;社区成功量壁砺更高 |
| 1(一梯队并列) | 曦望 S2 | `Sunrise_pt-200-x1`(抓包证实) | 与壁砺并列,勿臆造先后「性能」 |
| 2 | 天垓150 | Iluvatar 天垓族(字符串待核) | 条件回退;饱和+唯一性碰撞多 |
| 3 | 天垓100 | (待核) | 一般不扇出 |
| 2 | 天垓150 | Iluvatar_bi-150 | 条件回退;饱和+唯一性碰撞多 |
| 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)。

View File

@@ -1,27 +1,11 @@
# Queue-gated live submit
# Live gate update — 2026-10-17
Victor-approved live defaults on. ALLOW_LIVE_SUBMIT=0 stops submission;
STRATEGY_ID remains mandatory. LIVE_CONFIRM is accepted but optional.
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.
Each add requires a fresh authenticated onlyMine task/page snapshot, complete
through data.total (maximum 20 pages). All nonterminal records count;
queue >= LIVE_QUEUE_THRESHOLD (default 95) stops the cycle. Authentication,
transport, HTTP, platform, JSON and pagination errors fail closed. Environment
queue hints can only add blockers.
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.
Cycles prioritize Section A verified_reuse, other A, then B; reject banned
families, unmapped GPUs and placeholders; use build-config with recorded sample
fallback; reserve model identity in memory and LIVE_STATE_FILE before calling
TaskAddClient. Remote task identities also block duplicates. Default cap is 3
attempts per cycle. Failed calls retain their reservations to prevent retries.
The state file defaults to /tmp/live-submitted.json; preserve or mount it for
restart durability beyond ephemeral container storage.
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.
Server runs a daemon worker every 600 seconds (minimum 60), using the shutdown
Event for interruptible waits. Missing credentials hold instead of crashing.
Docker FROM is unchanged; templates are copied for runtime fallback.
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 54 tests, including mocked platform calls and local
SIGTERM integration. Sandbox initially denied local sockets; the suite passed
with local socket permission. No real ModelHub calls, token writes, commits,
tags or pushes were performed.
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.

View File

@@ -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()

39
scripts/bake_ledger.py Normal file
View File

@@ -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()

View File

@@ -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',
}

3661
src/data/candidates.json Normal file

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

35
src/ledger.py Normal file
View File

@@ -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)

View File

@@ -9,7 +9,8 @@ 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
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()
@@ -32,6 +33,12 @@ def _run_cycle(candidates, client, queue_reader, now):
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:
@@ -40,11 +47,11 @@ def _run_cycle(candidates, client, queue_reader, now):
emit('hold', 'invalid_cycle_cap')
return decisions
if candidates is None:
fallback = Path(__file__).parent / 'data/pending-submit.md'
try:
candidates = build_ordered_candidates(
os.environ.get('PENDING_SUBMIT_MD', str(fallback)),
os.environ.get('PENDING_SUBMIT_TXT', str(fallback.with_suffix('.txt'))))
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
@@ -54,11 +61,10 @@ def _run_cycle(candidates, client, queue_reader, now):
return decisions
state = Path(os.environ.get('LIVE_STATE_FILE', '/tmp/live-submitted.json'))
try:
saved = json.loads(state.read_text()) if state.exists() else []
if not isinstance(saved, list) or any(not isinstance(s, str) for s in saved):
raise ValueError
known = set(map(model_key, saved)) | _submitted
except (OSError, ValueError):
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
@@ -70,9 +76,12 @@ def _run_cycle(candidates, client, queue_reader, now):
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 known:
emit('skip', 'duplicate', model=model)
if key in blocked:
emit('skip', 'do-not-resubmit', model=model)
continue
if candidate.get('skip_reason'):
emit('skip', 'pending_filter', model=model)
@@ -80,10 +89,27 @@ def _run_cycle(candidates, client, queue_reader, now):
if attempts >= cap:
emit('hold', 'cycle_cap')
break
gpu = next(iter(candidate.get('try_next_gpus') or []), candidate.get('targetGpu'))
if gpu not in GPU_ENUMS:
emit('skip', 'gpu_unmapped', model=model)
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)
@@ -94,32 +120,12 @@ def _run_cycle(candidates, client, queue_reader, now):
if any(not value or '<' in str(value) or '>' in str(value) for value in payload.values()):
emit('skip', 'placeholder', model=model)
continue
# Fresh complete snapshot immediately before reserving and sending each add.
pair = (key, GPU_ENUMS[gpu])
known.add(pair)
persisted.add(pair)
_submitted.add(pair)
try:
count, why, remote = reader()
except Exception:
count, why, remote = None, 'reader_error', set()
if count is None:
# Only internal fixed reasons are emitted; injected errors can contain secrets.
safe = why if why in {'missing/invalid_token', 'http_error', 'malformed_json', 'transport/http_error', 'platform_error', 'malformed_data', 'malformed_records', 'pagination_total_changed', 'pagination_inconsistent', 'incomplete_pagination'} else 'reader_error'
emit('hold', 'queue_read_failed:' + safe)
break
if type(count) is not int or count < 0:
emit('hold', 'queue_read_failed:invalid_count')
break
if count >= queue_threshold():
emit('hold', f'queue>={queue_threshold()}', count=count)
break
if key in {model_key(m) for m in remote}:
emit('skip', 'duplicate', model=model)
continue
known.add(key)
_submitted.add(key)
try:
state.parent.mkdir(parents=True, exist_ok=True)
temporary = state.with_name(state.name + '.tmp')
temporary.write_text(json.dumps(sorted(known)))
temporary.replace(state)
persist(state, persisted)
except OSError:
emit('hold', 'state_write_failed')
break

View File

@@ -15,6 +15,10 @@ def model_key(value):
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:
@@ -39,8 +43,9 @@ def read_queue():
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 type(data.get('total')) is not int or data['total'] < 0:
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()
@@ -52,14 +57,12 @@ def read_queue():
return None, 'pagination_inconsistent', set()
for record in records:
count += str(record.get('status', '')).strip().lower() not in TERMINAL
for field in ('modelAddress', 'model_address'):
if record.get(field):
models.add(model_key(record[field]))
if record.get('modelAuthorName') and record.get('modelId'):
models.add(model_key(f"{record['modelAuthorName']}/{record['modelId']}"))
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 token.lower() not in m}
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()

View File

@@ -59,6 +59,13 @@ 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)
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:

View File

@@ -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:
raw = os.environ.get(key)
if raw is None:
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
from .task_add_client import resolve_token_with_source
try:
resolve_token_with_source()
return True
except (ValueError, OSError, UnicodeError):
return False

View File

@@ -19,13 +19,22 @@ def build_payload(model_address, target_gpu='<TARGET_GPU>', framework='<FRAMEWOR
configParams=config_params)
def resolve_token():
file = os.environ.get('XC_TOKEN_FILE')
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):
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')
return token
def resolve_token():
return resolve_token_with_source()[0]
class TaskAddClient:

View File

@@ -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'])

View File

@@ -14,7 +14,7 @@ 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'}, clear=True)
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()
@@ -64,10 +64,10 @@ class TestLiveCycle(unittest.TestCase):
self.cycle([candidate('ok')], ())
self.client.submit.assert_called_once()
self.assertIn('banned:', self.logs.getvalue())
self.assertIn('duplicate', 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'}))
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()
@@ -76,6 +76,7 @@ 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': ''}):
run_cycle([candidate('a')], self.client)
self.client.submit.assert_not_called()
self.assertIn('missing/invalid_token', self.logs.getvalue())
@@ -105,5 +106,5 @@ class TestLiveCycle(unittest.TestCase):
rows = [dict(model_address='org/unmapped', targetGpu='unknown'), candidate('<MODEL>')]
self.cycle(rows, ())
self.client.submit.assert_not_called()
self.assertIn('gpu_unmapped', self.logs.getvalue())
self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
self.assertIn('placeholder', self.logs.getvalue())

71
tests/test_live_ledger.py Normal file
View File

@@ -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)

View File

@@ -14,17 +14,17 @@ class TestLiveQueue(unittest.TestCase):
return r
def test_pagination_and_status(self):
pages = [self.response({'code': 0, 'data': {'total': 3, 'records': [
{'status': 'waiting', 'modelAddress': 'org/a'}, {'status': 'success', 'verifyResult': -1}]}}),
self.response({'code': 0, 'data': {'total': 3, 'records': [{'status': '等待中'}]}})]
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', models)
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': 40100}, {'code': 40101}, {'code': 0, 'data': {'total': 1, 'records': []}}, []):
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:

View File

@@ -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,36 +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["ALLOW_LIVE_SUBMIT"] = "0"
env["PORT"] = port
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
url = f"http://127.0.0.1:{port}/health"
deadline = time.time() + 5
while time.time() < deadline:
try:
with urllib.request.urlopen(url, timeout=0.5) as resp:
data = json.loads(resp.read().decode("utf-8"))
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")
# Read readiness from stderr; no sockets or network calls.
import selectors
selector = selectors.DefaultSelector()
self.addCleanup(selector.close)
selector.register(proc.stderr, selectors.EVENT_READ)
self.assertTrue(selector.select(timeout=5), 'server readiness timeout')
self.assertIn(b'Listening on', proc.stderr.readline())
t0 = time.time()
proc.send_signal(signal.SIGTERM)