diff --git a/README.md b/README.md index 818e84d..334a048 100644 --- a/README.md +++ b/README.md @@ -23,9 +23,8 @@ Victor 的确认已内置,`LIVE_CONFIRM` 仍接受但不再必需。每次 tas `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=30` 次;预检失败仍由 task/add 判定。用 `python3 scripts/sync_reject_marks.py ` 合并永久拒绝日志到 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。 diff --git a/docs/platform-contract-agent.md b/docs/platform-contract-agent.md index ae4ed49..775f13f 100644 --- a/docs/platform-contract-agent.md +++ b/docs/platform-contract-agent.md @@ -62,8 +62,9 @@ Victor 确认已内置,`LIVE_CONFIRM` 仍接受但不必需。 ASYNC_QUEUE_DEPTH / QUEUE_SNAPSHOT_JSON 仅为额外阻塞提示。 每周期最多 `LIVE_MAX_SUBMITS_PER_CYCLE`(默认 3)次;周期 `LIVE_CYCLE_SECONDS` 默认 3600、最小 60 秒,通过 shutdown Event 等待。 -候选 A verified_reuse 优先,禁提过滤、GPU enum、占位符检查及请求前持久化去重。 -`LIVE_STATE_FILE` 默认 /tmp/live-submitted.json;保留该文件可跨周期/重启去重。 +候选 A verified_reuse 优先,禁提过滤、GPU enum、占位符检查及远端、内置 ledger 和进程内去重。 +永久拒绝对保存在 `src/data/reject-marks.json`,随镜像重建保留;不写容器状态文件。 +`LIVE_PRECHECK` 默认 1,只读预检已验证/验证中 GPU;`LIVE_MAX_PRECHECKS_PER_CYCLE` 默认 30。 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=<安全原因>`。 diff --git a/scripts/sync_reject_marks.py b/scripts/sync_reject_marks.py new file mode 100644 index 0000000..37ca4a8 --- /dev/null +++ b/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() diff --git a/src/data/reject-marks.json b/src/data/reject-marks.json new file mode 100644 index 0000000..6e96bab --- /dev/null +++ b/src/data/reject-marks.json @@ -0,0 +1,61 @@ +{ + "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" + } + ] +} diff --git a/src/ledger.py b/src/ledger.py index af7a237..3a9e47b 100644 --- a/src/ledger.py +++ b/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)) diff --git a/src/live_cycle.py b/src/live_cycle.py index c38dd94..6ee4719 100644 --- a/src/live_cycle.py +++ b/src/live_cycle.py @@ -10,47 +10,21 @@ from .gate import evaluate_gate, queue_threshold 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, token_probe_status -from .ledger import load_baked, load_persisted, persist, rebuild +from .ledger import load_baked +from .marks import load_marks, REJECT_CODES, PERMANENT_PAIR_CODES +from .precheck import precheck_model, SAFE_STATUSES -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', -} -_DONE_CODES = {60027, 60028} _submitted = set() -_rejected = set() +_marked = {} _cycle_lock = threading.Lock() -def run_cycle(candidates=None, client=None, queue_reader=None, now=None, stats_reader=None): +def run_cycle(candidates=None, client=None, queue_reader=None, now=None, stats_reader=None, precheck=None): with _cycle_lock: - return _run_cycle(candidates, client, queue_reader, now, stats_reader) + return _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck) -def _run_cycle(candidates, client, queue_reader, now, stats_reader): +def _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck): decisions = [] def emit(action, reason, **fields): @@ -120,14 +94,51 @@ def _run_cycle(candidates, client, queue_reader, now, stats_reader): 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 | _rejected + 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', '30')) + if precheck_cap < 1: + raise ValueError + except ValueError: + precheck_cap = 30 + 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 @@ -151,11 +162,35 @@ def _run_cycle(candidates, client, queue_reader, now, stats_reader): if counted >= cap: emit('hold', 'cycle_cap') break + mapping = {} 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) + 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() @@ -165,7 +200,7 @@ def _run_cycle(candidates, client, queue_reader, now, stats_reader): 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 @@ -184,13 +219,7 @@ def _run_cycle(candidates, client, queue_reader, now, stats_reader): 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) @@ -203,15 +232,10 @@ def _run_cycle(candidates, client, queue_reader, now, stats_reader): 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) - if code not in _DONE_CODES: - _rejected.add(pair) - persisted.discard(pair) - _submitted.discard(pair) - try: - persist(state, persisted) - except OSError: - emit('hold', 'state_write_failed') - break + _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 diff --git a/src/marks.py b/src/marks.py new file mode 100644 index 0000000..d1216d4 --- /dev/null +++ b/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 diff --git a/src/precheck.py b/src/precheck.py new file mode 100644 index 0000000..f9df732 --- /dev/null +++ b/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 diff --git a/src/server.py b/src/server.py index 4af7462..91b05c7 100644 --- a/src/server.py +++ b/src/server.py @@ -59,11 +59,10 @@ 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: diff --git a/tests/test_live_cycle.py b/tests/test_live_cycle.py index b563d16..99964df 100644 --- a/tests/test_live_cycle.py +++ b/tests/test_live_cycle.py @@ -1,11 +1,8 @@ import io -import json -from pathlib import Path import os -import tempfile import unittest from unittest.mock import Mock, patch -from src.live_cycle import run_cycle, _submitted, _rejected +from src.live_cycle import run_cycle, _submitted, _marked def candidate(name, section='B', foothold='unknown'): @@ -14,13 +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() - _rejected.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')) @@ -34,10 +29,10 @@ 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 state_pairs(self): - return json.loads(Path(os.environ['LIVE_STATE_FILE']).read_text()) + def submitted_pairs(self): + return sorted(m + '|' + g for m, g in _submitted) - def test_done_rejection_keeps_reservation_and_continues(self): + def test_done_rejection_marks_pair_and_continues(self): for code in (60027, 60028): with self.subTest(code=code): self.client.submit.side_effect = [ @@ -46,14 +41,14 @@ class TestLiveCycle(unittest.TestCase): 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.assertIn('org/' + str(code) + '|Biren_166m', self.state_pairs()) + 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_removed_but_not_retried_in_process(self): + 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.state_pairs(), []) + self.assertEqual(self.submitted_pairs(), []) self.cycle(counts=()) self.client.submit.assert_called_once() @@ -62,12 +57,12 @@ class TestLiveCycle(unittest.TestCase): decisions = self.cycle() self.assertEqual(decisions[0]['action'], 'rejected') self.assertEqual(decisions[0]['reason'], 'unknown') - self.assertEqual(self.state_pairs(), []) + self.assertEqual(self.submitted_pairs(), []) def test_attempt_cap_and_invalid_fallback(self): for value, expected in ((None, 15), ('bad', 15), ('0', 15), ('-1', 15), ('2', 2)): with self.subTest(value=value): - _rejected.clear() + _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}): @@ -87,17 +82,16 @@ class TestLiveCycle(unittest.TestCase): 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.state_pairs(), []) + 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() - Path(os.environ['LIVE_STATE_FILE']).unlink(missing_ok=True) self.client.reset_mock() def submit(payload): - self.assertIn('org/a|Biren_166m', self.state_pairs()) + self.assertIn('org/a|Biren_166m', self.submitted_pairs()) if isinstance(result, Exception): raise result return result @@ -134,10 +128,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()) @@ -160,7 +153,7 @@ class TestLiveCycle(unittest.TestCase): 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_rejected_token_probe_does_not_leak(self): + def test_marked_token_probe_does_not_leak(self): with patch.dict(os.environ, {'XC_TOKEN': ''}): self.cycle() self.client.submit.assert_not_called() @@ -184,7 +177,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() diff --git a/tests/test_live_ledger.py b/tests/test_live_ledger.py index aa754c0..fb1451c 100644 --- a/tests/test_live_ledger.py +++ b/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() @@ -67,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) diff --git a/tests/test_reject_marks.py b/tests/test_reject_marks.py new file mode 100644 index 0000000..82a33e3 --- /dev/null +++ b/tests/test_reject_marks.py @@ -0,0 +1,258 @@ +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), 8) + 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_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(32)] + for value, count in (('2', 2), ('bad', 30), ('0', 30)): + _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')