live: baked reject-marks (8) + read-only search-by-model-id precheck (已验证→60027/验证中→60028 skip, no submit); in-run marks; drop /tmp state; sync_reject_marks.py (95 tests)

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` 默认
+禁提、重复、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`。
-`/tmp/live-submitted.json`，格式为 `model|gpuEnum` 字符串列表，在请求前持久化；失败请求也不自动重试。旧版仅模型名状态文件须先离线迁移，否则 fail closed。
+只读平台预检默认开启（`LIVE_PRECHECK=0` 关闭），每周期最多 `LIVE_MAX_PRECHECKS_PER_CYCLE=30` 次；预检失败仍由 task/add 判定。用 `python3 scripts/sync_reject_marks.py <log.txt>` 合并永久拒绝日志到 baked marks，重建镜像后保留；支持 `--dry-run`、`--marks` 和 `--date`。
-去重合并远端全部状态的 `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。

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、占位符检查及请求前持久化去重。
+候选 A verified_reuse 优先，禁提过滤、GPU enum、占位符检查及远端、内置 ledger 和进程内去重。
-`LIVE_STATE_FILE` 默认 /tmp/live-submitted.json；保留该文件可跨周期/重启去重。
+永久拒绝对保存在 `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=<安全原因>`。

scripts/sync_reject_marks.py

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

src/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"
+    }
+  ]
+}

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):
+def rebuild(reader=None):
-    rows = json.loads(path.read_text()) if path.exists() else []
+    from .marks import load_marks
-    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)
+    marks = load_marks()
-    return count, remote | baked | persisted, blocked, (len(remote), len(baked), len(persisted))
+    return count, remote | baked, blocked, (len(remote), len(baked), len(marks))
-
-
-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

@@ -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,8 +162,32 @@ 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
@@ -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:
+            known.discard(pair)
-                    persist(state, persisted)
+            if code in PERMANENT_PAIR_CODES or code != 60007:
-                except OSError:
+                _marked[pair] = code
-                    emit('hold', 'state_write_failed')
-                    break
             if code == 60007:
                 emit('hold', 'task_num_limit')
                 break

src/marks.py

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

src/precheck.py

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

src/server.py

@@ -59,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')))
+        _, _, _, sizes = rebuild()
-        sys.stderr.write('[live] ledger rebuilt remote=%d baked=%d persisted=%d\n' % sizes)
+        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:

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.env = patch.dict(os.environ, {'LIVE_PRECHECK': '0', 'STRATEGY_ID': 'sid', 'CONFIG_PARAMS_SOURCE': 'sample', 'XC_TOKEN': 'synthetic-secret'}, clear=True)
-        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()
-        _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):
+    def submitted_pairs(self):
-        return json.loads(Path(os.environ['LIVE_STATE_FILE']).read_text())
+        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': '<private-secret-fragment>'}):
             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()
 

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()
+        env = patch.dict(os.environ, dict(LIVE_PRECHECK='0', STRATEGY_ID='sid', XC_TOKEN='synthetic-credential', CONFIG_PARAMS_SOURCE='sample'), clear=True)
-        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()
+        _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'])
+        self.assertIn(('unit/model', 'Iluvatar_bi-100'), _submitted)
-        _submitted.clear()
         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)

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