diff --git a/README.md b/README.md index cfcd504..818e84d 100644 --- a/README.md +++ b/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 的有效数字。 diff --git a/docs/gpu-perf-order.md b/docs/gpu-perf-order.md index c42a34e..4bba425 100644 --- a/docs/gpu-perf-order.md +++ b/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。 diff --git a/docs/platform-contract-agent.md b/docs/platform-contract-agent.md index 2a978c7..ae4ed49 100644 --- a/docs/platform-contract-agent.md +++ b/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=<安全原因>`。 diff --git a/src/health.py b/src/health.py index 8f0fee7..f82cdfc 100644 --- a/src/health.py +++ b/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(), } diff --git a/src/live_cycle.py b/src/live_cycle.py index 9371111..c38dd94 100644 --- a/src/live_cycle.py +++ b/src/live_cycle.py @@ -7,7 +7,7 @@ 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 @@ -45,12 +45,12 @@ _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): @@ -72,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: diff --git a/src/live_queue.py b/src/live_queue.py index 55242e0..b660626 100644 --- a/src/live_queue.py +++ b/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] diff --git a/tests/test_health.py b/tests/test_health.py index 35c88f6..84cc561 100644 --- a/tests/test_health.py +++ b/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) diff --git a/tests/test_live_cycle.py b/tests/test_live_cycle.py index c60863d..b563d16 100644 --- a/tests/test_live_cycle.py +++ b/tests/test_live_cycle.py @@ -194,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) diff --git a/tests/test_live_queue.py b/tests/test_live_queue.py index c69a0ab..93d4a3f 100644 --- a/tests/test_live_queue.py +++ b/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()))