live: success = verify_pass (verifyResult==1); per-cycle [live] stats line; /status stats + success_metric (76 tests)

README.md

@@ -2,7 +2,7 @@
 
 ModelHub XC（信创模盒）**适配智能体** Phase1 本地骨架。
 
-> **不代注册；Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自成功适配，数字一律 **待平台核实**。
+> **不代注册；Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自验证通过的适配（`verify_pass` / `verifyResult==1`），数字一律 **待平台核实**。
 > 冲突裁决：**适配 W1 优先**；本仓默认 live 开启，仅在实时验证队列低于阈值时提交。
 
 ## 契约要点
@@ -84,3 +84,7 @@ tests/      # stdlib unittest
 
 本仓库用于 Victor 本地/Gitea 交付。Victor 已批准上述队列闸门自动提交；不自动注册。
 `make dry-run-pending` 仍仅生成草稿；测试全部 mock 平台接口。
+
+成功指标统一为 `verify_pass`（`verifyResult==1`，验证通过）；任务 status 为 `success` 但 `verifyResult==-1` 属于验证失败，积分为 0。
+
+每轮 live 在凭证解析后记录一次 `[live] stats total=N queue=N verify_pass=N verify_fail=N failed=N unknown=N points=X`；读取失败则记录 `[live] stats unavailable reason=<安全原因>`，提交仍由原队列闸门决定。`GET /status` 保留凭证存在布尔值，新增 `success_metric: "verify_pass"` 和 `stats`：最近成功读取的上述数字及 `updated_at`（epoch 秒），尚未读取时为 null。queue 按未终结状态计数；积分仅累计 verify_pass 的有效数字。

docs/gpu-perf-order.md

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

docs/platform-contract-agent.md

@@ -23,14 +23,14 @@
 | Dockerfile | **仓库根目录**；base `modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim` | A |
 | 端口 | **8080**，`EXPOSE 8080` | A |
 | 健康检查 | `GET /health` → 200 `{"status":"ok"}` | A |
-| 可选状态 | `GET /status` → 布尔 `strategy_id_present` / `external_service_token_present`（**不回传值**） | 本仓 Phase1 |
+| 可选状态 | `GET /status` → 布尔 `strategy_id_present` / `external_service_token_present`（**不回传值**）；另含 `success_metric: "verify_pass"` 与 `stats`（最近统计及 epoch 秒 `updated_at`，未读取时 null） | 本仓 Phase1 |
 | 优雅退出 | `SIGTERM`，约 **≤30s** 平台窗口；本实现目标 ~1s 退出主循环 | B |
 | 平台注入 | `STRATEGY_ID`；`EXTERNAL_SERVICE_TOKEN`（及社区别名） | A/B |
 | **STRATEGY_ID 缺失** | **Phase1：进程启动失败（exit ≠0）** | 产品规划 Phase1 + PRD |
 | 资源 | 约 100m–1 CPU / 256–512Mi | B |
 | **发布智能体得分** | **0 分**（发布 ≠ 得分） | B |
 | Live 默认 | **开**（Victor 已批准）；`ALLOW_LIVE_SUBMIT=0` 停交，`LIVE_CONFIRM` 可选 | 本仓 |
-| 积分 | 来自**成功适配**；智能体代提成功计同一成功分；公式/周报 **待平台核实** | B / D |
+| 积分 | 来自**验证通过**（`verify_pass`，`verifyResult==1`）的适配；status=`success` 且 verifyResult=-1 为验证失败、0 分；公式/周报 **待平台核实** | B / D |
 
 ## 鉴权双轨（文档索引，Phase1 不代调）
 
@@ -65,3 +65,5 @@ ASYNC_QUEUE_DEPTH / QUEUE_SNAPSHOT_JSON 仅为额外阻塞提示。
 候选 A verified_reuse 优先，禁提过滤、GPU enum、占位符检查及请求前持久化去重。
 `LIVE_STATE_FILE` 默认 /tmp/live-submitted.json；保留该文件可跨周期/重启去重。
 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=<安全原因>`。

src/health.py

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

src/live_cycle.py

@@ -7,21 +7,50 @@ from pathlib import Path
 from .banned import banned_rule
 from .config_params import GPU_ENUMS, build_config_params_with_meta
 from .gate import evaluate_gate, queue_threshold
-from .live_queue import read_queue, model_key
+from .live_queue import read_queue, read_queue_stats, model_key
 from .pending_pool import build_ordered_candidates
 from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source, token_probe_status
 from .ledger import load_baked, load_persisted, persist, rebuild
 
+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()
 _cycle_lock = threading.Lock()
 
 
-def run_cycle(candidates=None, client=None, queue_reader=None, now=None):
+def run_cycle(candidates=None, client=None, queue_reader=None, now=None, stats_reader=None):
     with _cycle_lock:
-        return _run_cycle(candidates, client, queue_reader, now)
+        return _run_cycle(candidates, client, queue_reader, now, stats_reader)
 
 
-def _run_cycle(candidates, client, queue_reader, now):
+def _run_cycle(candidates, client, queue_reader, now, stats_reader):
     decisions = []
 
     def emit(action, reason, **fields):
@@ -43,6 +72,28 @@ def _run_cycle(candidates, client, queue_reader, now):
         emit('hold', 'missing/invalid_token')
         return decisions
     print('[live] token source=' + source, file=sys.stderr)
+    # An injected queue reader keeps existing offline callers offline.
+    reason, stats = 'not_read', None
+    if stats_reader is not None or queue_reader is None:
+        try:
+            _, reason, _, stats = (stats_reader or read_queue_stats)()
+        except Exception:
+            reason, stats = 'read_failed', None
+    if stats is not None:
+        try:
+            line = ' '.join(f'{key}={stats[key]:g}' if key == 'points'
+                            else f'{key}={int(stats[key])}' for key in
+                            ('total', 'queue', 'verify_pass', 'verify_fail', 'failed', 'unknown', 'points'))
+        except (KeyError, TypeError, ValueError, OverflowError):
+            stats, reason = None, 'invalid_stats'
+    if stats is None:
+        # Only known reasons may enter logs; injected errors may contain secrets.
+        safe_reasons = {'not_read', 'read_failed', 'invalid_stats', 'missing/invalid_token',
+                        'http_error', 'malformed_json', 'transport/http_error', 'platform_error',
+                        'malformed_data', 'malformed_records', 'pagination_total_changed',
+                        'pagination_inconsistent', 'incomplete_pagination'}
+        line = 'unavailable reason=' + (reason if isinstance(reason, str) and reason in safe_reasons else 'read_failed')
+    print('[live] stats ' + line, file=sys.stderr)
     try:
         cap = int(os.environ.get('LIVE_MAX_SUBMITS_PER_CYCLE', '3'))
         if cap < 1:
@@ -50,6 +101,12 @@ def _run_cycle(candidates, client, queue_reader, now):
     except ValueError:
         emit('hold', 'invalid_cycle_cap')
         return decisions
+    try:
+        attempt_cap = int(os.environ.get('LIVE_MAX_ATTEMPTS_PER_CYCLE', '15'))
+        if attempt_cap < 1:
+            raise ValueError
+    except ValueError:
+        attempt_cap = 15
     if candidates is None:
         try:
             if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
@@ -67,13 +124,14 @@ def _run_cycle(candidates, client, queue_reader, now):
     try:
         persisted = load_persisted(state) | _submitted
         baked, blocked = load_baked()
-        known = persisted | baked
+        known = persisted | baked | _rejected
     except (OSError, ValueError, KeyError, TypeError):
         emit('hold', 'state_read_failed')
         return decisions
     reader = queue_reader or read_queue
     client = client if client is not None else TaskAddClient()
     attempts = 0
+    counted = 0
     for candidate in candidates:
         model = candidate.get('model_address') or candidate.get('modelAddress') or ''
         rule = banned_rule(model + ' ' + str(candidate.get('model_name', candidate.get('modelName', ''))))
@@ -90,7 +148,7 @@ def _run_cycle(candidates, client, queue_reader, now):
         if candidate.get('skip_reason'):
             emit('skip', 'pending_filter', model=model)
             continue
-        if attempts >= cap:
+        if counted >= cap:
             emit('hold', 'cycle_cap')
             break
         choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')]
@@ -142,6 +200,29 @@ def _run_cycle(candidates, client, queue_reader, now):
             reason = 'task_add_response' if status is not None else 'transport_error'
         except Exception:
             status, code, reason = None, None, 'submit_error'
+        if code is not None and code != 0:
+            emit('rejected', REJECT_CODES.get(code, 'unknown'), model=model,
+                 gpu=GPU_ENUMS[gpu], code=code, http=status, queue_before=count)
+            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
+            if code == 60007:
+                emit('hold', 'task_num_limit')
+                break
+            if attempts >= attempt_cap:
+                emit('hold', 'attempt_cap')
+                break
+            continue
+        counted += 1
         emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
              http=status, code=code, **meta)
+        if attempts >= attempt_cap:
+            emit('hold', 'attempt_cap')
+            break
     return decisions

src/live_queue.py

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

tests/test_health.py

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

tests/test_live_cycle.py

@@ -1,9 +1,11 @@
 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
+from src.live_cycle import run_cycle, _submitted, _rejected
 
 
 def candidate(name, section='B', foothold='unknown'):
@@ -18,6 +20,7 @@ class TestLiveCycle(unittest.TestCase):
         self.env.start()
         self.addCleanup(self.env.stop)
         _submitted.clear()
+        _rejected.clear()
         self.client = Mock()
         self.client.submit.return_value = dict(http_status=200, platform_code=0)
         network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
@@ -31,6 +34,80 @@ 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 test_done_rejection_keeps_reservation_and_continues(self):
+        for code in (60027, 60028):
+            with self.subTest(code=code):
+                self.client.submit.side_effect = [
+                    dict(http_status=200, platform_code=code),
+                    dict(http_status=200, platform_code=0)]
+                with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
+                    decisions = self.cycle([candidate(str(code)), candidate('ok' + str(code))], (1, 1))
+                self.assertEqual([d['action'] for d in decisions], ['rejected', 'submitted'])
+                self.assertIn('org/' + str(code) + '|Biren_166m', self.state_pairs())
+        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):
+        self.client.submit.return_value = dict(http_status=200, platform_code=60020)
+        self.cycle()
+        self.assertEqual(self.state_pairs(), [])
+        self.cycle(counts=())
+        self.client.submit.assert_called_once()
+
+    def test_unknown_rejection(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=99999)
+        decisions = self.cycle()
+        self.assertEqual(decisions[0]['action'], 'rejected')
+        self.assertEqual(decisions[0]['reason'], 'unknown')
+        self.assertEqual(self.state_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()
+                self.client.reset_mock()
+                self.client.submit.return_value = dict(http_status=200, platform_code=60020)
+                with patch.dict(os.environ, {} if value is None else {'LIVE_MAX_ATTEMPTS_PER_CYCLE': value}):
+                    decisions = self.cycle([candidate(str(i)) for i in range(expected)], (1,) * expected)
+                self.assertEqual(self.client.submit.call_count, expected)
+                self.assertEqual(decisions[-1]['reason'], 'attempt_cap')
+
+    def test_submit_cap_counts_successes_after_rejections(self):
+        self.client.submit.side_effect = [dict(http_status=200, platform_code=c) for c in (60020, 0, 60028, 0, 0)]
+        decisions = self.cycle([candidate(str(i)) for i in range(7)], (1,) * 5)
+        self.assertEqual(self.client.submit.call_count, 5)
+        self.assertEqual(sum(d['action'] == 'submitted' for d in decisions), 3)
+        self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
+
+    def test_task_limit_stops_cycle(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=60007)
+        decisions = self.cycle([candidate('a'), candidate('b')])
+        self.assertEqual([d['action'] for d in decisions], ['rejected', 'hold'])
+        self.assertEqual(decisions[-1]['reason'], 'task_num_limit')
+        self.assertEqual(self.state_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())
+                    if isinstance(result, Exception):
+                        raise result
+                    return result
+                self.client.submit.side_effect = submit
+                with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
+                    decisions = self.cycle([candidate('a'), candidate('b')])
+                self.assertEqual(decisions[0]['action'], 'submitted')
+                self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
+                self.client.submit.assert_called_once()
+
     def test_high_and_failed_queue(self):
         for n in (95, None):
             decisions = self.cycle(counts=(n,))
@@ -117,3 +194,31 @@ class TestLiveCycle(unittest.TestCase):
         self.client.submit.assert_not_called()
         self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
         self.assertIn('placeholder', self.logs.getvalue())
+
+    def test_stats_logged_once_and_submission_continues(self):
+        from src.live_queue import compute_stats
+        stats = compute_stats([dict(status='success', verifyResult=1, stageAndNoviceTaskCredit='5.5')])
+        stats_reader = Mock(return_value=(0, 'ok', set(), stats))
+        run_cycle([candidate('stats')], self.client, lambda: (0, 'ok', set()), stats_reader=stats_reader)
+        stats_reader.assert_called_once_with()
+        lines = [s for s in self.logs.getvalue().splitlines() if s.startswith('[live] stats ')]
+        self.assertEqual(lines, ['[live] stats total=1 queue=0 verify_pass=1 verify_fail=0 failed=0 unknown=0 points=5.5'])
+        self.assertNotIn('synthetic-secret', self.logs.getvalue())
+        self.client.submit.assert_called_once()
+
+    def test_stats_failure_is_safe_and_does_not_gate(self):
+        for index, reader in enumerate((Mock(return_value=(None, 'platform_error', set(), None)),
+                       Mock(side_effect=OSError('synthetic-secret')),
+                       Mock(return_value=(None, 'synthetic-secret', set(), None)))):
+            self.logs.truncate(0)
+            self.logs.seek(0)
+            run_cycle([candidate('stats-failure-' + str(index))], self.client, lambda: (0, 'ok', set()), stats_reader=reader)
+            lines = [s for s in self.logs.getvalue().splitlines() if s.startswith('[live] stats ')]
+            self.assertEqual(len(lines), 1)
+            self.assertTrue(lines[0].startswith('[live] stats unavailable reason='))
+            self.assertNotIn('synthetic-secret', self.logs.getvalue())
+        self.assertEqual(self.client.submit.call_count, 3)
+
+    def test_injected_queue_skips_stats_network(self):
+        self.cycle()
+        self.assertEqual(self.logs.getvalue().count('[live] stats unavailable reason=not_read'), 1)

tests/test_live_queue.py

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