3 Commits

30 changed files with 5687 additions and 374 deletions

1
.gitignore vendored
View File

@@ -19,6 +19,7 @@ build/
# Dry-run / local artifacts (keep directory)
reports/*
!reports/.gitkeep
!reports/live-gate-summary.md
# Editor / OS
.DS_Store

View File

@@ -7,6 +7,7 @@ WORKDIR /app
# Copy only what the runtime needs (see .dockerignore)
COPY src/ ./src/
COPY templates/ ./templates/
ENV PORT=8080
ENV PYTHONUNBUFFERED=1

View File

@@ -12,11 +12,11 @@ help:
@echo "build-config-dry generate YAML via API (requires token; never creates task)"
@echo "probe-readonly authenticated read-only task/page GET"
@echo "dry-run-pending read-only adapt pending A→B report"
@echo "test stdlib unit tests + local SIGTERM integration (needs sockets)"
@echo "test stdlib unit tests + offline SIGTERM integration"
@echo "build Docker build using platform base (needs Docker + TCR access)"
@echo "smoke-health local HTTP smoke (needs sockets; default test STRATEGY_ID)"
@echo "dry-run read-only pending report (example fallback)"
@echo "No register/submit/deploy/push — default live off."
@echo "No register/submit/deploy/push — live defaults on; ALLOW_LIVE_SUBMIT=0 stops live."
test:
$(PYTHON) -m unittest discover -s tests -v

View File

@@ -2,8 +2,8 @@
ModelHub XC(信创模盒)**适配智能体** Phase1 本地骨架。
> **不代注册、不代提交。** 发布智能体本身 **不得分**。积分来自成功适配,数字一律 **待平台核实**。
> 冲突裁决:**适配 W1 优先**;本仓默认 dry-run,不抢验证队列。
> **不代注册;Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自成功适配,数字一律 **待平台核实**。
> 冲突裁决:**适配 W1 优先**;本仓默认 live 开启,仅在实时验证队列低于阈值时提交。
## 契约要点
@@ -12,13 +12,21 @@ ModelHub XC(信创模盒)**适配智能体** Phase1 本地骨架。
| 端口 | 8080 |
| 健康检查 | `GET /health` → `{"status":"ok"}` |
| 启动 | **必须**有非空 `STRATEGY_ID`,否则 exit ≠0 |
| Live | `ALLOW_LIVE_SUBMIT=1` **且** `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT`;Phase1 仍无真实 `task/add` 客户端 |
| Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`(勿擅自改默认 FROM) |
| Live | 默认开启;`ALLOW_LIVE_SUBMIT=0` 为停交开关;必须提供 `STRATEGY_ID` |
| Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`(保留 FROM) |
闸门以 `src/gate.py` 实现为准:仅 `ALLOW_LIVE_SUBMIT=1` 与
`LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT` 同时满足才开逻辑闸门。PRD 示例
`ALLOW_LIVE_SUBMIT=true` / `CONFIRM_LIVE=YES` 不受支持;本仓不读取确认文件。
即使开闸,无真实客户端仍返回 blocked;请保持默认关闭。
Victor 的确认已内置,`LIVE_CONFIRM` 仍接受但不再必需。每次 task/add 前,
通过 Xc-Token 全量分页读取 onlyMine task/page;非终态计入队列,读取失败停交。
`LIVE_QUEUE_THRESHOLD=95`(严格小于),`LIVE_MAX_SUBMITS_PER_CYCLE=3`,
`LIVE_CYCLE_SECONDS=600`(最小 60 秒)。环境队列提示仅可额外阻塞。
候选来自 `PENDING_SUBMIT_MD` / `PENDING_SUBMIT_TXT`,否则使用
`src/data/candidates.json`(随镜像内置);A verified_reuse → 其他 A → B。
禁提、重复、GPU 未映射、占位符候选跳过。按 try_next_gpus 顺序选择首个已映射且未提交的模型/GPU 对。`LIVE_STATE_FILE` 默认
`/tmp/live-submitted.json`,格式为 `model|gpuEnum` 字符串列表,在请求前持久化;失败请求也不自动重试。旧版仅模型名状态文件须先离线迁移,否则 fail closed。
去重合并远端全部状态的 `modelId` + `gpuType`、内置 `src/data/submitted-ledger.json` 和持久化状态。A 允许同模型换 GPU;do-not-resubmit 模型跨 GPU 全禁。服务启动重建一次并只记录三类 pair 数;每次提交前刷新远端,失败则 hold `ledger_rebuild_failed`。task/page total 接受非负整数或 ASCII 数字字符串。
凭证优先级:`EXTERNAL_SERVICE_TOKEN` → `XC_TOKEN_FILE` → `XC_TOKEN` → `MODELHUB_XC_TOKEN`;跳过空值和占位符。每周期只记录 source 环境变量名,使用 `Xc-Token` 请求头,不记录凭证值。
离线再生成:`python3 scripts/bake_candidates.py`、`python3 scripts/bake_ledger.py`,读取 `/workspace/modelhub-adapt` 本地文件;候选 342(A 46、B 296),ledger 206 对及 28 个模型禁提项。
配置优先 build-config,失败使用 sample 并记录来源及原因。token 缺失只 hold。
## 快速开始
@@ -34,6 +42,7 @@ make build # 需 Docker + 可达平台 base;失败见 docs/build-not
```bash
export STRATEGY_ID=local-dev
export ALLOW_LIVE_SUBMIT=0 # 本地开发停交
python3 -m src.server
# → http://127.0.0.1:8080/health
```
@@ -71,6 +80,5 @@ tests/ # stdlib unittest
## 许可与边界
本仓库用于 Victor 本地/Gitea 交付骨架;任何 ModelHub 写操作须 Victor 本人在控制台或显式批准后执行。
Phase2:`make dry-run-pending` 只读生成 A→B 草稿,名称硬门与禁提过滤已接入;需 Victor 人工提交/确认。task/add 客户端默认不调用,必须显式注入并通过双闸门、队列及单次预算;测试仅 mock HTTP。
本仓库用于 Victor 本地/Gitea 交付。Victor 已批准上述队列闸门自动提交;不自动注册。
`make dry-run-pending` 仍仅生成草稿;测试全部 mock 平台接口。

View File

@@ -19,8 +19,8 @@
|-----------------|------------|------------------------------|------|
| 1(一梯队并列) | 壁砺166M | `Biren_166m` | ≈ 曦望 S2;社区成功量壁砺更高 |
| 1(一梯队并列) | 曦望 S2 | `Sunrise_pt-200-x1`(抓包证实) | 与壁砺并列,勿臆造先后「性能」 |
| 2 | 天垓150 | Iluvatar 天垓族(字符串待核) | 条件回退;饱和+唯一性碰撞多 |
| 3 | 天垓100 | (待核) | 一般不扇出 |
| 2 | 天垓150 | Iluvatar_bi-150 | 条件回退;饱和+唯一性碰撞多 |
| 3 | 天垓100 | Iluvatar_bi-100 | 一般不扇出 |
| 4 | 智铠 mrv-100 | `Iluvatar_mrv-100` | 我方唯一成功样本族;不当最快卡宣传 |
并列一梯队时:若 pending `try-next` 已给出相对顺序则保留;否则可按「壁砺 / 曦望」与池内缺口并列候选,报告标 `order_source=experience`。
@@ -44,6 +44,9 @@
- 结论:**API**,未公开文档但已抓包证实,依据 [autoconfig-capture-2026-10-16.md](/workspace/modelhub-adapt/logs/autoconfig-capture-2026-10-16.md)。
- `POST /api/adapt/task/build-config`,query 为 `gpuType` / `framework` / `taskType`,空 body;成功 JSON `code=0`,`data` 为 YAML。
- UI 无「自动配置」标签;选择框架触发 effect 填入 configParams。模型地址不进入该请求。
- 抓包观察 Authorization(值未记录);容器默认 Xc-Token 社区实践,可显式切换 Authorization,尚无 token 鉴权实证。
- 默认 dry-run 离线 sample;CONFIG_PARAMS_SOURCE=build-config 或 ALLOW_BUILD_CONFIG=1 且 token 可解析时调用 API,失败不可静默回退。
- 2026-10-17 主机只读探针确认 task/page 使用 Xc-Token 成功;容器继续使用 Xc-Token,Authorization Bearer 属于浏览器路径。
- 已核对 GPU 枚举:天垓100=Iluvatar_bi-100,天垓150=Iluvatar_bi-150,智铠100=Iluvatar_mrv-100,MTT S4000/摩尔线程=Mthreads_s4000;保留壁砺及曦望映射。枚举证据不代表官方性能排序。
- 默认 dry-run:`resolve_token()` 成功时优先 build-config;失败回退 sample,逐条报告 configParams_source、GAP 和 fallback reason。无 token → sample。CONFIG_PARAMS_SOURCE=sample 强制离线;=build-config 强制 API 路径(dry-run with_meta 仍可回退)。make build-config-dry 保持失败非零退出。
- API gpuType 仅映射壁砺/壁砺166M/Biren_166m 与曦望/曦望 S2/Sunrise_pt-200-x1;未知卡不请求 API,报告 GAP: gpu enum unmapped。
- 选策优化后仍为 experience_order,perf_order=unset;order_source=experience;无正式性能表。Section A 仅按 foothold 文本将 verified_reuse 提前,其余 A 保持文件序,再接 B。API foothold evidence: GAP (not wired)。
- **live 前必须有 token + API 调用实证**;sample 单独不是 production-ready。队列 ≥95 继续禁 live,本轮无 live。

View File

@@ -2,7 +2,7 @@
- 日期:2026-10-16 Asia/Shanghai
- 依据:产品规划 `agent-product-plan-2026-10-16.md` §1 + PRD `02-agent-contribution-prd.md`
- **冲突裁决:适配 W1 优先**(队列 ≥95 禁 live;本骨架默认 dry-run)
- **冲突裁决:适配 W1 优先**(队列 ≥95 禁 live;默认 live 开启,实时队列闸门控制)
## 核实等级
@@ -29,7 +29,7 @@
| **STRATEGY_ID 缺失** | **Phase1:进程启动失败(exit ≠0)** | 产品规划 Phase1 + PRD |
| 资源 | 约 100m–1 CPU / 256–512Mi | B |
| **发布智能体得分** | **0 分**(发布 ≠ 得分) | B |
| Live 默认 | **关**;双重闸门 `ALLOW_LIVE_SUBMIT=1` + `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT` | 本仓 |
| Live 默认 | **开**(Victor 已批准);`ALLOW_LIVE_SUBMIT=0` 停交,`LIVE_CONFIRM` 可选 | 本仓 |
| 积分 | 来自**成功适配**;智能体代提成功计同一成功分;公式/周报 **待平台核实** | B / D |
## 鉴权双轨(文档索引,Phase1 不代调)
@@ -41,7 +41,7 @@
## 禁止
- 代注册、代 live 提交、Makefile 含 register/submit/deploy/push
- 代注册、绕过实时队列闸门、Makefile 含 register/submit/deploy/push
- 日志/响应/镜像层写入 token 或 STRATEGY_ID **值**
- 把「智能体已发布」写入千分进度
@@ -51,12 +51,17 @@
|------|------|
| 异步验证队列 | 适配 W1 优先;≥95 全体停交(含智能体 live) |
| Codex | hold 期扩池/归因 > 智能体脚手架 |
| 本骨架 | 默认 dry-run;本轮无真实 `task/add` HTTP 客户端 |
| 本骨架 | 默认 live;每次 task/add 前实时读取 onlyMine 队列 |
## 闸门配置以实现为准
`src/gate.py` 只接受 `ALLOW_LIVE_SUBMIT=1` 与
`LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT`(去除首尾空白后精确匹配)。
PRD 的 `ALLOW_LIVE_SUBMIT=true` / `CONFIRM_LIVE=YES` 仅为需求示例,
本仓不支持这些值/别名,也不读取确认文件。开闸无客户端仍 blocked。
这只是本地逻辑闸门,不代表已获 live 授权或已校验队列与候选硬门。
`ALLOW_LIVE_SUBMIT` 默认 1,0 为 kill switch;`STRATEGY_ID` 必需。
Victor 确认已内置,`LIVE_CONFIRM` 仍接受但不必需。
`LIVE_QUEUE_THRESHOLD` 默认 95,只有完整分页的 onlyMine task/page 非终态数严格小于阈值才提交。
缺凭证、HTTP/平台错误、JSON 错误、分页不完整或无效配置均停交。
ASYNC_QUEUE_DEPTH / QUEUE_SNAPSHOT_JSON 仅为额外阻塞提示。
每周期最多 `LIVE_MAX_SUBMITS_PER_CYCLE`(默认 3)次;周期
`LIVE_CYCLE_SECONDS` 默认 600、最小 60 秒,通过 shutdown Event 等待。
候选 A verified_reuse 优先,禁提过滤、GPU enum、占位符检查及请求前持久化去重。
`LIVE_STATE_FILE` 默认 /tmp/live-submitted.json;保留该文件可跨周期/重启去重。
build-config 失败回退 sample 并记录原因;日志不记录响应正文或凭证。

View File

@@ -1,13 +1,14 @@
# Tag 文案草稿(仍默认 dry-run)— 未授权勿 push
# Tag 文案草稿(仍默认 dry-run)
- 建议 Tag 名(待 Victor 确认):`v0.2.0-phase2-dryrun` 或 `v0.2.0-submit-ready-dryrun`
- 相对 Phase1 `v0.1.0-phase1@0d0f3b6`:pending A 池 + 经验序排序、只读 task/page 探针、task/add 客户端仍双重闸门默认关、configParams sample 仅 dry-run
- **不含**:live 开闸、真实自动配置、开闸环境变量默认开启
- 推 Tag 条件:仅当 Victor **明确**说「先上线 dry-run 客户端」;产品当前:**先别 push Tag**
- 队列满 / configParams 缺口期间:**不开闸**
- 建议 Tag:`v0.2.0-phase2-dryrun`
- 基线:控制台当前 `v0.1.0-phase1` @ `0d0f3b6`
- 相对 Phase1:pending **Section A** + 经验序;只读 `task/page` 探针;`build-config` 客户端(默认离线 sample,opt-in 才调 API);`task/add` 双重闸门默认关
- **不含**:live 开闸、默认真实提交
- 推送:仅 Victor 本人;Agent 不代 push
- 队列≥95 / 无书面批准:**不开闸**
推送检查清单(人工):
1. `make test` 绿
2. ALLOW_LIVE_SUBMIT 默认非 1;LIVE_CONFIRM 未设
3. 控制台重建后日志仍为 dry-run / Phase2 格式
检查清单:
1. `make test` 42/42
2. `ALLOW_LIVE_SUBMIT` 默认非 1
3. 重建后 `/health` 200;`/status` token/strategy 布尔
4. 不代注册、不代 live

View File

@@ -0,0 +1,11 @@
# Live gate update — 2026-10-17
Queue snapshots accept integer or ASCII digit-string totals and derive model/GPU pairs from modelId and gpuType across all statuses. Nonterminal tasks count toward the queue gate. Invalid or incomplete snapshots stop submissions with ledger_rebuild_failed.
Token resolution prefers EXTERNAL_SERVICE_TOKEN, XC_TOKEN_FILE, XC_TOKEN, then MODELHUB_XC_TOKEN; empty and placeholder credentials are skipped. Missing credentials hold even with injected submit clients. Each allowed cycle logs only the selected source name. Xc-Token remains the authentication header.
Offline generators read local adapt evidence only. Input candidates: A=60, B=296; after pending and banned filters: A=46, B=296, total=342. Verified reuse A precedes other A, then B. Baked ledger: 206 pairs and 28 blocked models. Baked JSON ships through COPY src/; .dockerignore does not exclude src/data/*.json.
The ledger merges remote pairs, baked pairs, and persisted model|gpu pairs. Explicit do-not-resubmit models are blocked on every GPU; other models may use the next mapped, unsubmitted GPU in candidate order. Reservations are written before task/add, including failed attempts. Legacy model-only state files fail closed and require offline migration. Startup rebuild logs source counts; each prospective submission refreshes remote identities and queue count.
Validation: make test passes 59 tests using stdlib only. Added coverage for pair dedupe and restart persistence, baked blocks, next-GPU selection, missing tokens, token source logging, digit-string totals, invalid totals, and baked candidate ordering/filtering. SIGTERM uses a mocked HTTP server and real subprocess signal without network calls. Dockerfile FROM unchanged. No git commit, push, tag, live probe, build-config call, or live submission performed.

View File

@@ -0,0 +1,23 @@
"""Bake the local pending pool without network access."""
import json
import sys
from collections import Counter
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from src.pending_pool import build_ordered_candidates
from src.banned import banned_rule
def main():
rows = build_ordered_candidates('/workspace/modelhub-adapt/pending-submit.md', '/workspace/modelhub-adapt/pending-submit.txt')
print('input', dict(Counter(r['section'] for r in rows)))
rows = [dict(model_address=r['model_address'], section=r['section'], foothold_class=r['foothold_class'], try_next_gpus=r['try_next_gpus'], framework='vllm') for r in rows if not r['skip_reason'] and not banned_rule(r['model_address'])]
rows.sort(key=lambda r: (r['section'] != 'A', r['foothold_class'] != 'verified_reuse'))
output = Path(__file__).resolve().parents[1] / 'src/data/candidates.json'
output.parent.mkdir(exist_ok=True)
output.write_text(json.dumps(rows, ensure_ascii=False, indent=2) + '\n')
print('baked', dict(Counter(r['section'] for r in rows)), 'total', len(rows))
if __name__ == '__main__':
main()

39
scripts/bake_ledger.py Normal file
View File

@@ -0,0 +1,39 @@
"""Extract only model/GPU identities from local evidence; never copy raw fields."""
import json
import re
import sys
from pathlib import Path
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.pending_pool import load_do_not_resubmit
def extract(value, source, pairs):
if isinstance(value, dict):
model = value.get('modelId') or value.get('model') or value.get('model_address')
gpu = value.get('gpuType') or value.get('gpu_code') or value.get('gpu') or value.get('gpuLabel')
if isinstance(model, str) and re.fullmatch(r'[\w.-]+/[\w.\-/]+', model) and isinstance(gpu, str) and gpu in GPU_ENUMS:
pairs.setdefault((model_key(model), GPU_ENUMS[gpu]), source)
for child in value.values():
extract(child, source, pairs)
elif isinstance(value, list):
for child in value:
extract(child, source, pairs)
def main():
base = Path('/workspace/modelhub-adapt')
files = [base / 'logs/task-page-snapshot-2026-10-17.json', *sorted((base / 'logs').glob('submit-session-*.json')), *sorted((base / 'logs').glob('submit-batch-*.json'))]
pairs = {}
for path in files:
extract(json.loads(path.read_text()), path.name, pairs)
blocks = sorted(map(model_key, load_do_not_resubmit(base / 'do-not-resubmit.md')))
output = Path(__file__).resolve().parents[1] / 'src/data/submitted-ledger.json'
output.parent.mkdir(exist_ok=True)
output.write_text(json.dumps(dict(pairs=[dict(model=m, gpu=g, source=s) for (m,g),s in sorted(pairs.items())], blocked_models=blocks), indent=2) + '\n')
print('baked pairs', len(pairs), 'blocked models', len(blocks))
if __name__ == '__main__':
main()

View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Read-only Phase2 pending draft report. Offline by default; opt-in configuration generation only."""
"""Read-only Phase2 pending draft report. Token-resolvable build-config first; explicit sample fallback."""
import json
import os
import sys
@@ -10,7 +10,7 @@ from zoneinfo import ZoneInfo
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from src.pending_pool import build_ordered_candidates, input_path, experience_order
from src.config_params import build_config_params, config_params_source, SOURCE, GAP
from src.config_params import build_config_params_with_meta, SOURCE, GAP
from src.queue_depth import queue_depth
from src.task_add_client import build_payload
@@ -27,9 +27,9 @@ def _render(candidates, ts, src):
c.get('section', 'B') == os.environ['PENDING_SECTION'].strip().upper()]
depth = queue_depth()
lines = [f'# 代提候选 / 需 Victor 确认 — Dry-run — {ts} (Asia/Shanghai)', '',
'> 模式:dry-run;零网络写;需 Victor 人工提交/确认',
'> 模式:dry-run;仅配置生成 API;无任务提交;需 Victor 人工提交/确认',
'> 积分:待平台核实', '> perf_order=unset;order_source=experience(经验序 / 待平台核实)',
f'> configParams source: {SOURCE}', f'> configParams_source={config_params_source()}', f'> {GAP}', '',
f'> configParams source: {SOURCE}', '> API foothold evidence: GAP (not wired)', f'> {GAP}', '',
'## A→B 顺序草稿', '', f'输入:{src};候选条数:{len(candidates)}',
f'队列深度:{depth if depth is not None else "未知,需人工确认"}',
'队列≥95:禁 live;适配 W1 优先,不抢空位。', '']
@@ -45,8 +45,11 @@ def _render(candidates, ts, src):
f'过滤:{c.get("skip_reason") or "名称硬门未命中;元数据/唯一性待人工核实"}', ''])
lines.extend([f'未适配卡(仅 pending 缺口):{", ".join(gpus) or "未提供,需确认"}',
'perf_order=unset;order_source=experience;天垓100 一般不扇出', ''])
yaml, meta = build_config_params_with_meta(target)
lines.extend([f'foothold_class={c.get("foothold_class", "unknown")}; evidence_source={c.get("evidence_source", "GAP")}; no_download={c.get("no_download", False)}',
'; '.join(f'{key}={value}' for key, value in meta.items()), ''])
draft = build_payload(model, target, 'vllm', strategy_id='<STRATEGY_ID>',
config_params=build_config_params(target))
config_params=yaml)
# Display only; this truncated draft must never be submitted.
draft['configParams'] = draft['configParams'][:400] + '\n… [truncated; dry-run only]'
lines.extend(['build-config query: gpuType=<targetGpu>&framework=vllm&taskType=text-generation; POST empty body', 'POST 字段草稿(skipped 行禁止提交):', '```json', json.dumps(draft, ensure_ascii=False, indent=2), '```',

15
src/banned.py Normal file
View File

@@ -0,0 +1,15 @@
"""Case-insensitive live model exclusions."""
import re
RULES = (
('gpt2', r'gpt-?2|dialogpt|distilgpt2'),
('phi', r'phi[-_. ]?(?:3(?:\.5)?|4)(?:\D|$)'),
('nemotron-h', r'nemotron-h'), ('olmo2', r'olmo-?2'),
('lfm2', r'lfm2'), ('spark2.5', r'spark2_5|spark-x2\.5'),
('breeze', r'breeze'), ('gemma2', r'gemma-?2'),
('mistral-v0.1', r'mistral.*v0\.1'), ('tinyllama', r'tinyllama'),
)
def banned_rule(model):
return next((name for name, pattern in RULES if re.search(pattern, model, re.I)), None)

View File

@@ -1,31 +1,71 @@
"""Offline sample by default; explicit opt-in enables captured build-config API."""
"""Dry-run configuration: token-resolvable build-config first; failures fall back to sample with GAP."""
import os
from pathlib import Path
from src.build_config_client import build_config, BuildConfigError
from src.task_add_client import resolve_token
SOURCE = 'POST /api/adapt/task/build-config (discovered via capture); default templates/config_params.sample.yaml'
GAP = 'GAP: build-config API discovered via capture (autoconfig-capture-2026-10-16.md); live/实锤 needs token + API call and auth proof. Sample is dry-run draft only, NOT production-ready.'
SOURCE = 'POST /api/adapt/task/build-config (discovered via capture); fallback templates/config_params.sample.yaml'
GAP = 'GAP: build-config API discovered via capture; sample is dry-run draft only, NOT production-ready. API foothold evidence: GAP (not wired).'
GPU_ENUMS = {
'壁砺': 'Biren_166m', '壁砺166M': 'Biren_166m', 'Biren_166m': 'Biren_166m',
'曦望': 'Sunrise_pt-200-x1', '曦望 S2': 'Sunrise_pt-200-x1',
'Sunrise_pt-200-x1': 'Sunrise_pt-200-x1',
'天垓100': 'Iluvatar_bi-100', 'Iluvatar_bi-100': 'Iluvatar_bi-100',
'天垓150': 'Iluvatar_bi-150', 'Iluvatar_bi-150': 'Iluvatar_bi-150',
'智铠': 'Iluvatar_mrv-100', '智铠100': 'Iluvatar_mrv-100',
'智铠 mrv-100': 'Iluvatar_mrv-100', 'Iluvatar_mrv-100': 'Iluvatar_mrv-100',
'MTT S4000': 'Mthreads_s4000', '摩尔线程': 'Mthreads_s4000', 'Mthreads_s4000': 'Mthreads_s4000',
}
def config_params_source():
"""Requested source preference for dry-run.
- CONFIG_PARAMS_SOURCE=sample → force offline sample
- CONFIG_PARAMS_SOURCE=build-config → force API path (with soft fallback in with_meta)
- default (empty): prefer build-config when resolve_token() succeeds; else sample
- ALLOW_BUILD_CONFIG=1 remains a legacy opt-in synonym when source empty and token ok
"""
source = os.environ.get('CONFIG_PARAMS_SOURCE', '').strip()
if source not in ('', 'sample', 'build-config'):
raise BuildConfigError('unsupported CONFIG_PARAMS_SOURCE')
if source == 'build-config':
return 'build-config'
if os.environ.get('ALLOW_BUILD_CONFIG') == '1':
if source:
return source
try:
resolve_token()
except (ValueError, OSError):
return 'sample'
# Token resolvable → prefer build-config (kickoff select-opt default).
return 'build-config'
def build_config_params_with_meta(gpu_type=None, framework='vllm', task_type='text-generation'):
"""Return (YAML, metadata); API failures explicitly fall back for dry-run.
scripts/build_config_dry.py calls build_config directly and remains strict.
"""
source = config_params_source()
meta = {'configParams_source': 'sample'}
try:
resolve_token()
except (ValueError, OSError):
meta['auth'] = 'missing'
else:
meta['auth'] = 'ok'
gpu_enum = GPU_ENUMS.get(gpu_type)
if source == 'build-config' and gpu_enum is None:
meta['configParams_fallback_reason'] = 'GAP: gpu enum unmapped'
elif source == 'build-config':
try:
resolve_token()
except (ValueError, OSError):
pass
yaml = build_config(gpu_enum, framework, task_type)
except BuildConfigError as error:
meta['configParams_fallback_reason'] = 'GAP: ' + error.error_attribution
else:
return 'build-config'
return 'sample'
meta['configParams_source'] = 'build-config'
return yaml, meta
return (Path(__file__).resolve().parents[1] / 'templates/config_params.sample.yaml').read_text(), meta
def build_config_params(gpu_type=None, framework='vllm', task_type='text-generation'):
"""Return YAML; opted-in API failures propagate, never fall back to sample."""
if config_params_source() == 'build-config':
return build_config(gpu_type, framework, task_type)
return (Path(__file__).resolve().parents[1] / 'templates/config_params.sample.yaml').read_text()
"""Compatible YAML-only dry-run wrapper, including explicit fallback."""
return build_config_params_with_meta(gpu_type, framework, task_type)[0]

3661
src/data/candidates.json Normal file

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -1,123 +1,74 @@
"""Dual live-submit gate. Default: dry-run / blocked.
"""Live defaults on with baked-in approval. LIVE_CONFIRM remains optional.
Live is allowed only when ALL of:
- ALLOW_LIVE_SUBMIT=1
- LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT
- STRATEGY_ID present
Phase2 client exists but requires explicit injection. No automatic wiring.
Queue ≥95 blocks; at most one injected submit per process.
Environment hints only block; a fresh authenticated API snapshot authorizes each add.
"""
from __future__ import annotations
import os
import threading
from .queue_depth import queue_depth
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional, Protocol
from .strategy import strategy_id_present
_submit_count = 0
_submit_lock = threading.Lock()
LIVE_CONFIRM_VALUE = "I_CONFIRM_LIVE_SUBMIT"
LIVE_CONFIRM_VALUE = 'I_CONFIRM_LIVE_SUBMIT'
class SubmitClient(Protocol):
"""Minimal interface for a submit client (injected fake in tests)."""
def live_enabled():
return os.environ.get('ALLOW_LIVE_SUBMIT', '1').strip() == '1'
def submit(self, payload: Dict[str, Any]) -> Dict[str, Any]:
...
def queue_threshold():
value = int(os.environ.get('LIVE_QUEUE_THRESHOLD', '95'))
if value < 1:
raise ValueError('invalid threshold')
return value
@dataclass(frozen=True)
class GateDecision:
mode: str # "dry-run" | "live" | "blocked"
mode: str
allowed: bool
reason: str
def _allow_live_flag() -> bool:
return os.environ.get("ALLOW_LIVE_SUBMIT", "0").strip() == "1"
def _live_confirm_ok() -> bool:
return os.environ.get("LIVE_CONFIRM", "").strip() == LIVE_CONFIRM_VALUE
def evaluate_gate() -> GateDecision:
"""Evaluate whether live submit is permitted."""
def evaluate_gate():
if not live_enabled():
return GateDecision('dry-run', False, 'kill_switch')
if not strategy_id_present():
return GateDecision(
mode="blocked",
allowed=False,
reason="STRATEGY_ID missing or empty",
)
if not _allow_live_flag():
return GateDecision(
mode="dry-run",
allowed=False,
reason="ALLOW_LIVE_SUBMIT is not 1 (default dry-run)",
)
if not _live_confirm_ok():
return GateDecision(
mode="blocked",
allowed=False,
reason=f"LIVE_CONFIRM must be exactly {LIVE_CONFIRM_VALUE!r}",
)
depth = queue_depth()
if depth is None and ("ASYNC_QUEUE_DEPTH" in os.environ or "QUEUE_SNAPSHOT_JSON" in os.environ):
return GateDecision("blocked", False, "queue depth invalid/unknown; human confirmation required")
if depth is not None and depth >= 95:
return GateDecision("blocked", False, "queue ≥95: 禁 live")
return GateDecision(
mode="live",
allowed=True,
reason="dual gate open (ALLOW_LIVE_SUBMIT=1 and LIVE_CONFIRM set)",
)
return GateDecision('blocked', False, 'STRATEGY_ID missing or empty')
try:
threshold = queue_threshold()
except ValueError:
return GateDecision('blocked', False, 'invalid_threshold')
# Evaluate each configured hint independently; neither can override the other.
for key in ('ASYNC_QUEUE_DEPTH', 'QUEUE_SNAPSHOT_JSON'):
if key not in os.environ:
continue
if key == 'ASYNC_QUEUE_DEPTH':
try:
depth = int(os.environ[key])
except ValueError:
depth = None
else:
import json
from pathlib import Path
try:
data = json.loads(Path(os.environ[key]).read_text())
value = next(data[k] for k in ('depth', 'queue_depth', 'async_queue_depth', 'ASYNC_QUEUE_DEPTH') if k in data)
depth = None if isinstance(value, bool) else int(str(value))
except (OSError, ValueError, TypeError, StopIteration):
depth = None
if depth is None or depth < 0:
return GateDecision('blocked', False, 'queue_hint_invalid')
if depth >= threshold:
return GateDecision('blocked', False, f'queue>={threshold} hint_count={depth}')
return GateDecision('live', True, 'live_enabled')
def attempt_live_submit(
payload: Dict[str, Any],
client: Optional[SubmitClient] = None,
) -> Dict[str, Any]:
"""Attempt a live submit only if the gate opens and a client is injected.
Phase2 TaskAddClient must be explicitly injected and dual-gated.
Tests must use fake clients or mock HTTP only.
"""
global _submit_count
decision = evaluate_gate()
if not decision.allowed:
return {
"ok": False,
"mode": decision.mode,
"reason": decision.reason,
"submitted": False,
}
def attempt_live_submit(payload, client=None, queue_reader=None):
"""Compatibility entry point; delegates to the same fresh queue cycle."""
if client is None:
return {
"ok": False,
"mode": "blocked",
"reason": (
"Phase2 task/add client exists; explicit client injection required; "
"no real request without injection and dual gate"
),
"submitted": False,
}
with _submit_lock:
if _submit_count >= 1:
return dict(ok=False, mode="blocked", submitted=False,
reason="process limit: at most 1 real add")
# Reserve before invocation: a transport failure may still have created a task.
_submit_count += 1
result = client.submit(payload)
return {
"ok": True,
"mode": "live",
"reason": decision.reason,
"submitted": True,
"result": result,
}
return dict(ok=False, submitted=False, mode='blocked', reason='no real request without explicit client')
from .live_cycle import run_cycle
decisions = run_cycle([dict(model_address=payload.get('modelAddress'), targetGpu=payload.get('targetGpu'),
framework=payload.get('framework', 'vllm'), configParams=payload.get('configParams'))],
client=client, queue_reader=queue_reader)
result = decisions[-1]
return dict(result, ok=result['action'] == 'submitted', submitted=result['action'] == 'submitted',
mode='live' if result['action'] == 'submitted' else 'blocked')

35
src/ledger.py Normal file
View File

@@ -0,0 +1,35 @@
"""Pair reservations and explicit model blocks; no credentials or response bodies."""
import json
from pathlib import Path
from .live_queue import model_key, read_queue
DATA = Path(__file__).parent / 'data/submitted-ledger.json'
def load_baked(path=DATA):
data = json.loads(Path(path).read_text())
return ({(model_key(r['model']), r['gpu']) for r in data['pairs']},
{model_key(m) for m in data['blocked_models']})
def load_persisted(path):
rows = json.loads(path.read_text()) if path.exists() else []
if not isinstance(rows, list) or any(not isinstance(r, str) or r.count('|') != 1 for r in rows):
raise ValueError('Invalid pair ledger')
return {(model_key(m), g) for m, g in (r.split('|') for r in rows) if m and g}
def rebuild(state, reader=None):
count, reason, remote = (reader or read_queue)()
if type(count) is not int or count < 0:
raise ValueError('ledger_rebuild_failed')
baked, blocked = load_baked()
persisted = load_persisted(state)
return count, remote | baked | persisted, blocked, (len(remote), len(baked), len(persisted))
def persist(path, pairs):
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(path.name + '.tmp')
temporary.write_text(json.dumps(sorted(m + '|' + g for m, g in pairs)))
temporary.replace(path)

143
src/live_cycle.py Normal file
View File

@@ -0,0 +1,143 @@
"""Queue-gated live cycles; all remote collaborators are injectable."""
import json
import os
import sys
import threading
from pathlib import Path
from .banned import banned_rule
from .config_params import GPU_ENUMS, build_config_params_with_meta
from .gate import evaluate_gate, queue_threshold
from .live_queue import read_queue, model_key
from .pending_pool import build_ordered_candidates
from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source
from .ledger import load_baked, load_persisted, persist, rebuild
_submitted = set()
_cycle_lock = threading.Lock()
def run_cycle(candidates=None, client=None, queue_reader=None, now=None):
with _cycle_lock:
return _run_cycle(candidates, client, queue_reader, now)
def _run_cycle(candidates, client, queue_reader, now):
decisions = []
def emit(action, reason, **fields):
decision = dict(action=action, reason=reason, **fields)
decisions.append(decision)
print('[live] ' + action + ' ' + ' '.join(f'{k}={str(v).replace(chr(10), " ").replace(chr(13), " ")}' for k, v in dict(fields, reason=reason).items()), file=sys.stderr)
gate = evaluate_gate()
if not gate.allowed:
emit('hold', gate.reason)
return decisions
try:
_, source = resolve_token_with_source()
except (ValueError, OSError, UnicodeError):
emit('hold', 'missing/invalid_token')
return decisions
print('[live] token source=' + source, file=sys.stderr)
try:
cap = int(os.environ.get('LIVE_MAX_SUBMITS_PER_CYCLE', '3'))
if cap < 1:
raise ValueError
except ValueError:
emit('hold', 'invalid_cycle_cap')
return decisions
if candidates is None:
try:
if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
candidates = build_ordered_candidates()
else:
candidates = json.loads((Path(__file__).parent / 'data/candidates.json').read_text())
except (OSError, ValueError, IndexError):
emit('hold', 'candidate_read_failed')
return decisions
candidates = sorted(candidates, key=lambda c: (c.get('section', 'B') != 'A', c.get('foothold_class') != 'verified_reuse'))
if not candidates:
emit('hold', 'no_candidates')
return decisions
state = Path(os.environ.get('LIVE_STATE_FILE', '/tmp/live-submitted.json'))
try:
persisted = load_persisted(state) | _submitted
baked, blocked = load_baked()
known = persisted | baked
except (OSError, ValueError, KeyError, TypeError):
emit('hold', 'state_read_failed')
return decisions
reader = queue_reader or read_queue
client = client if client is not None else TaskAddClient()
attempts = 0
for candidate in candidates:
model = candidate.get('model_address') or candidate.get('modelAddress') or ''
rule = banned_rule(model + ' ' + str(candidate.get('model_name', candidate.get('modelName', ''))))
if rule:
emit('skip', 'banned:' + rule, model=model)
continue
if not model or '<' in model or '>' in model:
emit('skip', 'placeholder')
continue
key = model_key(model)
if key in blocked:
emit('skip', 'do-not-resubmit', model=model)
continue
if candidate.get('skip_reason'):
emit('skip', 'pending_filter', model=model)
continue
if attempts >= cap:
emit('hold', 'cycle_cap')
break
choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')]
gpu = next((g for g in choices if g in GPU_ENUMS and (key, GPU_ENUMS[g]) not in known), None)
if gpu is None:
emit('skip', 'all_gpus_done_or_unmapped', model=model)
continue
# Rebuild remote identities before config calls and every submission.
try:
count, why, remote = reader()
if type(count) is not int or count < 0:
raise ValueError
known.update(remote)
except Exception:
emit('hold', 'ledger_rebuild_failed')
break
gpu = next((g for g in choices if g in GPU_ENUMS and (key, GPU_ENUMS[g]) not in known), None)
if gpu is None:
emit('skip', 'all_gpus_done_or_unmapped', model=model)
continue
if count >= queue_threshold():
emit('hold', f'queue>={queue_threshold()}', count=count)
break
framework = candidate.get('framework') or 'vllm'
try:
yaml, meta = build_config_params_with_meta(gpu, framework)
except Exception:
emit('skip', 'config_failed', model=model)
continue
payload = build_payload(model, GPU_ENUMS[gpu], framework, os.environ['STRATEGY_ID'].strip(), yaml)
if any(not value or '<' in str(value) or '>' in str(value) for value in payload.values()):
emit('skip', 'placeholder', model=model)
continue
pair = (key, GPU_ENUMS[gpu])
known.add(pair)
persisted.add(pair)
_submitted.add(pair)
try:
persist(state, persisted)
except OSError:
emit('hold', 'state_write_failed')
break
attempts += 1
try:
result = client.submit(payload)
status, code = result.get('http_status'), result.get('platform_code')
status = status if type(status) is int else None
code = code if type(code) is int else None
reason = 'task_add_response' if status is not None else 'transport_error'
except Exception:
status, code, reason = None, None, 'submit_error'
emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
http=status, code=code, **meta)
return decisions

68
src/live_queue.py Normal file
View File

@@ -0,0 +1,68 @@
"""Fresh authenticated queue snapshots. Never expose response bodies or credentials."""
import json
import os
import urllib.request
from .task_add_client import resolve_token
TERMINAL = {'success', 'failed', 'fail', 'error', 'cancelled', 'canceled'}
def model_key(value):
value = str(value or '').strip().lower().rstrip('/')
for prefix in ('https://huggingface.co/', 'http://huggingface.co/', 'https://modelscope.cn/models/'):
if value.startswith(prefix):
value = value[len(prefix):]
return value
def valid_total(value):
return (type(value) is int and value >= 0) or (isinstance(value, str) and value.isascii() and value.isdigit())
def read_queue():
"""Return (count, reason, model identities); count None means fail closed."""
try:
token = resolve_token()
except (ValueError, OSError, UnicodeError):
return None, 'missing/invalid_token', set()
count, models, seen, total = 0, set(), 0, None
for page in range(1, 21):
try:
request = urllib.request.Request(
os.environ.get('MODELHUB_API_BASE', 'https://modelhub.org.cn').rstrip('/')
+ f'/api/adapt/task/page?onlyMine=true&pageSize=100&current={page}',
headers={'Xc-Token': token, 'Accept': 'application/json'}, method='GET')
with urllib.request.urlopen(request, timeout=30) as response:
if not 200 <= response.status < 300:
return None, 'http_error', set()
body = json.loads(response.read())
except (ValueError, UnicodeError):
return None, 'malformed_json', set()
except OSError:
return None, 'transport/http_error', set()
if not isinstance(body, dict) or type(body.get('code')) not in (int, str) or str(body['code']) != '0':
return None, 'platform_error', set()
data = body.get('data')
if not isinstance(data, dict) or not valid_total(data.get('total')):
return None, 'malformed_data', set()
data['total'] = int(data['total'])
records = data.get('records')
if not isinstance(records, list) or any(not isinstance(r, dict) for r in records):
return None, 'malformed_records', set()
if total is not None and total != data['total']:
return None, 'pagination_total_changed', set()
total = data['total']
seen += len(records)
if seen > total:
return None, 'pagination_inconsistent', set()
for record in records:
count += str(record.get('status', '')).strip().lower() not in TERMINAL
model, gpu = model_key(record.get('modelId')), record.get('gpuType')
if model and isinstance(gpu, str) and gpu:
models.add((model, gpu))
if seen == total:
# Do not retain arbitrary response fields, which may echo credentials.
return count, 'ok', {m for m in models if all(token.lower() not in part.lower() for part in m)}
if not records:
break
return None, 'incomplete_pagination', set()

View File

@@ -42,8 +42,31 @@ def experience_order(gpus):
return sorted(dict.fromkeys(gpus), key=tier)
def _candidate(model, section, gpu_text, notes):
def classify_foothold(text):
"""Classify literal pending evidence; mixed markers prefer verified."""
for label, markers in (
('verified_reuse', ('已验证成功', '验证通过')),
('in_flight', ('正在验证', '验证排队中', '验证进行中')),
('failed_foothold', ('验证失败',)),
):
if any(marker in text for marker in markers):
return label
return 'unknown'
def has_no_download_marker(text):
"""True only for explicit 免下载/已下载; never inferred from verified alone."""
return any(marker in text for marker in ('免下载', '已下载'))
def _candidate(model, section, gpu_text, notes, foothold_text=''):
evidence_text = foothold_text if section == 'A' else ''
foothold_class = classify_foothold(evidence_text) if section == 'A' else 'unknown'
no_download = has_no_download_marker(evidence_text + ' ' + notes)
evidence = 'pending-foothold' if (foothold_class != 'unknown' or no_download) else 'GAP'
return dict(model_address=model, section=section,
foothold_text=foothold_text, foothold_class=foothold_class,
evidence_source=evidence, no_download=no_download,
try_next_gpus=experience_order(re.findall(GPUS, gpu_text)),
perf_order="unset", order_source="experience",
notes=notes, comment=notes, skip_reason=None,
@@ -62,7 +85,7 @@ def parse_pending_md(path):
cols = [s.strip() for s in line.strip('|').split('|')]
model = re.search(r'`([^`]+)`', cols[1])
if model:
rows.append(_candidate(model[1], section, cols[3] if section == 'A' else cols[-1], cols[-1]))
rows.append(_candidate(model[1], section, cols[3] if section == 'A' else cols[-1], cols[-1], cols[2] if section == 'A' else ''))
return rows
@@ -76,7 +99,7 @@ def parse_pending_txt(path):
match = re.match(r'#([AB])', comment)
section = match[1] if match else 'B'
gpu_text = re.split(r'(?:still\s+)?try\s*:', comment, maxsplit=1)[-1] if 'try:' in comment else (comment if section == 'B' else '')
row = _candidate(parts[0].strip(), section, gpu_text, comment)
row = _candidate(parts[0].strip(), section, gpu_text, comment, comment if section == 'A' else '')
if not match:
row.update(needs_human=True, status='needs_human')
rows.append(row)
@@ -103,4 +126,4 @@ def build_ordered_candidates(md_path=None, txt_path=None, dnr_path=None, banned_
# Metadata gates (weights, context, riskTag, uniqueness) remain unverified.
row['needs_human'] = not bool(reason)
row['status'] = 'skipped' if reason else 'needs_human'
return sorted(rows, key=lambda r: r['section'] != 'A')
return sorted(rows, key=lambda r: (r['section'] != 'A', not (r['foothold_class'] == 'verified_reuse' or r.get('no_download'))))

View File

@@ -17,6 +17,8 @@ from typing import Optional
from .health import handle_get
from .strategy import require_strategy_id
from .gate import live_enabled
from .live_cycle import run_cycle
_shutdown = threading.Event()
_httpd: Optional[ThreadingHTTPServer] = None
@@ -56,6 +58,27 @@ def _on_sigterm(signum: int, frame) -> None: # noqa: ARG001
threading.Thread(target=httpd.shutdown, daemon=True).start()
def _live_worker():
from pathlib import Path
from .ledger import rebuild
try:
_, _, _, sizes = rebuild(Path(os.environ.get('LIVE_STATE_FILE', '/tmp/live-submitted.json')))
sys.stderr.write('[live] ledger rebuilt remote=%d baked=%d persisted=%d\n' % sizes)
except Exception:
sys.stderr.write('[live] hold reason=ledger_rebuild_failed\n')
try:
interval = max(60, int(os.environ.get('LIVE_CYCLE_SECONDS', '600')))
except ValueError:
interval = 600
while not _shutdown.is_set():
try:
run_cycle()
except Exception:
sys.stderr.write('[live] hold reason=cycle_error\n')
if _shutdown.wait(interval):
break
def main() -> None:
global _httpd
@@ -71,6 +94,9 @@ def main() -> None:
httpd.daemon_threads = True
_httpd = httpd
if live_enabled():
threading.Thread(target=_live_worker, daemon=True).start()
sys.stderr.write(f"Listening on {host}:{port} (GET /health, GET /status)\n")
try:
httpd.serve_forever(poll_interval=0.5)

View File

@@ -12,10 +12,10 @@ from typing import Optional
# Community alias order (level C); we only check presence, never echo values.
_TOKEN_ENV_KEYS = (
"EXTERNAL_SERVICE_TOKEN",
"XC_TOKEN_FILE",
"XC_TOKEN",
"MODELHUB_XC_TOKEN",
"EXTERNAL_SERVICE_TOKEN",
)
@@ -62,20 +62,9 @@ def _token_from_file(path: str) -> bool:
def external_service_token_present() -> bool:
"""Whether any known token env / file alias is present (boolean only)."""
for key in _TOKEN_ENV_KEYS:
raw = os.environ.get(key)
if raw is None:
continue
value = raw.strip()
if not value:
continue
if key == "XC_TOKEN_FILE":
if _token_from_file(value):
return True
continue
# Reject obvious placeholders (community practice, level C)
lowered = value.lower()
if lowered in ("tmp", "placeholder", "changeme", "xxx"):
continue
from .task_add_client import resolve_token_with_source
try:
resolve_token_with_source()
return True
return False
except (ValueError, OSError, UnicodeError):
return False

View File

@@ -19,13 +19,22 @@ def build_payload(model_address, target_gpu='<TARGET_GPU>', framework='<FRAMEWOR
configParams=config_params)
def resolve_token_with_source():
for name in ('EXTERNAL_SERVICE_TOKEN', 'XC_TOKEN_FILE', 'XC_TOKEN', 'MODELHUB_XC_TOKEN'):
raw = os.environ.get(name, '').strip()
if name == 'XC_TOKEN_FILE' and raw:
try:
raw = Path(raw).read_text().strip()
except (OSError, UnicodeError):
continue
if raw and '<' not in raw and '>' not in raw and not re.search(
r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp|xxx)(?:$|[^a-z0-9])', raw):
return raw, name
raise ValueError('Missing or invalid service credential')
def resolve_token():
file = os.environ.get('XC_TOKEN_FILE')
token = Path(file).read_text().strip() if file else next(
(os.environ[k].strip() for k in ('XC_TOKEN', 'MODELHUB_XC_TOKEN', 'EXTERNAL_SERVICE_TOKEN') if os.environ.get(k)), '')
if not token or re.search(r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp)(?:$|[^a-z0-9])', token):
raise ValueError('Missing or invalid service credential')
return token
return resolve_token_with_source()[0]
class TaskAddClient:

View File

@@ -4,7 +4,7 @@ import unittest
import os
from unittest.mock import patch
from pathlib import Path
from src.config_params import build_config_params, config_params_source, GAP
from src.config_params import build_config_params, config_params_source, build_config_params_with_meta, GAP
from src.build_config_client import BuildConfigError
from scripts.dry_run import _render
@@ -16,32 +16,68 @@ class TestConfigParams(unittest.TestCase):
self.addCleanup(env.stop)
def test_source_selection_and_failure(self):
with patch("urllib.request.urlopen") as network:
self.assertEqual(config_params_source(), "sample")
build_config_params()
network.assert_not_called()
self.assertIn("discovered via capture", GAP)
with patch.dict(os.environ, {"CONFIG_PARAMS_SOURCE": "build-config"}):
with self.assertRaisesRegex(BuildConfigError, "missing/invalid credential"):
build_config_params("gpu")
with patch.dict(os.environ, {"ALLOW_BUILD_CONFIG": "1", "XC_TOKEN": "test-secret"}), patch("src.config_params.build_config", return_value="api: yaml") as client:
self.assertEqual(config_params_source(), "build-config")
self.assertEqual(build_config_params("gpu"), "api: yaml")
client.assert_called_once_with("gpu", "vllm", "text-generation")
with patch.dict(os.environ, {"CONFIG_PARAMS_SOURCE": "build-config"}), patch("src.config_params.build_config", side_effect=BuildConfigError("transport/timeout")):
with self.assertRaisesRegex(BuildConfigError, "transport/timeout"):
build_config_params("gpu")
# No token → sample, no network
with patch('src.config_params.resolve_token', side_effect=ValueError), patch('src.config_params.build_config') as client:
self.assertEqual(config_params_source(), 'sample')
yaml, meta = build_config_params_with_meta('壁砺')
self.assertEqual(meta['auth'], 'missing')
self.assertEqual(meta['configParams_source'], 'sample')
client.assert_not_called()
# Token alone → prefer build-config (kickoff default)
for gpu, enum in [('壁砺', 'Biren_166m'), ('壁砺166M', 'Biren_166m'),
('Biren_166m', 'Biren_166m'), ('曦望', 'Sunrise_pt-200-x1'),
('曦望 S2', 'Sunrise_pt-200-x1'), ('Sunrise_pt-200-x1', 'Sunrise_pt-200-x1')]:
with self.subTest(gpu=gpu), patch('src.config_params.resolve_token', return_value='synthetic'), patch('src.config_params.build_config', return_value='api: yaml') as client:
self.assertEqual(config_params_source(), 'build-config')
yaml, meta = build_config_params_with_meta(gpu)
self.assertEqual(yaml, 'api: yaml')
self.assertEqual(meta['configParams_source'], 'build-config')
self.assertEqual(meta['auth'], 'ok')
client.assert_called_once_with(enum, 'vllm', 'text-generation')
# Failure → sample + GAP reason
with patch('src.config_params.resolve_token', return_value='synthetic'), patch('src.config_params.build_config', side_effect=BuildConfigError('transport/timeout')):
yaml, meta = build_config_params_with_meta('壁砺')
self.assertEqual(meta['configParams_source'], 'sample')
self.assertEqual(meta['configParams_fallback_reason'], 'GAP: transport/timeout')
self.assertIn('max_model_len: 4096', yaml)
# Forced build-config also soft-falls back in with_meta
with patch.dict(os.environ, {'CONFIG_PARAMS_SOURCE': 'build-config'}), patch('src.config_params.resolve_token', return_value='synthetic'), patch('src.config_params.build_config', side_effect=BuildConfigError('transport/timeout')):
yaml, meta = build_config_params_with_meta('壁砺')
self.assertEqual(meta['configParams_fallback_reason'], 'GAP: transport/timeout')
# Unmapped GPU → no API call
for gpu in ('MetaX', 'MetaX_c-500', '<GPU>', None):
with patch('src.config_params.resolve_token', return_value='synthetic'), patch('src.config_params.build_config') as client:
_, meta = build_config_params_with_meta(gpu)
self.assertIn('GAP: gpu enum unmapped', meta['configParams_fallback_reason'])
client.assert_not_called()
# Force sample even with token
with patch.dict(os.environ, {'CONFIG_PARAMS_SOURCE': 'sample'}), patch('src.config_params.resolve_token', return_value='synthetic'), patch('src.config_params.build_config') as client:
self.assertEqual(config_params_source(), 'sample')
build_config_params('壁砺')
client.assert_not_called()
def test_report_api_failure_metadata(self):
with patch('src.config_params.resolve_token', return_value='synthetic'), patch('src.config_params.build_config', side_effect=BuildConfigError('transport/timeout')):
report = _render([{'model_address': 'org/a', 'try_next_gpus': ['壁砺'],
'foothold_class': 'verified_reuse', 'evidence_source': 'pending-foothold',
'no_download': False}], 'test', 'fixture')
for value in ('foothold_class=verified_reuse', 'evidence_source=pending-foothold', 'no_download=False',
'configParams_source=sample', 'configParams_fallback_reason=GAP: transport/timeout'):
self.assertIn(value, report)
def test_official_sample_exact(self):
scrape = Path('/workspace/modelhub-adapt/logs/api-docs-scrape-2026-10-16.md')
if scrape.is_file():
body = json.loads(re.search(r'```json\n(.*?)\n```', scrape.read_text(), re.S)[1])
self.assertEqual(build_config_params(), body['configParams'])
self.assertIn('max_model_len: 4096', build_config_params())
self.assertIn('framework: vllm', build_config_params())
with patch('src.config_params.resolve_token', side_effect=ValueError):
self.assertEqual(build_config_params(), body['configParams'])
with patch('src.config_params.resolve_token', side_effect=ValueError):
self.assertIn('max_model_len: 4096', build_config_params())
self.assertIn('framework: vllm', build_config_params())
self.assertIn('NOT production-ready', GAP)
def test_report_draft_and_gap(self):
report = _render([{'model_address': 'org/a', 'try_next_gpus': ['壁砺']}], 'test', 'fixture')
for value in ('代提候选 / 需 Victor 确认', 'configParams source', 'configParams_source=sample', 'GAP:', 'truncated', 'max_model_len: 4096', 'perf_order=unset', 'order_source=experience'):
with patch('src.config_params.resolve_token', side_effect=ValueError):
report = _render([{'model_address': 'org/a', 'try_next_gpus': ['壁砺']}], 'test', 'fixture')
for value in ('代提候选 / 需 Victor 确认', 'configParams source', 'configParams_source=sample', 'foothold_class=unknown', 'evidence_source=GAP', 'no_download=False', 'GAP:', 'truncated', 'max_model_len: 4096', 'perf_order=unset', 'order_source=experience'):
self.assertIn(value, report)

View File

@@ -1,159 +1,44 @@
"""Gate defaults to dry-run / blocked; live only with dual confirm + STRATEGY_ID."""
from __future__ import annotations
"""Default live approval, kill switch, strategy and conservative hint gates."""
import os
import tempfile
import unittest
from unittest import mock
from src.gate import LIVE_CONFIRM_VALUE, attempt_live_submit, evaluate_gate
class FakeClient:
def __init__(self) -> None:
self.calls = []
def submit(self, payload):
self.calls.append(payload)
return {"task_id": "fake-1"}
from pathlib import Path
from unittest.mock import patch
from src.gate import evaluate_gate, attempt_live_submit, LIVE_CONFIRM_VALUE
class TestGate(unittest.TestCase):
def setUp(self):
self.counter = mock.patch('src.gate._submit_count', 0)
self.counter.start()
self.addCleanup(self.counter.stop)
self.env = mock.patch.dict(os.environ, {}, clear=True)
self.env.start()
self.addCleanup(self.env.stop)
env = patch.dict(os.environ, {'STRATEGY_ID': 'sid'}, clear=True)
env.start()
self.addCleanup(env.stop)
def test_queue_and_process_limit(self):
client = FakeClient()
with mock.patch.dict(os.environ, {
'STRATEGY_ID': 'sid', 'ALLOW_LIVE_SUBMIT': '1',
'LIVE_CONFIRM': LIVE_CONFIRM_VALUE, 'ASYNC_QUEUE_DEPTH': '95'
}):
self.assertIn('≥95', attempt_live_submit({}, client)['reason'])
self.assertEqual(client.calls, [])
os.environ['ASYNC_QUEUE_DEPTH'] = '94'
self.assertTrue(attempt_live_submit({}, client)['submitted'])
self.assertFalse(attempt_live_submit({}, client)['submitted'])
self.assertEqual(len(client.calls), 1)
def test_default_live_and_optional_confirmation(self):
self.assertTrue(evaluate_gate().allowed)
for value in ('', 'YES', LIVE_CONFIRM_VALUE):
with patch.dict(os.environ, {'LIVE_CONFIRM': value}):
self.assertTrue(evaluate_gate().allowed)
def _clear_gate_env(self):
return mock.patch.dict(
os.environ,
{
"STRATEGY_ID": "sid-1",
"ALLOW_LIVE_SUBMIT": "0",
"LIVE_CONFIRM": "",
},
clear=False,
)
def test_kill_switch_and_missing_strategy(self):
for env in ({'ALLOW_LIVE_SUBMIT': '0'}, {'ALLOW_LIVE_SUBMIT': 'true'}, {'STRATEGY_ID': ' '}):
with patch.dict(os.environ, env):
self.assertFalse(evaluate_gate().allowed)
def test_default_dry_run(self) -> None:
with self._clear_gate_env():
d = evaluate_gate()
self.assertFalse(d.allowed)
self.assertEqual(d.mode, "dry-run")
def test_hints_and_threshold(self):
for value in ('95', 'invalid', '-1'):
with patch.dict(os.environ, {'ASYNC_QUEUE_DEPTH': value}):
self.assertFalse(evaluate_gate().allowed)
with patch.dict(os.environ, {'ASYNC_QUEUE_DEPTH': '94'}):
self.assertTrue(evaluate_gate().allowed)
with patch.dict(os.environ, {'LIVE_QUEUE_THRESHOLD': 'bad'}):
self.assertFalse(evaluate_gate().allowed)
with tempfile.TemporaryDirectory() as d:
p = Path(d) / 'hint.json'
p.write_text('{"depth":95}')
with patch.dict(os.environ, {'QUEUE_SNAPSHOT_JSON': str(p), 'ASYNC_QUEUE_DEPTH': '1'}):
self.assertFalse(evaluate_gate().allowed)
def test_missing_strategy_blocked(self) -> None:
env = {
"ALLOW_LIVE_SUBMIT": "1",
"LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
}
# Remove STRATEGY_ID
base = {k: v for k, v in os.environ.items() if k != "STRATEGY_ID"}
base.update(env)
with mock.patch.dict(os.environ, base, clear=True):
d = evaluate_gate()
self.assertFalse(d.allowed)
self.assertEqual(d.mode, "blocked")
def test_allow_without_confirm_blocked(self) -> None:
with mock.patch.dict(
os.environ,
{
"STRATEGY_ID": "sid-1",
"ALLOW_LIVE_SUBMIT": "1",
"LIVE_CONFIRM": "YES",
},
clear=False,
):
d = evaluate_gate()
self.assertFalse(d.allowed)
self.assertEqual(d.mode, "blocked")
def test_dual_gate_opens(self) -> None:
with mock.patch.dict(
os.environ,
{
"STRATEGY_ID": "sid-1",
"ALLOW_LIVE_SUBMIT": "1",
"LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
},
clear=False,
):
d = evaluate_gate()
self.assertTrue(d.allowed)
self.assertEqual(d.mode, "live")
def test_open_gate_without_client_still_blocked(self) -> None:
"""Phase2: an open gate never constructs a network client."""
with mock.patch.dict(
os.environ,
{
"STRATEGY_ID": "sid-1",
"ALLOW_LIVE_SUBMIT": "1",
"LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
},
clear=False,
):
result = attempt_live_submit({"modelAddress": "x/y"})
self.assertFalse(result["submitted"])
self.assertIn("no real", result["reason"].lower())
self.assertEqual(result["mode"], "blocked")
self.assertFalse(result["ok"])
def test_prd_example_and_whitespace_id_never_submit(self) -> None:
cases = [
{"ALLOW_LIVE_SUBMIT": "true", "LIVE_CONFIRM": LIVE_CONFIRM_VALUE},
{"ALLOW_LIVE_SUBMIT": "1", "LIVE_CONFIRM": "", "CONFIRM_LIVE": "YES"},
{"ALLOW_LIVE_SUBMIT": "1", "LIVE_CONFIRM": LIVE_CONFIRM_VALUE, "STRATEGY_ID": " "},
]
for overrides in cases:
with self.subTest(overrides=overrides):
client = FakeClient()
env = {"STRATEGY_ID": "sid-test", **overrides}
with mock.patch.dict(os.environ, env, clear=True):
result = attempt_live_submit({}, client=client)
self.assertFalse(result["submitted"])
self.assertFalse(result["ok"])
self.assertEqual(client.calls, [])
self.assertTrue(result["reason"])
def test_injected_fake_client(self) -> None:
client = FakeClient()
with mock.patch.dict(
os.environ,
{
"STRATEGY_ID": "sid-1",
"ALLOW_LIVE_SUBMIT": "1",
"LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
},
clear=False,
):
result = attempt_live_submit({"modelAddress": "x/y"}, client=client)
self.assertTrue(result["submitted"])
self.assertEqual(len(client.calls), 1)
def test_default_attempt_rejects(self) -> None:
client = FakeClient()
with self._clear_gate_env():
result = attempt_live_submit({"modelAddress": "x/y"}, client=client)
self.assertFalse(result["submitted"])
self.assertEqual(len(client.calls), 0)
if __name__ == "__main__":
unittest.main()
def test_no_client_no_network(self):
with patch('urllib.request.urlopen') as network:
self.assertFalse(attempt_live_submit({})['submitted'])
network.assert_not_called()

110
tests/test_live_cycle.py Normal file
View File

@@ -0,0 +1,110 @@
import io
import os
import tempfile
import unittest
from unittest.mock import Mock, patch
from src.live_cycle import run_cycle, _submitted
def candidate(name, section='B', foothold='unknown'):
return dict(model_address='org/' + name, section=section, foothold_class=foothold, try_next_gpus=['壁砺'])
class TestLiveCycle(unittest.TestCase):
def setUp(self):
directory = tempfile.TemporaryDirectory()
self.addCleanup(directory.cleanup)
self.env = patch.dict(os.environ, {'STRATEGY_ID': 'sid', 'LIVE_STATE_FILE': directory.name + '/state.json', 'CONFIG_PARAMS_SOURCE': 'sample', 'XC_TOKEN': 'synthetic-secret'}, clear=True)
self.env.start()
self.addCleanup(self.env.stop)
_submitted.clear()
self.client = Mock()
self.client.submit.return_value = dict(http_status=200, platform_code=0)
network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
network.start()
self.addCleanup(network.stop)
logs = patch('sys.stderr', new_callable=io.StringIO)
self.logs = logs.start()
self.addCleanup(logs.stop)
def cycle(self, rows=None, counts=(94,)):
reader = Mock(side_effect=[(n, 'platform_error' if n is None else 'ok', set()) for n in counts])
return run_cycle(rows or [candidate('a')], self.client, reader)
def test_high_and_failed_queue(self):
for n in (95, None):
decisions = self.cycle(counts=(n,))
self.assertEqual(decisions[-1]['action'], 'hold')
self.client.submit.assert_not_called()
self.assertIn('[live] hold', self.logs.getvalue())
def test_order_and_cap(self):
rows = [candidate('b'), candidate('a', 'A'), candidate('verified', 'A', 'verified_reuse'), candidate('c'), candidate('d')]
self.cycle(rows, (90, 90, 90))
self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/verified', 'org/a', 'org/b'])
self.assertIn('cycle_cap', self.logs.getvalue())
def test_midcycle(self):
self.cycle([candidate(str(i)) for i in range(5)], (93, 94, 95))
self.assertEqual(self.client.submit.call_count, 2)
def test_kill_and_threshold(self):
for env in ({'ALLOW_LIVE_SUBMIT': '0'}, {'LIVE_QUEUE_THRESHOLD': 'bad'}, {'LIVE_QUEUE_THRESHOLD': '20'}):
with patch.dict(os.environ, env):
self.cycle(counts=(20,))
self.client.submit.assert_not_called()
with patch.dict(os.environ, {'LIVE_QUEUE_THRESHOLD': '100'}):
self.cycle(counts=(99,))
self.client.submit.assert_called_once()
def test_banned_duplicate_persistence(self):
rows = [candidate('GPT-2'), candidate('ok'), candidate('ok')]
self.cycle(rows)
_submitted.clear()
self.cycle([candidate('ok')], ())
self.client.submit.assert_called_once()
self.assertIn('banned:', self.logs.getvalue())
self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
def test_remote_duplicate_and_transport_reservation(self):
run_cycle([candidate('remote')], self.client, lambda: (1, 'ok', {('org/remote', 'Biren_166m')}))
self.client.submit.assert_not_called()
self.client.submit.side_effect = OSError('do not log')
self.cycle()
self.cycle(counts=())
self.client.submit.assert_called_once()
self.assertNotIn('do not log', self.logs.getvalue())
def test_missing_token_real_reader(self):
with patch.dict(os.environ, {'XC_TOKEN': ''}):
run_cycle([candidate('a')], self.client)
self.client.submit.assert_not_called()
self.assertIn('missing/invalid_token', self.logs.getvalue())
def test_worker_waits_and_stops(self):
from src.server import _live_worker
with patch('src.server._shutdown') as shutdown, patch('src.server.run_cycle') as cycle:
shutdown.is_set.return_value = False
shutdown.wait.return_value = True
with patch.dict(os.environ, {'LIVE_CYCLE_SECONDS': '1'}):
_live_worker()
cycle.assert_called_once()
shutdown.wait.assert_called_once_with(60)
def test_unauthenticated_platform_zero_submits(self):
import json
response = Mock()
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
response.status = 200
response.read.return_value = json.dumps({'code': 40100}).encode()
with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}), patch('urllib.request.urlopen', return_value=response):
run_cycle([candidate('a')], self.client)
self.client.submit.assert_not_called()
def test_unmapped_and_placeholder(self):
rows = [dict(model_address='org/unmapped', targetGpu='unknown'), candidate('<MODEL>')]
self.cycle(rows, ())
self.client.submit.assert_not_called()
self.assertIn('all_gpus_done_or_unmapped', self.logs.getvalue())
self.assertIn('placeholder', self.logs.getvalue())

71
tests/test_live_ledger.py Normal file
View File

@@ -0,0 +1,71 @@
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from src.live_cycle import run_cycle, _submitted
from src.ledger import load_baked, rebuild
from src.banned import banned_rule
from src.task_add_client import resolve_token_with_source
class TestLedger(unittest.TestCase):
def setUp(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
self.state = Path(temp.name) / 'state.json'
env = patch.dict(os.environ, dict(STRATEGY_ID='sid', XC_TOKEN='synthetic-credential', CONFIG_PARAMS_SOURCE='sample', LIVE_STATE_FILE=str(self.state)), clear=True)
env.start()
self.addCleanup(env.stop)
_submitted.clear()
self.addCleanup(_submitted.clear)
net = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
net.start()
self.addCleanup(net.stop)
self.client = Mock()
self.client.submit.return_value = dict(http_status=200, platform_code=0)
def cycle(self, remote=(), model='unit/model', gpus=None):
return run_cycle([dict(model_address=model, try_next_gpus=gpus or ['壁砺', '天垓100'])], self.client, lambda: (0, 'ok', set(remote)))
def test_pair_and_restart(self):
self.cycle([('unit/model', 'Biren_166m')])
self.assertEqual(self.client.submit.call_args.args[0]['targetGpu'], 'Iluvatar_bi-100')
self.assertEqual(json.loads(self.state.read_text()), ['unit/model|Iluvatar_bi-100'])
_submitted.clear()
self.cycle([('unit/model', 'Biren_166m')])
self.client.submit.assert_called_once()
def test_baked_and_blocked(self):
pairs, blocked = load_baked()
model, gpu = next(iter(pairs))
self.cycle(model=model, gpus=[gpu])
self.cycle(model=next(iter(blocked)))
self.client.submit.assert_not_called()
def test_failure_and_missing_token(self):
result = run_cycle([dict(model_address='unit/model', try_next_gpus=['壁砺'])], self.client, lambda: (None, 'failed', set()))
self.assertEqual(result[-1]['reason'], 'ledger_rebuild_failed')
with patch.dict(os.environ, {'XC_TOKEN': ''}):
self.cycle()
self.client.submit.assert_not_called()
def test_source_and_no_value_logs(self):
import io
with patch.dict(os.environ, {'EXTERNAL_SERVICE_TOKEN': 'platform-credential', 'XC_TOKEN_FILE': '/nonexistent'}), patch('sys.stderr', new_callable=io.StringIO) as log:
self.assertEqual(resolve_token_with_source(), ('platform-credential', 'EXTERNAL_SERVICE_TOKEN'))
self.cycle()
self.assertEqual(log.getvalue().count('token source=EXTERNAL_SERVICE_TOKEN'), 1)
self.assertNotIn('platform-credential', log.getvalue())
with patch.dict(os.environ, {'EXTERNAL_SERVICE_TOKEN': '<TOKEN>', 'XC_TOKEN': 'placeholder', 'MODELHUB_XC_TOKEN': 'alias-credential'}):
self.assertEqual(resolve_token_with_source()[1], 'MODELHUB_XC_TOKEN')
def test_candidates_and_rebuild(self):
rows = json.loads(Path('src/data/candidates.json').read_text())
self.assertGreater(len(rows), 0)
self.assertEqual([r['section'] for r in rows], sorted(r['section'] for r in rows))
self.assertFalse(any(banned_rule(r['model_address']) for r in rows))
count, pairs, blocked, sizes = rebuild(self.state, lambda: (0, 'ok', {('unit/model', 'Biren_166m')}))
self.assertIn(('unit/model', 'Biren_166m'), pairs)
self.assertEqual(sizes[0], 1)

55
tests/test_live_queue.py Normal file
View File

@@ -0,0 +1,55 @@
import json
import os
import unittest
from unittest.mock import patch, MagicMock
from src.live_queue import read_queue
class TestLiveQueue(unittest.TestCase):
def response(self, body):
r = MagicMock()
r.__enter__.return_value = r
r.status = 200
r.read.return_value = json.dumps(body).encode()
return r
def test_pagination_and_status(self):
pages = [self.response({'code': 0, 'data': {'total': '3', 'records': [
{'status': 'waiting', 'modelId': 'org/a', 'gpuType': 'Biren_166m'}, {'status': 'success', 'verifyResult': -1}]}}),
self.response({'code': 0, 'data': {'total': '3', 'records': [{'status': '等待中'}]}})]
with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', side_effect=pages) as call:
count, why, models = read_queue()
self.assertEqual(count, 2)
self.assertIn(('org/a', 'Biren_166m'), models)
self.assertIn('current=2', call.call_args.args[0].full_url)
def test_fail_closed(self):
for body in ({'code': 0, 'data': {'total': 'bad', 'records': []}}, {'code': 40100}, {'code': 40101}, {'code': 0, 'data': {'total': 1, 'records': []}}, []):
with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', return_value=self.response(body)):
self.assertIsNone(read_queue()[0])
with patch.dict(os.environ, {}, clear=True), patch('urllib.request.urlopen') as call:
self.assertIsNone(read_queue()[0])
call.assert_not_called()
def test_page_limit_and_transport(self):
with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True):
with patch('urllib.request.urlopen', return_value=self.response({'code': 0, 'data': {'total': 21, 'records': [{'status': 'waiting'}]}})) as call:
self.assertIsNone(read_queue()[0])
self.assertEqual(call.call_count, 20)
with patch('urllib.request.urlopen', side_effect=OSError('synthetic-secret')):
self.assertNotIn('synthetic-secret', str(read_queue()))
def test_invalid_json_http_and_auth_header(self):
with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True):
r = self.response({})
r.read.return_value = b'invalid synthetic-secret'
with patch('urllib.request.urlopen', return_value=r):
self.assertEqual(read_queue()[1], 'malformed_json')
r.status = 503
with patch('urllib.request.urlopen', return_value=r):
self.assertIsNone(read_queue()[0])
with patch('urllib.request.urlopen', return_value=self.response({'code': 0, 'data': {'total': 0, 'records': []}})) as call:
self.assertEqual(read_queue()[0], 0)
req = call.call_args.args[0]
self.assertEqual(req.get_header('Xc-token'), 'synthetic-secret')
self.assertEqual(req.method, 'GET')

View File

@@ -47,3 +47,40 @@ class TestPendingPool(unittest.TestCase):
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]['try_next_gpus'], ['天垓150'])
self.assertEqual(rows[0]['perf_order'], 'unset')
def test_foothold_evidence_and_stable_verified_priority(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
md = root / 'pending.md'
md.write_text('## Section B\n| 1 | `org/b` | 0 | 壁砺 |\n## Section A\n| 1 | `org/unknown` | 曦望 | 壁砺 | 已验证成功 |\n| 2 | `org/flight` | 验证失败; 正在验证 | 壁砺 | — |\n| 3 | `org/verified` | 他人:已验证成功; 正在验证 | 壁砺 | — |\n| 4 | `org/fail` | 验证失败 | 曦望 | — |\n| 5 | `org/pass` | 既有:验证通过 | 曦望 | — |\n| 6 | `org/queue` | 验证排队中 | 壁砺 | — |\n| 7 | `org/progress` | 验证进行中 | 壁砺 | — |\n')
rows = build_ordered_candidates(md, dnr_path=root/'none', banned_path=root/'none')
self.assertEqual([r['model_address'] for r in rows],
['org/verified', 'org/pass', 'org/unknown', 'org/flight',
'org/fail', 'org/queue', 'org/progress', 'org/b'])
self.assertEqual([r['foothold_class'] for r in rows],
['verified_reuse', 'verified_reuse', 'unknown', 'in_flight',
'failed_foothold', 'in_flight', 'in_flight', 'unknown'])
for row in rows:
self.assertEqual(row['evidence_source'], 'GAP' if row['foothold_class'] == 'unknown' else 'pending-foothold')
self.assertEqual(rows[0]['foothold_text'], '他人:已验证成功; 正在验证')
txt = root / 'pending.txt'
txt.write_text('org/a\t#A done:曦望; try:壁砺\norg/v\t#A 验证通过; try:壁砺\n')
parsed = parse_pending_txt(txt)
self.assertEqual(parsed[0]['foothold_class'], 'unknown')
self.assertEqual(parsed[0]['evidence_source'], 'GAP')
self.assertEqual(parsed[1]['foothold_class'], 'verified_reuse')
def test_no_download_marker_without_inventing(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
md = root / 'pending.md'
md.write_text('## Section A\n| 1 | `org/plain` | 曦望:验证排队中 | 壁砺 | — |\n| 2 | `org/cached` | 壁砺:已下载; 曦望:免下载 | 天垓150 | — |\n| 3 | `org/verified` | 壁砺:已验证成功 | 天垓150 | — |\n')
rows = build_ordered_candidates(md, dnr_path=root/'none', banned_path=root/'none')
by = {r['model_address']: r for r in rows}
self.assertFalse(by['org/plain']['no_download'])
self.assertEqual(by['org/plain']['evidence_source'], 'pending-foothold')
self.assertTrue(by['org/cached']['no_download'])
self.assertEqual(by['org/cached']['evidence_source'], 'pending-foothold')
self.assertFalse(by['org/verified']['no_download']) # verified alone does not invent no_download
self.assertEqual([r['model_address'] for r in rows],
['org/cached', 'org/verified', 'org/plain'])

View File

@@ -9,7 +9,6 @@ import subprocess
import sys
import time
import unittest
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
@@ -17,35 +16,36 @@ ROOT = Path(__file__).resolve().parents[1]
class TestSigterm(unittest.TestCase):
def test_sigterm_exits_quickly(self) -> None:
port = "19090"
env = os.environ.copy()
env["STRATEGY_ID"] = "sigterm-test"
env["PORT"] = port
env["ALLOW_LIVE_SUBMIT"] = "0"
env["HOST"] = "127.0.0.1"
proc = subprocess.Popen(
[sys.executable, "-m", "src.server"],
[sys.executable, '-c', """
from unittest.mock import patch
import src.server as server
class OfflineServer:
def __init__(self, *args): pass
def serve_forever(self, **kwargs): server._shutdown.wait(10)
def shutdown(self): pass
def server_close(self): pass
with patch.object(server, 'ThreadingHTTPServer', OfflineServer):
server.main()
"""],
cwd=str(ROOT),
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
)
try:
# wait until health is up
url = f"http://127.0.0.1:{port}/health"
deadline = time.time() + 5
while time.time() < deadline:
try:
with urllib.request.urlopen(url, timeout=0.5) as resp:
data = json.loads(resp.read().decode("utf-8"))
if resp.status == 200 and data.get("status") == "ok":
break
except Exception:
time.sleep(0.1)
else:
if proc.poll() is not None:
self.fail(f"server exited before health check: {proc.stderr.read().decode(errors='replace')}")
self.fail("server did not become healthy in time")
# Read readiness from stderr; no sockets or network calls.
import selectors
selector = selectors.DefaultSelector()
self.addCleanup(selector.close)
selector.register(proc.stderr, selectors.EVENT_READ)
self.assertTrue(selector.select(timeout=5), 'server readiness timeout')
self.assertIn(b'Listening on', proc.stderr.readline())
t0 = time.time()
proc.send_signal(signal.SIGTERM)