forked from luopingyi/xc_agent_platform_demo
Compare commits
7 Commits
v0.2.4-liv
...
v0.2.9-liv
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c95b66bb40 | ||
|
|
9f265595e7 | ||
|
|
b0d66da881 | ||
|
|
78cfea508d | ||
|
|
2361e7e84f | ||
|
|
82f3af1cc2 | ||
|
|
79555c7db6 |
@@ -18,6 +18,7 @@ ENV PYTHONUNBUFFERED=1
|
|||||||
ENV EXTERNAL_SERVICE_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
|
ENV EXTERNAL_SERVICE_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
|
||||||
ENV XC_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
|
ENV XC_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
|
||||||
ENV XC_TOKEN_FILE=/app/secrets/xc_token
|
ENV XC_TOKEN_FILE=/app/secrets/xc_token
|
||||||
|
ENV ALL_TASKS_FOOTHOLD=1
|
||||||
|
|
||||||
EXPOSE 8080
|
EXPOSE 8080
|
||||||
|
|
||||||
|
|||||||
15
README.md
15
README.md
@@ -2,7 +2,7 @@
|
|||||||
|
|
||||||
ModelHub XC(信创模盒)**适配智能体** Phase1 本地骨架。
|
ModelHub XC(信创模盒)**适配智能体** Phase1 本地骨架。
|
||||||
|
|
||||||
> **不代注册;Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自成功适配,数字一律 **待平台核实**。
|
> **不代注册;Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自验证通过的适配(`verify_pass` / `verifyResult==1`),数字一律 **待平台核实**。
|
||||||
> 冲突裁决:**适配 W1 优先**;本仓默认 live 开启,仅在实时验证队列低于阈值时提交。
|
> 冲突裁决:**适配 W1 优先**;本仓默认 live 开启,仅在实时验证队列低于阈值时提交。
|
||||||
|
|
||||||
## 契约要点
|
## 契约要点
|
||||||
@@ -20,12 +20,11 @@ Tag `v0.2.3-live-token` 内置个人凭证,风险与停交说明见 [token bak
|
|||||||
Victor 的确认已内置,`LIVE_CONFIRM` 仍接受但不再必需。每次 task/add 前,
|
Victor 的确认已内置,`LIVE_CONFIRM` 仍接受但不再必需。每次 task/add 前,
|
||||||
通过 Xc-Token 全量分页读取 onlyMine task/page;非终态计入队列,读取失败停交。
|
通过 Xc-Token 全量分页读取 onlyMine task/page;非终态计入队列,读取失败停交。
|
||||||
`LIVE_QUEUE_THRESHOLD=95`(严格小于),`LIVE_MAX_SUBMITS_PER_CYCLE=3`,
|
`LIVE_QUEUE_THRESHOLD=95`(严格小于),`LIVE_MAX_SUBMITS_PER_CYCLE=3`,
|
||||||
`LIVE_CYCLE_SECONDS=600`(最小 60 秒)。环境队列提示仅可额外阻塞。
|
`LIVE_CYCLE_SECONDS=3600`(最小 60 秒)。环境队列提示仅可额外阻塞。
|
||||||
候选来自 `PENDING_SUBMIT_MD` / `PENDING_SUBMIT_TXT`,否则使用
|
候选来自 `PENDING_SUBMIT_MD` / `PENDING_SUBMIT_TXT`,否则使用
|
||||||
`src/data/candidates.json`(随镜像内置);A verified_reuse → 其他 A → B。
|
`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` 请求头,不记录凭证值。
|
凭证优先级:`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 个模型禁提项。
|
离线再生成:`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。
|
配置优先 build-config,失败使用 sample 并记录来源及原因。token 缺失只 hold。
|
||||||
@@ -84,3 +83,9 @@ tests/ # stdlib unittest
|
|||||||
|
|
||||||
本仓库用于 Victor 本地/Gitea 交付。Victor 已批准上述队列闸门自动提交;不自动注册。
|
本仓库用于 Victor 本地/Gitea 交付。Victor 已批准上述队列闸门自动提交;不自动注册。
|
||||||
`make dry-run-pending` 仍仅生成草稿;测试全部 mock 平台接口。
|
`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 的有效数字。
|
||||||
|
|
||||||
|
`ALL_TASKS_FOOTHOLD=1` 可启用全部任务落点(默认 `0`):只读取通过、失败、running 三桶,硬顶 3/3/2 页、每页 ≤100、单请求 ≤5 秒、总读取 ≤20 秒。任一桶错误整轮回退原候选,stderr 输出固定原因码与分页/截断计数。目标卡只扣除切片中已观察的卡;切片不保证全站覆盖、最新排序、已下载或零重复排队,仍保留原队列门控、marks 与 precheck。上限可用 `ALL_TASKS_MAX_PAGES_PASS/FAIL/RUNNING`、`ALL_TASKS_PAGE_SIZE` 调低,不能调高硬顶。
|
||||||
|
|||||||
@@ -50,3 +50,5 @@
|
|||||||
- API gpuType 仅映射壁砺/壁砺166M/Biren_166m 与曦望/曦望 S2/Sunrise_pt-200-x1;未知卡不请求 API,报告 GAP: gpu enum unmapped。
|
- 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)。
|
- 选策优化后仍为 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。
|
- **live 前必须有 token + API 调用实证**;sample 单独不是 production-ready。队列 ≥95 继续禁 live,本轮无 live。
|
||||||
|
|
||||||
|
成功量 / 成功样本的指标口径:成功指标统一为 `verify_pass`(`verifyResult==1`,验证通过);任务 status 为 `success` 但 `verifyResult==-1` 属于验证失败,积分为 0。
|
||||||
|
|||||||
@@ -23,14 +23,14 @@
|
|||||||
| Dockerfile | **仓库根目录**;base `modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim` | A |
|
| Dockerfile | **仓库根目录**;base `modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim` | A |
|
||||||
| 端口 | **8080**,`EXPOSE 8080` | A |
|
| 端口 | **8080**,`EXPOSE 8080` | A |
|
||||||
| 健康检查 | `GET /health` → 200 `{"status":"ok"}` | 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 |
|
| 优雅退出 | `SIGTERM`,约 **≤30s** 平台窗口;本实现目标 ~1s 退出主循环 | B |
|
||||||
| 平台注入 | `STRATEGY_ID`;`EXTERNAL_SERVICE_TOKEN`(及社区别名) | A/B |
|
| 平台注入 | `STRATEGY_ID`;`EXTERNAL_SERVICE_TOKEN`(及社区别名) | A/B |
|
||||||
| **STRATEGY_ID 缺失** | **Phase1:进程启动失败(exit ≠0)** | 产品规划 Phase1 + PRD |
|
| **STRATEGY_ID 缺失** | **Phase1:进程启动失败(exit ≠0)** | 产品规划 Phase1 + PRD |
|
||||||
| 资源 | 约 100m–1 CPU / 256–512Mi | B |
|
| 资源 | 约 100m–1 CPU / 256–512Mi | B |
|
||||||
| **发布智能体得分** | **0 分**(发布 ≠ 得分) | B |
|
| **发布智能体得分** | **0 分**(发布 ≠ 得分) | B |
|
||||||
| Live 默认 | **开**(Victor 已批准);`ALLOW_LIVE_SUBMIT=0` 停交,`LIVE_CONFIRM` 可选 | 本仓 |
|
| Live 默认 | **开**(Victor 已批准);`ALLOW_LIVE_SUBMIT=0` 停交,`LIVE_CONFIRM` 可选 | 本仓 |
|
||||||
| 积分 | 来自**成功适配**;智能体代提成功计同一成功分;公式/周报 **待平台核实** | B / D |
|
| 积分 | 来自**验证通过**(`verify_pass`,`verifyResult==1`)的适配;status=`success` 且 verifyResult=-1 为验证失败、0 分;公式/周报 **待平台核实** | B / D |
|
||||||
|
|
||||||
## 鉴权双轨(文档索引,Phase1 不代调)
|
## 鉴权双轨(文档索引,Phase1 不代调)
|
||||||
|
|
||||||
@@ -61,7 +61,10 @@ Victor 确认已内置,`LIVE_CONFIRM` 仍接受但不必需。
|
|||||||
缺凭证、HTTP/平台错误、JSON 错误、分页不完整或无效配置均停交。
|
缺凭证、HTTP/平台错误、JSON 错误、分页不完整或无效配置均停交。
|
||||||
ASYNC_QUEUE_DEPTH / QUEUE_SNAPSHOT_JSON 仅为额外阻塞提示。
|
ASYNC_QUEUE_DEPTH / QUEUE_SNAPSHOT_JSON 仅为额外阻塞提示。
|
||||||
每周期最多 `LIVE_MAX_SUBMITS_PER_CYCLE`(默认 3)次;周期
|
每周期最多 `LIVE_MAX_SUBMITS_PER_CYCLE`(默认 3)次;周期
|
||||||
`LIVE_CYCLE_SECONDS` 默认 600、最小 60 秒,通过 shutdown Event 等待。
|
`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 并记录原因;日志不记录响应正文或凭证。
|
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=<安全原因>`。
|
||||||
|
|||||||
58
scripts/sync_reject_marks.py
Normal file
58
scripts/sync_reject_marks.py
Normal file
@@ -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()
|
||||||
213
src/all_tasks_foothold.py
Normal file
213
src/all_tasks_foothold.py
Normal file
@@ -0,0 +1,213 @@
|
|||||||
|
"""Bounded all-scope evidence slices, isolated from my-queue state."""
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
import urllib.parse
|
||||||
|
import urllib.request
|
||||||
|
from .config_params import GPU_ENUMS
|
||||||
|
from .live_queue import model_key, valid_total
|
||||||
|
from .pending_pool import classify_foothold
|
||||||
|
from .task_add_client import resolve_token
|
||||||
|
|
||||||
|
TARGET_GPUS = ('Biren_166m', 'Sunrise_pt-200-x1', 'Iluvatar_bi-150', 'Iluvatar_mrv-100')
|
||||||
|
RANK = {'verified_reuse': 0, 'failed_foothold': 1, 'in_flight': 2}
|
||||||
|
REASONS = {'ok', 'missing/invalid_token', 'transport_error', 'http_error',
|
||||||
|
'platform_error', 'malformed_data', 'filter_mismatch', 'deadline',
|
||||||
|
'conflicting_task', 'reader_error'}
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class FootholdSnapshot:
|
||||||
|
candidates: list = field(default_factory=list)
|
||||||
|
reason: str = 'ok'
|
||||||
|
buckets: dict = field(default_factory=dict)
|
||||||
|
|
||||||
|
|
||||||
|
def _limit(env, default):
|
||||||
|
try:
|
||||||
|
value = int(os.environ.get(env, str(default)))
|
||||||
|
return min(value, default) if value > 0 else default
|
||||||
|
except ValueError:
|
||||||
|
return default
|
||||||
|
|
||||||
|
|
||||||
|
def classify_api_record(record, bucket):
|
||||||
|
"""Positive whitelist. A mismatch invalidates the complete snapshot."""
|
||||||
|
if not isinstance(record, dict):
|
||||||
|
return None
|
||||||
|
status = record.get('status')
|
||||||
|
status = status.strip().lower() if isinstance(status, str) else None
|
||||||
|
result = record.get('verifyResult')
|
||||||
|
if type(result) is str and result.strip() in ('1', '-1', '0'):
|
||||||
|
result = int(result.strip())
|
||||||
|
if result is not None and type(result) is not int:
|
||||||
|
return None
|
||||||
|
expected = {'pass': ('success', 1), 'fail': ('success', -1)}
|
||||||
|
if bucket in expected:
|
||||||
|
if (status, result) != expected[bucket]:
|
||||||
|
return None
|
||||||
|
elif bucket != 'running' or status != 'running' or result not in (None, 0):
|
||||||
|
return None
|
||||||
|
model, gpu = record.get('modelId'), record.get('gpuType')
|
||||||
|
if not isinstance(model, str) or not re.fullmatch(r'[\w.-]+/[\w./-]+', model_key(model)):
|
||||||
|
return None
|
||||||
|
if not isinstance(gpu, str) or not re.fullmatch(r'[\w .-]+', gpu.strip()):
|
||||||
|
return None
|
||||||
|
gpu = GPU_ENUMS.get(gpu.strip(), gpu.strip())
|
||||||
|
return (model.strip(), gpu, {'pass': 'verified_reuse', 'fail': 'failed_foothold', 'running': 'in_flight'}[bucket])
|
||||||
|
|
||||||
|
|
||||||
|
def build_candidates(evidence, *, target_gpus=TARGET_GPUS):
|
||||||
|
groups = {}
|
||||||
|
targets = list(dict.fromkeys(GPU_ENUMS[g] for g in target_gpus if g in GPU_ENUMS))
|
||||||
|
for model, gpu, kind in evidence:
|
||||||
|
key = model_key(model)
|
||||||
|
row = groups.setdefault(key, dict(model_address=model, section='A',
|
||||||
|
foothold_class=kind, evidence_source='all-tasks', no_download=False, seen_gpus=[]))
|
||||||
|
if gpu not in row['seen_gpus']:
|
||||||
|
row['seen_gpus'].append(gpu)
|
||||||
|
if RANK[kind] < RANK[row['foothold_class']]:
|
||||||
|
row['foothold_class'] = kind
|
||||||
|
for row in groups.values():
|
||||||
|
row['try_next_gpus'] = [g for g in targets if g not in row['seen_gpus']]
|
||||||
|
return sorted(groups.values(), key=candidate_priority)
|
||||||
|
|
||||||
|
|
||||||
|
def _kind(row):
|
||||||
|
text = row.get('foothold_text', '')
|
||||||
|
if text and row.get('evidence_source') != 'all-tasks':
|
||||||
|
return classify_foothold(text)
|
||||||
|
kind = row.get('foothold_class', 'unknown')
|
||||||
|
if kind == 'in_flight' and row.get('evidence_source') != 'all-tasks':
|
||||||
|
return 'unknown'
|
||||||
|
return kind
|
||||||
|
|
||||||
|
|
||||||
|
def candidate_priority(row):
|
||||||
|
return (RANK.get(_kind(row), 3), row.get('section', 'B') != 'A')
|
||||||
|
|
||||||
|
|
||||||
|
def _choices(row):
|
||||||
|
values = row['try_next_gpus'] if 'try_next_gpus' in row else [row.get('targetGpu')]
|
||||||
|
return list(dict.fromkeys(GPU_ENUMS[g] for g in values if isinstance(g, str) and g in GPU_ENUMS))
|
||||||
|
|
||||||
|
|
||||||
|
def merge_candidates(base, dynamic):
|
||||||
|
"""Keep base constraints; intersect explicit targets and subtract seen cards."""
|
||||||
|
rows = {}
|
||||||
|
constrained = set()
|
||||||
|
for incoming in [*base, *dynamic]:
|
||||||
|
key = model_key(incoming.get('model_address') or incoming.get('modelAddress'))
|
||||||
|
row = dict(incoming)
|
||||||
|
row['foothold_class'] = _kind(row)
|
||||||
|
explicit_targets = 'try_next_gpus' in row or 'targetGpu' in row
|
||||||
|
row['try_next_gpus'] = _choices(row)
|
||||||
|
if key not in rows:
|
||||||
|
rows[key] = row
|
||||||
|
if explicit_targets:
|
||||||
|
constrained.add(key)
|
||||||
|
continue
|
||||||
|
existing = rows[key]
|
||||||
|
seen = set(existing.get('seen_gpus', [])) | set(row.get('seen_gpus', []))
|
||||||
|
existing['seen_gpus'] = list(seen)
|
||||||
|
if key not in constrained and explicit_targets:
|
||||||
|
existing['try_next_gpus'] = row['try_next_gpus']
|
||||||
|
constrained.add(key)
|
||||||
|
incoming_choices = row['try_next_gpus'] if explicit_targets else existing['try_next_gpus']
|
||||||
|
existing['try_next_gpus'] = [g for g in existing['try_next_gpus'] if g in incoming_choices and g not in seen]
|
||||||
|
if candidate_priority(row)[0] < candidate_priority(existing)[0]:
|
||||||
|
existing['foothold_class'] = row['foothold_class']
|
||||||
|
existing['evidence_source'] = row.get('evidence_source')
|
||||||
|
if row.get('skip_reason') and not existing.get('skip_reason'):
|
||||||
|
existing['skip_reason'] = row['skip_reason']
|
||||||
|
return sorted(rows.values(), key=candidate_priority)
|
||||||
|
|
||||||
|
|
||||||
|
def load_all_tasks_foothold(*, opener=None, token_resolver=None, clock=None):
|
||||||
|
opener = opener or urllib.request.urlopen
|
||||||
|
clock = clock or time.monotonic
|
||||||
|
buckets = {}
|
||||||
|
def fail(reason):
|
||||||
|
return FootholdSnapshot(reason=reason, buckets=buckets)
|
||||||
|
try:
|
||||||
|
token = (token_resolver or resolve_token)()
|
||||||
|
except Exception:
|
||||||
|
return fail('missing/invalid_token')
|
||||||
|
deadline = clock() + 20
|
||||||
|
size = _limit('ALL_TASKS_PAGE_SIZE', 100)
|
||||||
|
evidence, tasks = [], {}
|
||||||
|
for bucket, cap, filters in (
|
||||||
|
('pass', 3, {'verifyResult': 1, 'status': 'success'}),
|
||||||
|
('fail', 3, {'verifyResult': -1, 'status': 'success'}),
|
||||||
|
('running', 2, {'status': 'running'}),
|
||||||
|
):
|
||||||
|
info = buckets[bucket] = dict(pages=0, valid=0, unique=0, truncated=False)
|
||||||
|
fingerprints = set()
|
||||||
|
raw_count = 0
|
||||||
|
for page in range(1, _limit('ALL_TASKS_MAX_PAGES_' + bucket.upper(), cap) + 1):
|
||||||
|
remaining = deadline - clock()
|
||||||
|
if remaining <= 0:
|
||||||
|
return fail('deadline')
|
||||||
|
query = urllib.parse.urlencode(dict(onlyMine='false', pageSize=size, current=page, **filters))
|
||||||
|
request = urllib.request.Request(os.environ.get('MODELHUB_API_BASE', 'https://modelhub.org.cn').rstrip('/') + '/api/adapt/task/page?' + query,
|
||||||
|
headers={'Xc-Token': token, 'Accept': 'application/json'})
|
||||||
|
try:
|
||||||
|
with opener(request, timeout=min(5, remaining)) as response:
|
||||||
|
if not 200 <= response.status < 300:
|
||||||
|
return fail('http_error')
|
||||||
|
body = json.loads(response.read())
|
||||||
|
except (ValueError, UnicodeError):
|
||||||
|
return fail('malformed_data')
|
||||||
|
except Exception:
|
||||||
|
return fail('transport_error')
|
||||||
|
if clock() >= deadline:
|
||||||
|
return fail('deadline')
|
||||||
|
if not isinstance(body, dict) or type(body.get('code')) not in (int, str) or str(body['code']) != '0':
|
||||||
|
return fail('platform_error')
|
||||||
|
data = body.get('data')
|
||||||
|
if not isinstance(data, dict) or not valid_total(data.get('total')):
|
||||||
|
return fail('malformed_data')
|
||||||
|
records = data.get('records')
|
||||||
|
if not isinstance(records, list) or len(records) > size:
|
||||||
|
return fail('malformed_data')
|
||||||
|
info['pages'] += 1
|
||||||
|
page_evidence = []
|
||||||
|
for record in records:
|
||||||
|
item = classify_api_record(record, bucket)
|
||||||
|
if item is None or any(token.lower() in part.lower() for part in item):
|
||||||
|
return fail('filter_mismatch')
|
||||||
|
task = record.get('taskId')
|
||||||
|
if task is not None and (type(task) not in (str, int) or not str(task).strip()):
|
||||||
|
return fail('malformed_data')
|
||||||
|
identity = ('task', str(task)) if task is not None else ('evidence', model_key(item[0]), item[1], item[2])
|
||||||
|
normalized = (model_key(item[0]), *item[1:])
|
||||||
|
if identity in tasks and tasks[identity] != normalized:
|
||||||
|
return fail('conflicting_task')
|
||||||
|
info['valid'] += 1
|
||||||
|
page_evidence.append(normalized)
|
||||||
|
if identity not in tasks:
|
||||||
|
tasks[identity] = normalized
|
||||||
|
evidence.append(item)
|
||||||
|
info['unique'] += 1
|
||||||
|
fingerprint = tuple(page_evidence)
|
||||||
|
raw_count += len(records)
|
||||||
|
info['truncated'] = bool(records) and raw_count < int(data['total'])
|
||||||
|
if fingerprint in fingerprints:
|
||||||
|
info['truncated'] = True
|
||||||
|
break
|
||||||
|
fingerprints.add(fingerprint)
|
||||||
|
if not records or not info['truncated']:
|
||||||
|
break
|
||||||
|
return FootholdSnapshot(build_candidates(evidence), buckets=buckets)
|
||||||
|
|
||||||
|
|
||||||
|
def report_snapshot(snapshot):
|
||||||
|
reason = snapshot.reason if snapshot.reason in REASONS else 'reader_error'
|
||||||
|
print('[foothold] reason=' + reason, file=sys.stderr)
|
||||||
|
for bucket in ('pass', 'fail', 'running'):
|
||||||
|
if bucket in snapshot.buckets:
|
||||||
|
info = snapshot.buckets[bucket]
|
||||||
|
print('[foothold] bucket=' + bucket + ' ' + ' '.join(f'{k}={int(info[k])}' for k in ('pages', 'valid', 'unique', 'truncated')), file=sys.stderr)
|
||||||
@@ -2,6 +2,10 @@
|
|||||||
import re
|
import re
|
||||||
|
|
||||||
RULES = (
|
RULES = (
|
||||||
|
('gguf', r'gguf|\.gguf|ggml'),
|
||||||
|
('openchat', r'openchat'),
|
||||||
|
# Full-weight exports override adapter hints anywhere in the identifier.
|
||||||
|
('peft-lora', r'^(?!.*(?:merged|merge|fullft|full-ft)).*(?:\blora\b|adapter_model|adapter-config|peft)'),
|
||||||
('gpt2', r'gpt-?2|dialogpt|distilgpt2'),
|
('gpt2', r'gpt-?2|dialogpt|distilgpt2'),
|
||||||
('phi', r'phi[-_. ]?(?:3(?:\.5)?|4)(?:\D|$)'),
|
('phi', r'phi[-_. ]?(?:3(?:\.5)?|4)(?:\D|$)'),
|
||||||
('nemotron-h', r'nemotron-h'), ('olmo2', r'olmo-?2'),
|
('nemotron-h', r'nemotron-h'), ('olmo2', r'olmo-?2'),
|
||||||
|
|||||||
117
src/data/reject-marks.json
Normal file
117
src/data/reject-marks.json
Normal file
@@ -0,0 +1,117 @@
|
|||||||
|
{
|
||||||
|
"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"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "Qurtana/Qwen-2.5-0.5B-MathInstruct.rev.2",
|
||||||
|
"gpu": "Biren_166m",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:SCORE_GAP",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "prithivMLmods/Reasoning-SmolLM2-135M",
|
||||||
|
"gpu": "Biren_166m",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:SCORE_GAP",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "Qwen/Qwen2.5-Coder-7B-Instruct",
|
||||||
|
"gpu": "Sunrise_pt-200-x1",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:SCORE_GAP",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "FlofloB/100k_fineweb_continued_pretraining_Qwen2.5-0.5B-Instruct_Unsloth_merged_16bit",
|
||||||
|
"gpu": "Biren_166m",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:SCORE_GAP",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "prithivMLmods/Bellatrix-Tiny-0.5B",
|
||||||
|
"gpu": "Biren_166m",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:SCORE_GAP",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "sunmengjie/DeepSeek-R1-Distill-Qwen-1.5B-GRPO",
|
||||||
|
"gpu": "Biren_166m",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:PEFT_UNMERGED",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "DevQuasar/efficientscaling.Z1-7B-GGUF",
|
||||||
|
"gpu": "Mthreads_s4000",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:NO_IDLE_DEVICE",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"model": "abaryan/CyberXP_Agent_Llama_3.2_1B",
|
||||||
|
"gpu": "Biren_166m",
|
||||||
|
"code": 60029,
|
||||||
|
"reason": "VERIFY_FAIL:REF_CACHE_MISS",
|
||||||
|
"date": "2026-10-17"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
|||||||
import json
|
import json
|
||||||
from typing import Any, Dict, Tuple
|
from typing import Any, Dict, Tuple
|
||||||
|
|
||||||
|
from .live_queue import last_stats
|
||||||
from .strategy import external_service_token_present, strategy_id_present
|
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]:
|
def status_payload() -> Dict[str, Any]:
|
||||||
"""Booleans only — never include STRATEGY_ID or token values."""
|
"""Safe presence flags and last-known verification statistics."""
|
||||||
return {
|
return {
|
||||||
|
"stats": last_stats(),
|
||||||
|
"success_metric": "verify_pass",
|
||||||
"strategy_id_present": strategy_id_present(),
|
"strategy_id_present": strategy_id_present(),
|
||||||
"external_service_token_present": external_service_token_present(),
|
"external_service_token_present": external_service_token_present(),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,24 +12,11 @@ def load_baked(path=DATA):
|
|||||||
{model_key(m) for m in data['blocked_models']})
|
{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)()
|
count, reason, remote = (reader or read_queue)()
|
||||||
if type(count) is not int or count < 0:
|
if type(count) is not int or count < 0:
|
||||||
raise ValueError('ledger_rebuild_failed')
|
raise ValueError('ledger_rebuild_failed')
|
||||||
baked, blocked = load_baked()
|
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)
|
|
||||||
|
|||||||
@@ -7,21 +7,26 @@ from pathlib import Path
|
|||||||
from .banned import banned_rule
|
from .banned import banned_rule
|
||||||
from .config_params import GPU_ENUMS, build_config_params_with_meta
|
from .config_params import GPU_ENUMS, build_config_params_with_meta
|
||||||
from .gate import evaluate_gate, queue_threshold
|
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 .pending_pool import build_ordered_candidates
|
||||||
|
from .all_tasks_foothold import (load_all_tasks_foothold, merge_candidates, candidate_priority,
|
||||||
|
FootholdSnapshot, report_snapshot)
|
||||||
from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source, token_probe_status
|
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
|
||||||
|
|
||||||
_submitted = set()
|
_submitted = set()
|
||||||
|
_marked = {}
|
||||||
_cycle_lock = threading.Lock()
|
_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, precheck=None, foothold_reader=None):
|
||||||
with _cycle_lock:
|
with _cycle_lock:
|
||||||
return _run_cycle(candidates, client, queue_reader, now)
|
return _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck, foothold_reader)
|
||||||
|
|
||||||
|
|
||||||
def _run_cycle(candidates, client, queue_reader, now):
|
def _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck, foothold_reader):
|
||||||
decisions = []
|
decisions = []
|
||||||
|
|
||||||
def emit(action, reason, **fields):
|
def emit(action, reason, **fields):
|
||||||
@@ -43,6 +48,28 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
emit('hold', 'missing/invalid_token')
|
emit('hold', 'missing/invalid_token')
|
||||||
return decisions
|
return decisions
|
||||||
print('[live] token source=' + source, file=sys.stderr)
|
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:
|
try:
|
||||||
cap = int(os.environ.get('LIVE_MAX_SUBMITS_PER_CYCLE', '3'))
|
cap = int(os.environ.get('LIVE_MAX_SUBMITS_PER_CYCLE', '3'))
|
||||||
if cap < 1:
|
if cap < 1:
|
||||||
@@ -50,6 +77,12 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
except ValueError:
|
except ValueError:
|
||||||
emit('hold', 'invalid_cycle_cap')
|
emit('hold', 'invalid_cycle_cap')
|
||||||
return decisions
|
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:
|
if candidates is None:
|
||||||
try:
|
try:
|
||||||
if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
|
if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
|
||||||
@@ -59,21 +92,73 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
except (OSError, ValueError, IndexError):
|
except (OSError, ValueError, IndexError):
|
||||||
emit('hold', 'candidate_read_failed')
|
emit('hold', 'candidate_read_failed')
|
||||||
return decisions
|
return decisions
|
||||||
candidates = sorted(candidates, key=lambda c: (c.get('section', 'B') != 'A', c.get('foothold_class') != 'verified_reuse'))
|
if os.environ.get('ALL_TASKS_FOOTHOLD', '0') == '1' and (foothold_reader is not None or queue_reader is None):
|
||||||
|
try:
|
||||||
|
snapshot = (foothold_reader or load_all_tasks_foothold)()
|
||||||
|
if not isinstance(snapshot, FootholdSnapshot):
|
||||||
|
raise ValueError
|
||||||
|
except Exception:
|
||||||
|
snapshot = FootholdSnapshot(reason='reader_error')
|
||||||
|
report_snapshot(snapshot)
|
||||||
|
dynamic = snapshot.candidates if snapshot.reason == 'ok' else []
|
||||||
|
if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
|
||||||
|
section = os.environ.get('PENDING_SECTION', '').strip().upper()
|
||||||
|
if section:
|
||||||
|
dynamic = [row for row in dynamic if row.get('section') == section]
|
||||||
|
candidates = merge_candidates(candidates, dynamic)
|
||||||
|
candidates = sorted(candidates, key=candidate_priority)
|
||||||
if not candidates:
|
if not candidates:
|
||||||
emit('hold', 'no_candidates')
|
emit('hold', 'no_candidates')
|
||||||
return decisions
|
return decisions
|
||||||
state = Path(os.environ.get('LIVE_STATE_FILE', '/tmp/live-submitted.json'))
|
|
||||||
try:
|
try:
|
||||||
persisted = load_persisted(state) | _submitted
|
marks = load_marks()
|
||||||
|
except ValueError:
|
||||||
|
emit('hold', 'marks_read_failed')
|
||||||
|
return decisions
|
||||||
|
try:
|
||||||
baked, blocked = load_baked()
|
baked, blocked = load_baked()
|
||||||
known = persisted | baked
|
known = baked | _submitted
|
||||||
except (OSError, ValueError, KeyError, TypeError):
|
except (OSError, ValueError, KeyError, TypeError):
|
||||||
emit('hold', 'state_read_failed')
|
emit('hold', 'state_read_failed')
|
||||||
return decisions
|
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
|
reader = queue_reader or read_queue
|
||||||
client = client if client is not None else TaskAddClient()
|
client = client if client is not None else TaskAddClient()
|
||||||
attempts = 0
|
attempts = 0
|
||||||
|
counted = 0
|
||||||
for candidate in candidates:
|
for candidate in candidates:
|
||||||
model = candidate.get('model_address') or candidate.get('modelAddress') or ''
|
model = candidate.get('model_address') or candidate.get('modelAddress') or ''
|
||||||
rule = banned_rule(model + ' ' + str(candidate.get('model_name', candidate.get('modelName', ''))))
|
rule = banned_rule(model + ' ' + str(candidate.get('model_name', candidate.get('modelName', ''))))
|
||||||
@@ -90,11 +175,35 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
if candidate.get('skip_reason'):
|
if candidate.get('skip_reason'):
|
||||||
emit('skip', 'pending_filter', model=model)
|
emit('skip', 'pending_filter', model=model)
|
||||||
continue
|
continue
|
||||||
if attempts >= cap:
|
if counted >= cap:
|
||||||
emit('hold', 'cycle_cap')
|
emit('hold', 'cycle_cap')
|
||||||
break
|
break
|
||||||
choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')]
|
mapping = {}
|
||||||
gpu = next((g for g in choices if g in GPU_ENUMS and (key, GPU_ENUMS[g]) not in known), None)
|
choices = candidate['try_next_gpus'] if 'try_next_gpus' in candidate else [candidate.get('targetGpu')]
|
||||||
|
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:
|
if gpu is None:
|
||||||
emit('skip', 'all_gpus_done_or_unmapped', model=model)
|
emit('skip', 'all_gpus_done_or_unmapped', model=model)
|
||||||
continue
|
continue
|
||||||
@@ -107,7 +216,7 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
except Exception:
|
except Exception:
|
||||||
emit('hold', 'ledger_rebuild_failed')
|
emit('hold', 'ledger_rebuild_failed')
|
||||||
break
|
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:
|
if gpu is None:
|
||||||
emit('skip', 'all_gpus_done_or_unmapped', model=model)
|
emit('skip', 'all_gpus_done_or_unmapped', model=model)
|
||||||
continue
|
continue
|
||||||
@@ -126,13 +235,7 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
continue
|
continue
|
||||||
pair = (key, GPU_ENUMS[gpu])
|
pair = (key, GPU_ENUMS[gpu])
|
||||||
known.add(pair)
|
known.add(pair)
|
||||||
persisted.add(pair)
|
|
||||||
_submitted.add(pair)
|
_submitted.add(pair)
|
||||||
try:
|
|
||||||
persist(state, persisted)
|
|
||||||
except OSError:
|
|
||||||
emit('hold', 'state_write_failed')
|
|
||||||
break
|
|
||||||
attempts += 1
|
attempts += 1
|
||||||
try:
|
try:
|
||||||
result = client.submit(payload)
|
result = client.submit(payload)
|
||||||
@@ -142,6 +245,24 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
reason = 'task_add_response' if status is not None else 'transport_error'
|
reason = 'task_add_response' if status is not None else 'transport_error'
|
||||||
except Exception:
|
except Exception:
|
||||||
status, code, reason = None, None, 'submit_error'
|
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)
|
||||||
|
_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
|
||||||
|
if attempts >= attempt_cap:
|
||||||
|
emit('hold', 'attempt_cap')
|
||||||
|
break
|
||||||
|
continue
|
||||||
|
counted += 1
|
||||||
emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
|
emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
|
||||||
http=status, code=code, **meta)
|
http=status, code=code, **meta)
|
||||||
|
if attempts >= attempt_cap:
|
||||||
|
emit('hold', 'attempt_cap')
|
||||||
|
break
|
||||||
return decisions
|
return decisions
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
"""Fresh authenticated queue snapshots. Never expose response bodies or credentials."""
|
"""Fresh authenticated queue snapshots. Never expose response bodies or credentials."""
|
||||||
import json
|
import json
|
||||||
|
import math
|
||||||
|
import time
|
||||||
import os
|
import os
|
||||||
import urllib.request
|
import urllib.request
|
||||||
from .task_add_client import resolve_token
|
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'}
|
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):
|
def model_key(value):
|
||||||
value = str(value or '').strip().lower().rstrip('/')
|
value = str(value or '').strip().lower().rstrip('/')
|
||||||
for prefix in ('https://huggingface.co/', 'http://huggingface.co/', 'https://modelscope.cn/models/'):
|
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())
|
return (type(value) is int and value >= 0) or (isinstance(value, str) and value.isascii() and value.isdigit())
|
||||||
|
|
||||||
|
|
||||||
def read_queue():
|
def read_queue_stats():
|
||||||
"""Return (count, reason, model identities); count None means fail closed."""
|
"""Return (count, reason, models, stats); count None means fail closed."""
|
||||||
|
global _last_stats
|
||||||
|
|
||||||
try:
|
try:
|
||||||
token = resolve_token()
|
token = resolve_token()
|
||||||
except (ValueError, OSError, UnicodeError):
|
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
|
count, models, seen, total = 0, set(), 0, None
|
||||||
|
stats = compute_stats([])
|
||||||
for page in range(1, 21):
|
for page in range(1, 21):
|
||||||
try:
|
try:
|
||||||
request = urllib.request.Request(
|
request = urllib.request.Request(
|
||||||
@@ -34,27 +85,30 @@ def read_queue():
|
|||||||
headers={'Xc-Token': token, 'Accept': 'application/json'}, method='GET')
|
headers={'Xc-Token': token, 'Accept': 'application/json'}, method='GET')
|
||||||
with urllib.request.urlopen(request, timeout=30) as response:
|
with urllib.request.urlopen(request, timeout=30) as response:
|
||||||
if not 200 <= response.status < 300:
|
if not 200 <= response.status < 300:
|
||||||
return None, 'http_error', set()
|
return None, 'http_error', set(), None
|
||||||
body = json.loads(response.read())
|
body = json.loads(response.read())
|
||||||
except (ValueError, UnicodeError):
|
except (ValueError, UnicodeError):
|
||||||
return None, 'malformed_json', set()
|
return None, 'malformed_json', set(), None
|
||||||
except OSError:
|
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':
|
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')
|
data = body.get('data')
|
||||||
if not isinstance(data, dict) or not valid_total(data.get('total')):
|
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'])
|
data['total'] = int(data['total'])
|
||||||
records = data.get('records')
|
records = data.get('records')
|
||||||
if not isinstance(records, list) or any(not isinstance(r, dict) for r in 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']:
|
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']
|
total = data['total']
|
||||||
seen += len(records)
|
seen += len(records)
|
||||||
if seen > total:
|
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:
|
for record in records:
|
||||||
count += str(record.get('status', '')).strip().lower() not in TERMINAL
|
count += str(record.get('status', '')).strip().lower() not in TERMINAL
|
||||||
model, gpu = model_key(record.get('modelId')), record.get('gpuType')
|
model, gpu = model_key(record.get('modelId')), record.get('gpuType')
|
||||||
@@ -62,7 +116,13 @@ def read_queue():
|
|||||||
models.add((model, gpu))
|
models.add((model, gpu))
|
||||||
if seen == total:
|
if seen == total:
|
||||||
# Do not retain arbitrary response fields, which may echo credentials.
|
# 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:
|
if not records:
|
||||||
break
|
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]
|
||||||
|
|||||||
53
src/marks.py
Normal file
53
src/marks.py
Normal file
@@ -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
|
||||||
@@ -46,12 +46,12 @@ def classify_foothold(text):
|
|||||||
"""Classify literal pending evidence; mixed markers prefer verified."""
|
"""Classify literal pending evidence; mixed markers prefer verified."""
|
||||||
for label, markers in (
|
for label, markers in (
|
||||||
('verified_reuse', ('已验证成功', '验证通过')),
|
('verified_reuse', ('已验证成功', '验证通过')),
|
||||||
('in_flight', ('正在验证', '验证排队中', '验证进行中')),
|
('in_flight', ('正在验证', '验证进行中')),
|
||||||
('failed_foothold', ('验证失败',)),
|
('failed_foothold', ('验证失败',)),
|
||||||
):
|
):
|
||||||
if any(marker in text for marker in markers):
|
if any(marker in text for marker in markers):
|
||||||
return label
|
return label
|
||||||
return 'unknown'
|
return 'queued_only' if '验证排队中' in text else 'unknown'
|
||||||
|
|
||||||
|
|
||||||
def has_no_download_marker(text):
|
def has_no_download_marker(text):
|
||||||
|
|||||||
55
src/precheck.py
Normal file
55
src/precheck.py
Normal file
@@ -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
|
||||||
@@ -59,15 +59,14 @@ def _on_sigterm(signum: int, frame) -> None: # noqa: ARG001
|
|||||||
|
|
||||||
|
|
||||||
def _live_worker():
|
def _live_worker():
|
||||||
from pathlib import Path
|
|
||||||
from .ledger import rebuild
|
from .ledger import rebuild
|
||||||
try:
|
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:
|
except Exception:
|
||||||
sys.stderr.write('[live] hold reason=ledger_rebuild_failed\n')
|
sys.stderr.write('[live] hold reason=ledger_rebuild_failed\n')
|
||||||
try:
|
try:
|
||||||
interval = max(60, int(os.environ.get('LIVE_CYCLE_SECONDS', '600')))
|
interval = max(60, int(os.environ.get('LIVE_CYCLE_SECONDS', '3600')))
|
||||||
except ValueError:
|
except ValueError:
|
||||||
interval = 600
|
interval = 600
|
||||||
while not _shutdown.is_set():
|
while not _shutdown.is_set():
|
||||||
|
|||||||
125
tests/test_all_tasks_foothold.py
Normal file
125
tests/test_all_tasks_foothold.py
Normal file
@@ -0,0 +1,125 @@
|
|||||||
|
import io
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import Mock, patch
|
||||||
|
from urllib.parse import parse_qs, urlparse
|
||||||
|
from src.all_tasks_foothold import (load_all_tasks_foothold, classify_api_record,
|
||||||
|
build_candidates, merge_candidates, candidate_priority, TARGET_GPUS, FootholdSnapshot)
|
||||||
|
from src import live_queue
|
||||||
|
|
||||||
|
|
||||||
|
def record(status='success', result=1, model='org/model', gpu='壁砺', **extra):
|
||||||
|
return dict(modelId=model, gpuType=gpu, status=status, verifyResult=result, **extra)
|
||||||
|
|
||||||
|
|
||||||
|
def response(records=(), total=None, **body):
|
||||||
|
result = Mock(status=200)
|
||||||
|
result.read.return_value = json.dumps(body or dict(code=0, data=dict(records=list(records), total=len(records) if total is None else total))).encode()
|
||||||
|
result.__enter__ = Mock(return_value=result)
|
||||||
|
result.__exit__ = Mock(return_value=False)
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
class TestFoothold(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.env = patch.dict(os.environ, {}, clear=True)
|
||||||
|
self.env.start()
|
||||||
|
self.addCleanup(self.env.stop)
|
||||||
|
self.network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
|
||||||
|
self.network.start()
|
||||||
|
self.addCleanup(self.network.stop)
|
||||||
|
|
||||||
|
def load(self, responses, **kw):
|
||||||
|
opener = Mock(side_effect=responses)
|
||||||
|
snapshot = load_all_tasks_foothold(opener=opener, token_resolver=lambda: 'offline-credential', **kw)
|
||||||
|
return snapshot, opener
|
||||||
|
|
||||||
|
def test_whitelist_rejects_waiting_conflicts_and_bad_types(self):
|
||||||
|
for status, result, bucket in [('waiting', 1, 'pass'), ('failed', -1, 'fail'),
|
||||||
|
('running', 1, 'running'), ('success', True, 'pass'),
|
||||||
|
('success', {}, 'pass'), ('other', 1, 'pass'), ('success', 1.0, 'pass')]:
|
||||||
|
self.assertIsNone(classify_api_record(record(status, result), bucket))
|
||||||
|
for result in (1, '1'):
|
||||||
|
self.assertIsNotNone(classify_api_record(record(result=result), 'pass'))
|
||||||
|
for result in (None, 0, '0'):
|
||||||
|
self.assertIsNotNone(classify_api_record(record(' RUNNING ', result), 'running'))
|
||||||
|
for fields in (dict(modelId=True), dict(gpuType=''), dict(modelId='')):
|
||||||
|
self.assertIsNone(classify_api_record(dict(record(), **fields), 'pass'))
|
||||||
|
|
||||||
|
def test_scope_buckets_aliases_order_and_queue_isolation(self):
|
||||||
|
before = live_queue.last_stats()
|
||||||
|
snap, opener = self.load([response([record()]), response([record(result=-1, model='org/fail')]),
|
||||||
|
response([record('running', None, model='org/run')])])
|
||||||
|
self.assertEqual(snap.reason, 'ok')
|
||||||
|
self.assertEqual([r['foothold_class'] for r in snap.candidates], ['verified_reuse', 'failed_foothold', 'in_flight'])
|
||||||
|
self.assertEqual(snap.candidates[0]['try_next_gpus'], list(TARGET_GPUS[1:]))
|
||||||
|
for call in opener.call_args_list:
|
||||||
|
query = parse_qs(urlparse(call.args[0].full_url).query)
|
||||||
|
self.assertEqual(query['onlyMine'], ['false'])
|
||||||
|
self.assertNotIn('waiting', query.get('status', []))
|
||||||
|
self.assertLessEqual(call.kwargs['timeout'], 5)
|
||||||
|
self.assertEqual(before, live_queue.last_stats())
|
||||||
|
self.assertEqual(len(opener.call_args_list), 3)
|
||||||
|
|
||||||
|
def test_any_bucket_error_discards_whole_snapshot(self):
|
||||||
|
for bad in (response([record('waiting', 1)]), response(data=None, code=0),
|
||||||
|
response(code=403), OSError('private-response')):
|
||||||
|
snap, _ = self.load([response([record()]), bad])
|
||||||
|
self.assertNotEqual(snap.reason, 'ok')
|
||||||
|
self.assertEqual(snap.candidates, [])
|
||||||
|
snap, _ = self.load([response(), response(), response()])
|
||||||
|
self.assertEqual(snap.reason, 'ok')
|
||||||
|
self.assertEqual(snap.candidates, [])
|
||||||
|
|
||||||
|
def test_hard_caps_total_drift_and_invalid_env(self):
|
||||||
|
with patch.dict(os.environ, {'ALL_TASKS_MAX_PAGES_PASS': '999', 'ALL_TASKS_MAX_PAGES_FAIL': 'bad',
|
||||||
|
'ALL_TASKS_MAX_PAGES_RUNNING': '-1', 'ALL_TASKS_PAGE_SIZE': '999'}):
|
||||||
|
pages = [response([record(model='org/p' + str(n))], total=1000+n) for n in range(3)]
|
||||||
|
pages += [response([record(result=-1, model='org/f' + str(n))], total=1000+n) for n in range(3)]
|
||||||
|
pages += [response([record('running', None, model='org/r' + str(n))], total=1000+n) for n in range(2)]
|
||||||
|
snap, opener = self.load(pages)
|
||||||
|
self.assertEqual(opener.call_count, 8)
|
||||||
|
self.assertTrue(all(b['truncated'] for b in snap.buckets.values()))
|
||||||
|
self.assertTrue(all('pageSize=100' in c.args[0].full_url for c in opener.call_args_list))
|
||||||
|
snap, opener = self.load([], clock=Mock(side_effect=[0, 21]))
|
||||||
|
self.assertEqual(snap.reason, 'deadline')
|
||||||
|
opener.assert_not_called()
|
||||||
|
|
||||||
|
def test_repeated_tasks_pages_and_conflicts(self):
|
||||||
|
r = record(taskId=1)
|
||||||
|
snap, opener = self.load([response([r], total=999), response([r], total=1000), response(), response()])
|
||||||
|
self.assertEqual(snap.buckets['pass']['unique'], 1)
|
||||||
|
self.assertTrue(snap.buckets['pass']['truncated'])
|
||||||
|
self.assertEqual(opener.call_count, 4)
|
||||||
|
snap, _ = self.load([response([r]), response([record(result=-1, taskId=1)])])
|
||||||
|
self.assertEqual(snap.reason, 'conflicting_task')
|
||||||
|
self.assertEqual(snap.candidates, [])
|
||||||
|
|
||||||
|
def test_missing_token_never_opens(self):
|
||||||
|
opener = Mock()
|
||||||
|
snap = load_all_tasks_foothold(opener=opener, token_resolver=Mock(side_effect=ValueError()))
|
||||||
|
self.assertEqual(snap.reason, 'missing/invalid_token')
|
||||||
|
opener.assert_not_called()
|
||||||
|
|
||||||
|
def test_merge_constraints_empty_targets_and_uniform_priority(self):
|
||||||
|
base = [dict(model_address='https://huggingface.co/ORG/Model/', framework='custom',
|
||||||
|
skip_reason='manual', section='B', foothold_class='verified_reuse',
|
||||||
|
try_next_gpus=['壁砺'], targetGpu='天垓150'),
|
||||||
|
dict(model_address='org/static', section='A', foothold_class='verified_reuse', try_next_gpus=['天垓150'])]
|
||||||
|
dynamic = build_candidates([('org/model', 'Biren_166m', 'in_flight'), ('org/run', 'MetaX_c-500', 'in_flight')])
|
||||||
|
merged = merge_candidates(base, dynamic)
|
||||||
|
row = next(r for r in merged if 'huggingface' in r['model_address'])
|
||||||
|
self.assertEqual(row['framework'], 'custom')
|
||||||
|
self.assertEqual(row['skip_reason'], 'manual')
|
||||||
|
self.assertEqual(row['try_next_gpus'], [])
|
||||||
|
self.assertEqual(row['foothold_class'], 'verified_reuse')
|
||||||
|
self.assertEqual(len(merged), 3)
|
||||||
|
self.assertEqual(merged[-1]['model_address'], 'org/run')
|
||||||
|
self.assertEqual(merged[-1]['try_next_gpus'], list(TARGET_GPUS))
|
||||||
|
self.assertEqual(candidate_priority(dict(foothold_class='in_flight'))[0], 3)
|
||||||
|
|
||||||
|
def test_pending_queue_only_text(self):
|
||||||
|
from src.pending_pool import classify_foothold
|
||||||
|
self.assertEqual(classify_foothold('验证排队中'), 'queued_only')
|
||||||
|
self.assertEqual(classify_foothold('验证排队中; 验证通过'), 'verified_reuse')
|
||||||
36
tests/test_banned.py
Normal file
36
tests/test_banned.py
Normal file
@@ -0,0 +1,36 @@
|
|||||||
|
import unittest
|
||||||
|
|
||||||
|
from src.banned import banned_rule
|
||||||
|
|
||||||
|
|
||||||
|
class TestBanned(unittest.TestCase):
|
||||||
|
def test_default_engine_and_openchat(self):
|
||||||
|
for model, rule in [('org/Model-GGUF', 'gguf'), ('org/model.gguf', 'gguf'),
|
||||||
|
('org/GGML-model', 'gguf'), ('OpenChat/model', 'openchat')]:
|
||||||
|
with self.subTest(model=model):
|
||||||
|
self.assertEqual(banned_rule(model), rule)
|
||||||
|
|
||||||
|
def test_unmerged_adapter_hints(self):
|
||||||
|
for model in ('org/model-lora', 'org/LORA', 'org/adapter_model',
|
||||||
|
'org/model-adapter-config', 'org/Model-PEFT'):
|
||||||
|
with self.subTest(model=model):
|
||||||
|
self.assertEqual(banned_rule(model), 'peft-lora')
|
||||||
|
|
||||||
|
def test_full_weights_and_unrelated_names(self):
|
||||||
|
for model in ('org/lora-merged', 'org/merge-lora', 'org/PEFT-FULLFT',
|
||||||
|
'org/full-ft-adapter_model', 'org/adapter-config-MERGED',
|
||||||
|
'org/flora', 'org/lorax', 'org/model_lora',
|
||||||
|
'org/model-GRPO', 'abaryan/CyberXP_Agent_Llama_3.2_1B'):
|
||||||
|
with self.subTest(model=model):
|
||||||
|
self.assertIsNone(banned_rule(model))
|
||||||
|
|
||||||
|
def test_full_weight_hint_does_not_override_other_bans(self):
|
||||||
|
self.assertEqual(banned_rule('org/lora-merged-GGUF'), 'gguf')
|
||||||
|
self.assertEqual(banned_rule('org/openchat-fullft'), 'openchat')
|
||||||
|
|
||||||
|
def test_existing_rules(self):
|
||||||
|
for model in ('gpt2', 'org/Phi-3-mini', 'nvidia/Nemotron-H-4B',
|
||||||
|
'allenai/OLMo-2-7B', 'org/LFM2', 'org/Spark-X2.5',
|
||||||
|
'org/Breeze', 'org/gemma2', 'org/mistral-v0.1', 'org/tinyllama'):
|
||||||
|
with self.subTest(model=model):
|
||||||
|
self.assertIsNotNone(banned_rule(model))
|
||||||
@@ -58,3 +58,18 @@ class TestHealth(unittest.TestCase):
|
|||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.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)
|
||||||
|
|||||||
@@ -1,9 +1,8 @@
|
|||||||
import io
|
import io
|
||||||
import os
|
import os
|
||||||
import tempfile
|
|
||||||
import unittest
|
import unittest
|
||||||
from unittest.mock import Mock, patch
|
from unittest.mock import Mock, patch
|
||||||
from src.live_cycle import run_cycle, _submitted
|
from src.live_cycle import run_cycle, _submitted, _marked
|
||||||
|
|
||||||
|
|
||||||
def candidate(name, section='B', foothold='unknown'):
|
def candidate(name, section='B', foothold='unknown'):
|
||||||
@@ -12,12 +11,11 @@ def candidate(name, section='B', foothold='unknown'):
|
|||||||
|
|
||||||
class TestLiveCycle(unittest.TestCase):
|
class TestLiveCycle(unittest.TestCase):
|
||||||
def setUp(self):
|
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.env.start()
|
||||||
self.addCleanup(self.env.stop)
|
self.addCleanup(self.env.stop)
|
||||||
_submitted.clear()
|
_submitted.clear()
|
||||||
|
_marked.clear()
|
||||||
self.client = Mock()
|
self.client = Mock()
|
||||||
self.client.submit.return_value = dict(http_status=200, platform_code=0)
|
self.client.submit.return_value = dict(http_status=200, platform_code=0)
|
||||||
network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
|
network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
|
||||||
@@ -31,6 +29,79 @@ class TestLiveCycle(unittest.TestCase):
|
|||||||
reader = Mock(side_effect=[(n, 'platform_error' if n is None else 'ok', set()) for n in counts])
|
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)
|
return run_cycle(rows or [candidate('a')], self.client, reader)
|
||||||
|
|
||||||
|
def submitted_pairs(self):
|
||||||
|
return sorted(m + '|' + g for m, g in _submitted)
|
||||||
|
|
||||||
|
def test_done_rejection_marks_pair_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.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_marked_and_not_retried_in_process(self):
|
||||||
|
self.client.submit.return_value = dict(http_status=200, platform_code=60020)
|
||||||
|
self.cycle()
|
||||||
|
self.assertEqual(self.submitted_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.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):
|
||||||
|
_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}):
|
||||||
|
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.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()
|
||||||
|
self.client.reset_mock()
|
||||||
|
def submit(payload):
|
||||||
|
self.assertIn('org/a|Biren_166m', self.submitted_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):
|
def test_high_and_failed_queue(self):
|
||||||
for n in (95, None):
|
for n in (95, None):
|
||||||
decisions = self.cycle(counts=(n,))
|
decisions = self.cycle(counts=(n,))
|
||||||
@@ -57,10 +128,9 @@ class TestLiveCycle(unittest.TestCase):
|
|||||||
self.cycle(counts=(99,))
|
self.cycle(counts=(99,))
|
||||||
self.client.submit.assert_called_once()
|
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')]
|
rows = [candidate('GPT-2'), candidate('ok'), candidate('ok')]
|
||||||
self.cycle(rows)
|
self.cycle(rows)
|
||||||
_submitted.clear()
|
|
||||||
self.cycle([candidate('ok')], ())
|
self.cycle([candidate('ok')], ())
|
||||||
self.client.submit.assert_called_once()
|
self.client.submit.assert_called_once()
|
||||||
self.assertIn('banned:', self.logs.getvalue())
|
self.assertIn('banned:', self.logs.getvalue())
|
||||||
@@ -83,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.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())
|
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>'}):
|
with patch.dict(os.environ, {'XC_TOKEN': '<private-secret-fragment>'}):
|
||||||
self.cycle()
|
self.cycle()
|
||||||
self.client.submit.assert_not_called()
|
self.client.submit.assert_not_called()
|
||||||
@@ -107,7 +177,7 @@ class TestLiveCycle(unittest.TestCase):
|
|||||||
response.__exit__ = Mock(return_value=False)
|
response.__exit__ = Mock(return_value=False)
|
||||||
response.status = 200
|
response.status = 200
|
||||||
response.read.return_value = json.dumps({'code': 40100}).encode()
|
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)
|
run_cycle([candidate('a')], self.client)
|
||||||
self.client.submit.assert_not_called()
|
self.client.submit.assert_not_called()
|
||||||
|
|
||||||
@@ -117,3 +187,91 @@ class TestLiveCycle(unittest.TestCase):
|
|||||||
self.client.submit.assert_not_called()
|
self.client.submit.assert_not_called()
|
||||||
self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
|
self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
|
||||||
self.assertIn('placeholder', 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)
|
||||||
|
|
||||||
|
def test_foothold_explicit_disabled_and_injected_queue_stay_offline(self):
|
||||||
|
from src.all_tasks_foothold import FootholdSnapshot
|
||||||
|
dynamic = Mock(return_value=FootholdSnapshot())
|
||||||
|
with patch.dict(os.environ, {'ALL_TASKS_FOOTHOLD': '1'}):
|
||||||
|
run_cycle([candidate('explicit')], self.client, lambda: (0, 'ok', set()), foothold_reader=dynamic)
|
||||||
|
dynamic.assert_not_called()
|
||||||
|
with patch('src.live_cycle.build_ordered_candidates', return_value=[candidate('base')]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline'}):
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=dynamic)
|
||||||
|
with patch.dict(os.environ, {'ALL_TASKS_FOOTHOLD': '1'}):
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()))
|
||||||
|
dynamic.assert_not_called()
|
||||||
|
|
||||||
|
def test_foothold_merge_order_constraints_pending_and_fallback(self):
|
||||||
|
from src.all_tasks_foothold import FootholdSnapshot, build_candidates
|
||||||
|
base = [candidate('base', 'A', 'verified_reuse')]
|
||||||
|
snapshot = FootholdSnapshot(build_candidates([('org/dynamic', 'Biren_166m', 'in_flight')]))
|
||||||
|
with patch('src.live_cycle.build_ordered_candidates', return_value=base) as pending, patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'PENDING_SECTION': 'A', 'ALL_TASKS_FOOTHOLD': '1'}):
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
|
||||||
|
pending.assert_called_once_with()
|
||||||
|
self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/base', 'org/dynamic'])
|
||||||
|
_submitted.clear()
|
||||||
|
self.client.reset_mock()
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=Mock(side_effect=OSError('private-response')))
|
||||||
|
self.assertEqual(self.client.submit.call_args.args[0]['modelAddress'], 'org/base')
|
||||||
|
self.assertIn('[foothold] reason=reader_error', self.logs.getvalue())
|
||||||
|
self.assertNotIn('private-response', self.logs.getvalue())
|
||||||
|
|
||||||
|
def test_foothold_empty_intersection_never_revives_target(self):
|
||||||
|
from src.all_tasks_foothold import FootholdSnapshot, build_candidates
|
||||||
|
base = [dict(candidate('empty'), targetGpu='曦望')]
|
||||||
|
snapshot = FootholdSnapshot(build_candidates([('org/empty', 'Biren_166m', 'verified_reuse')]))
|
||||||
|
with patch('src.live_cycle.build_ordered_candidates', return_value=base), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'ALL_TASKS_FOOTHOLD': '1'}):
|
||||||
|
decisions = run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
|
||||||
|
self.client.submit.assert_not_called()
|
||||||
|
self.assertEqual(decisions[-1]['reason'], 'all_gpus_done_or_unmapped')
|
||||||
|
|
||||||
|
def test_dynamic_still_respects_queue_marks_ledger_precheck(self):
|
||||||
|
from src.all_tasks_foothold import FootholdSnapshot
|
||||||
|
snapshot = FootholdSnapshot([dict(candidate('guarded'), evidence_source='all-tasks', foothold_class='verified_reuse')])
|
||||||
|
with patch('src.live_cycle.build_ordered_candidates', return_value=[]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'ALL_TASKS_FOOTHOLD': '1'}):
|
||||||
|
for count in (95, None):
|
||||||
|
decisions = run_cycle(None, self.client, lambda: (count, 'ok', set()), foothold_reader=lambda: snapshot)
|
||||||
|
self.assertEqual(decisions[-1]['action'], 'hold')
|
||||||
|
with patch('src.live_cycle.load_baked', return_value=({('org/guarded', 'Biren_166m')}, set())):
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
|
||||||
|
_marked[('org/guarded', 'Biren_166m')] = 60028
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot)
|
||||||
|
_marked.clear()
|
||||||
|
with patch.dict(os.environ, {'LIVE_PRECHECK': '1'}):
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot,
|
||||||
|
precheck=lambda _: ('ok', {'Biren_166m': 60027}))
|
||||||
|
self.client.submit.assert_not_called()
|
||||||
|
|
||||||
|
def test_pending_section_b_does_not_gain_dynamic_section_a(self):
|
||||||
|
from src.all_tasks_foothold import FootholdSnapshot
|
||||||
|
dynamic = FootholdSnapshot([candidate('dynamic', 'A', 'verified_reuse')])
|
||||||
|
with patch('src.live_cycle.build_ordered_candidates', return_value=[candidate('base')]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'PENDING_SECTION': 'B', 'ALL_TASKS_FOOTHOLD': '1'}):
|
||||||
|
run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: dynamic)
|
||||||
|
self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/base'])
|
||||||
|
|||||||
@@ -1,10 +1,9 @@
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import tempfile
|
|
||||||
import unittest
|
import unittest
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import Mock, patch
|
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.ledger import load_baked, rebuild
|
||||||
from src.banned import banned_rule
|
from src.banned import banned_rule
|
||||||
from src.task_add_client import resolve_token_with_source
|
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):
|
class TestLedger(unittest.TestCase):
|
||||||
def setUp(self):
|
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()
|
env.start()
|
||||||
self.addCleanup(env.stop)
|
self.addCleanup(env.stop)
|
||||||
_submitted.clear()
|
_submitted.clear()
|
||||||
|
_marked.clear()
|
||||||
self.addCleanup(_submitted.clear)
|
self.addCleanup(_submitted.clear)
|
||||||
net = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
|
net = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
|
||||||
net.start()
|
net.start()
|
||||||
@@ -29,11 +26,10 @@ class TestLedger(unittest.TestCase):
|
|||||||
def cycle(self, remote=(), model='unit/model', gpus=None):
|
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)))
|
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.cycle([('unit/model', 'Biren_166m')])
|
||||||
self.assertEqual(self.client.submit.call_args.args[0]['targetGpu'], 'Iluvatar_bi-100')
|
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.cycle([('unit/model', 'Biren_166m')])
|
||||||
self.client.submit.assert_called_once()
|
self.client.submit.assert_called_once()
|
||||||
|
|
||||||
@@ -67,6 +63,6 @@ class TestLedger(unittest.TestCase):
|
|||||||
self.assertGreater(len(rows), 0)
|
self.assertGreater(len(rows), 0)
|
||||||
self.assertEqual([r['section'] for r in rows], sorted(r['section'] for r in rows))
|
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))
|
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.assertIn(('unit/model', 'Biren_166m'), pairs)
|
||||||
self.assertEqual(sizes[0], 1)
|
self.assertEqual(sizes[0], 1)
|
||||||
|
|||||||
@@ -53,3 +53,52 @@ class TestLiveQueue(unittest.TestCase):
|
|||||||
req = call.call_args.args[0]
|
req = call.call_args.args[0]
|
||||||
self.assertEqual(req.get_header('Xc-token'), 'synthetic-secret')
|
self.assertEqual(req.get_header('Xc-token'), 'synthetic-secret')
|
||||||
self.assertEqual(req.method, 'GET')
|
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()))
|
||||||
|
|||||||
@@ -59,7 +59,7 @@ class TestPendingPool(unittest.TestCase):
|
|||||||
'org/fail', 'org/queue', 'org/progress', 'org/b'])
|
'org/fail', 'org/queue', 'org/progress', 'org/b'])
|
||||||
self.assertEqual([r['foothold_class'] for r in rows],
|
self.assertEqual([r['foothold_class'] for r in rows],
|
||||||
['verified_reuse', 'verified_reuse', 'unknown', 'in_flight',
|
['verified_reuse', 'verified_reuse', 'unknown', 'in_flight',
|
||||||
'failed_foothold', 'in_flight', 'in_flight', 'unknown'])
|
'failed_foothold', 'queued_only', 'in_flight', 'unknown'])
|
||||||
for row in rows:
|
for row in rows:
|
||||||
self.assertEqual(row['evidence_source'], 'GAP' if row['foothold_class'] == 'unknown' else 'pending-foothold')
|
self.assertEqual(row['evidence_source'], 'GAP' if row['foothold_class'] == 'unknown' else 'pending-foothold')
|
||||||
self.assertEqual(rows[0]['foothold_text'], '他人:已验证成功; 正在验证')
|
self.assertEqual(rows[0]['foothold_text'], '他人:已验证成功; 正在验证')
|
||||||
|
|||||||
276
tests/test_reject_marks.py
Normal file
276
tests/test_reject_marks.py
Normal file
@@ -0,0 +1,276 @@
|
|||||||
|
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), 16)
|
||||||
|
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_verify_fail_marks(self):
|
||||||
|
rows = json.loads(Path('src/data/reject-marks.json').read_text())['marks']
|
||||||
|
keys = [(row['model'].lower(), row['gpu']) for row in rows]
|
||||||
|
self.assertEqual(len(keys), len(set(keys)))
|
||||||
|
failures = [row for row in rows if row['reason'].startswith('VERIFY_FAIL:')]
|
||||||
|
self.assertEqual(len(failures), 8)
|
||||||
|
self.assertEqual(sorted(row['reason'] for row in failures),
|
||||||
|
sorted(['VERIFY_FAIL:SCORE_GAP'] * 5 +
|
||||||
|
['VERIFY_FAIL:PEFT_UNMERGED', 'VERIFY_FAIL:NO_IDLE_DEVICE',
|
||||||
|
'VERIFY_FAIL:REF_CACHE_MISS']))
|
||||||
|
marks = load_marks()
|
||||||
|
for row in failures:
|
||||||
|
self.assertEqual(row['code'], 60029)
|
||||||
|
self.assertIn(row['code'], PERMANENT_PAIR_CODES)
|
||||||
|
self.assertEqual(marks[row['model'].lower(), row['gpu']], 60029)
|
||||||
|
self.assertEqual(marks['abaryan/cyberxp_agent_llama_3.2_1b', 'Biren_166m'], 60029)
|
||||||
|
self.assertNotIn(('abaryan/cyberxp_agent_llama_3.2_1b', 'Iluvatar_bi-100'), marks)
|
||||||
|
|
||||||
|
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')
|
||||||
Reference in New Issue
Block a user