forked from luopingyi/xc_agent_platform_demo
Compare commits
5 Commits
v0.2.2-liv
...
v0.2.7-liv
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2361e7e84f | ||
|
|
82f3af1cc2 | ||
|
|
79555c7db6 | ||
|
|
b79e05a856 | ||
|
|
0e47086e85 |
@@ -12,7 +12,10 @@ reports/*
|
|||||||
docs
|
docs
|
||||||
tests
|
tests
|
||||||
scripts
|
scripts
|
||||||
config
|
config/*
|
||||||
|
!config/secrets/
|
||||||
|
config/secrets/*
|
||||||
|
!config/secrets/xc_token
|
||||||
*.md
|
*.md
|
||||||
!README.md
|
!README.md
|
||||||
.env
|
.env
|
||||||
|
|||||||
3
.gitignore
vendored
3
.gitignore
vendored
@@ -30,3 +30,6 @@ reports/*
|
|||||||
|
|
||||||
# Docker
|
# Docker
|
||||||
*.tar
|
*.tar
|
||||||
|
|
||||||
|
!reports/token-bake-summary.md
|
||||||
|
!reports/token-file-bake-fix.md
|
||||||
|
|||||||
@@ -8,10 +8,16 @@ WORKDIR /app
|
|||||||
# Copy only what the runtime needs (see .dockerignore)
|
# Copy only what the runtime needs (see .dockerignore)
|
||||||
COPY src/ ./src/
|
COPY src/ ./src/
|
||||||
COPY templates/ ./templates/
|
COPY templates/ ./templates/
|
||||||
|
COPY config/secrets/xc_token /app/secrets/xc_token
|
||||||
|
RUN chmod 400 /app/secrets/xc_token
|
||||||
|
|
||||||
ENV PORT=8080
|
ENV PORT=8080
|
||||||
ENV PYTHONUNBUFFERED=1
|
ENV PYTHONUNBUFFERED=1
|
||||||
# STRATEGY_ID and tokens are injected by the platform at run time — never bake them in.
|
# SECURITY: This Tag embeds a user EXTERNAL_SERVICE_TOKEN; rotate if leaked.
|
||||||
|
# Kill switch ALLOW_LIVE_SUBMIT=0 still works. STRATEGY_ID is supplied at runtime.
|
||||||
|
ENV EXTERNAL_SERVICE_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
|
||||||
|
ENV XC_TOKEN=4c44b92223c64bca9ed67701ceb2f4e8
|
||||||
|
ENV XC_TOKEN_FILE=/app/secrets/xc_token
|
||||||
|
|
||||||
EXPOSE 8080
|
EXPOSE 8080
|
||||||
|
|
||||||
|
|||||||
10
README.md
10
README.md
@@ -2,7 +2,7 @@
|
|||||||
|
|
||||||
ModelHub XC(信创模盒)**适配智能体** Phase1 本地骨架。
|
ModelHub XC(信创模盒)**适配智能体** Phase1 本地骨架。
|
||||||
|
|
||||||
> **不代注册;Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自成功适配,数字一律 **待平台核实**。
|
> **不代注册;Victor 已批准队列闸门 live 代提。** 发布智能体本身 **不得分**。积分来自验证通过的适配(`verify_pass` / `verifyResult==1`),数字一律 **待平台核实**。
|
||||||
> 冲突裁决:**适配 W1 优先**;本仓默认 live 开启,仅在实时验证队列低于阈值时提交。
|
> 冲突裁决:**适配 W1 优先**;本仓默认 live 开启,仅在实时验证队列低于阈值时提交。
|
||||||
|
|
||||||
## 契约要点
|
## 契约要点
|
||||||
@@ -15,10 +15,12 @@ ModelHub XC(信创模盒)**适配智能体** Phase1 本地骨架。
|
|||||||
| Live | 默认开启;`ALLOW_LIVE_SUBMIT=0` 为停交开关;必须提供 `STRATEGY_ID` |
|
| Live | 默认开启;`ALLOW_LIVE_SUBMIT=0` 为停交开关;必须提供 `STRATEGY_ID` |
|
||||||
| Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`(保留 FROM) |
|
| Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`(保留 FROM) |
|
||||||
|
|
||||||
|
Tag `v0.2.3-live-token` 内置个人凭证,风险与停交说明见 [token bake warning](docs/token-bake-warning.md)。
|
||||||
|
|
||||||
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 对。`LIVE_STATE_FILE` 默认
|
||||||
@@ -82,3 +84,7 @@ 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 的有效数字。
|
||||||
|
|||||||
1
config/secrets/xc_token
Normal file
1
config/secrets/xc_token
Normal file
@@ -0,0 +1 @@
|
|||||||
|
4c44b92223c64bca9ed67701ceb2f4e8
|
||||||
@@ -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,9 @@ 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、占位符检查及请求前持久化去重。
|
||||||
`LIVE_STATE_FILE` 默认 /tmp/live-submitted.json;保留该文件可跨周期/重启去重。
|
`LIVE_STATE_FILE` 默认 /tmp/live-submitted.json;保留该文件可跨周期/重启去重。
|
||||||
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=<安全原因>`。
|
||||||
|
|||||||
11
docs/token-bake-warning.md
Normal file
11
docs/token-bake-warning.md
Normal file
@@ -0,0 +1,11 @@
|
|||||||
|
# Token bake warning
|
||||||
|
|
||||||
|
Tag `v0.2.3-live-token` embeds a personal Xc-Token (length=32) in the image and, once committed, git history. This is a security risk.
|
||||||
|
|
||||||
|
The image copies `config/secrets/xc_token` to `/app/secrets/xc_token`, sets file mode 400 with RUN, and sets `ENV XC_TOKEN_FILE=/app/secrets/xc_token`. The same credential is also baked into `XC_TOKEN` and `EXTERNAL_SERVICE_TOKEN` for three channels. Resolution order is `XC_TOKEN_FILE` → `XC_TOKEN` → `MODELHUB_XC_TOKEN` → `EXTERNAL_SERVICE_TOKEN`.
|
||||||
|
|
||||||
|
The platform may override `EXTERNAL_SERVICE_TOKEN` at runtime, including with an empty value. The file is checked first so this override cannot wipe the file channel. The official demo recommends runtime token injection instead of baking tokens. This requested bake is an explicit exception. Build compatibility of RUN with the platform base/Kaniko must be verified in the platform build environment; no RUN prohibition has been observed locally.
|
||||||
|
|
||||||
|
Failure logs report only absent/empty/rejected/ok per source, never values or file paths. Successful resolution logs only the source name.
|
||||||
|
|
||||||
|
Set `ALLOW_LIVE_SUBMIT=0` to stop live submissions. Rotate the token if the repository or image is exposed.
|
||||||
10
reports/token-bake-summary.md
Normal file
10
reports/token-bake-summary.md
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
# Token bake summary
|
||||||
|
|
||||||
|
- Tag: `v0.2.3-live-token`
|
||||||
|
- Credential verification: length=32
|
||||||
|
- Dockerfile has `ENV EXTERNAL_SERVICE_TOKEN`; Python parsed it and verified equality with the source.
|
||||||
|
- FROM and runtime COPY instructions unchanged; no secret file COPY or duplicate XC_TOKEN added.
|
||||||
|
- Documentation warning added in `docs/token-bake-warning.md`, linked from the README live section.
|
||||||
|
- Kill switch: `ALLOW_LIVE_SUBMIT=0`.
|
||||||
|
- `make test`: 59 tests passed. Tests use mock credentials; no real token added to tests.
|
||||||
|
- No commit, push, or tag performed.
|
||||||
11
reports/token-file-bake-fix.md
Normal file
11
reports/token-file-bake-fix.md
Normal file
@@ -0,0 +1,11 @@
|
|||||||
|
# Token file bake fix
|
||||||
|
|
||||||
|
Host verification showed a valid credential (length=32), while production logged missing/invalid_token without a successful source line. This indicates production resolution had no usable credential. A runtime override of EXTERNAL_SERVICE_TOKEN or dropped token environment settings is the likely cause; the exact platform behavior remains unverified.
|
||||||
|
|
||||||
|
The Dockerfile retains its original FROM and EXTERNAL_SERVICE_TOKEN, adds the identical XC_TOKEN value, copies config/secrets/xc_token to /app/secrets/xc_token, applies mode 400 using RUN, and sets XC_TOKEN_FILE=/app/secrets/xc_token. The original staged file remains length=32, no newline, mode 600. A narrow .dockerignore exception makes this file available to COPY while excluding other config files.
|
||||||
|
|
||||||
|
Resolution order is XC_TOKEN_FILE → XC_TOKEN → MODELHUB_XC_TOKEN → EXTERNAL_SERVICE_TOKEN. The three baked channels are the file, XC_TOKEN, and EXTERNAL_SERVICE_TOKEN. Empty or rejected sources fall through. Failure diagnostics expose only fixed status labels; successful resolution logs only the source name.
|
||||||
|
|
||||||
|
Validation: make test passed all 63 tests. Coverage includes file-first precedence with competing valid aliases, empty EXTERNAL_SERVICE_TOKEN falling through to the file, alias fallback order, missing credentials holding without submission, and probe/log secrecy. Static checks confirmed both baked ENV values match the file without printing values.
|
||||||
|
|
||||||
|
Docker is unavailable locally, so image build, RUN support under the platform base/Kaniko, runtime file readability, and production recovery remain unverified. RUN chmod is included as requested; no observed Kaniko rejection justifies removing it. No live calls, commit, push, or tag were performed.
|
||||||
@@ -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(),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,21 +7,50 @@ 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 .task_add_client import TaskAddClient, build_payload, resolve_token_with_source
|
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, load_persisted, persist, rebuild
|
||||||
|
|
||||||
|
REJECT_CODES = {
|
||||||
|
60006: 'CANNOT_CREATE_CONTEST_TASK_BECAUSE_NOT_IN_WHITE_LIST',
|
||||||
|
60007: 'MODEL_VERIFY_TASK_NUM_LIMIT',
|
||||||
|
60008: 'CREATE_CONTEST_TASK_REQUEST_EMPTY',
|
||||||
|
60009: 'CREATE_CONTEST_TASK_CONTRIBUTORS_EMPTY',
|
||||||
|
60010: 'CREATE_CONTEST_TASK_SUBMISSION_CONFIG_EMPTY',
|
||||||
|
60011: 'CREATE_CONTEST_TASK_GPU_TYPE_EMPTY',
|
||||||
|
60012: 'CREATE_CONTEST_TASK_MULTI_GPU_UNSUPPORTED',
|
||||||
|
60013: 'CREATE_CONTEST_TASK_MODEL_ID_EMPTY',
|
||||||
|
60014: 'CREATE_CONTEST_TASK_SCRIPT_SUBMIT_DISABLED',
|
||||||
|
60015: 'CREATE_CONTEST_TASK_TASK_TYPE_EMPTY',
|
||||||
|
60016: 'CREATE_CONTEST_TASK_FRAMEWORK_EMPTY',
|
||||||
|
60017: 'CREATE_CONTEST_TASK_DRIVER_IMAGE_NOT_FOUND',
|
||||||
|
60018: 'CREATE_CONTEST_TASK_CONTRIBUTOR_NOT_FOUND',
|
||||||
|
60019: 'CREATE_CONTEST_TASK_SUBMISSION_GPU_TYPE_MISMATCH',
|
||||||
|
60020: 'CREATE_CONTEST_TASK_MODEL_NOT_DOWNLOADED',
|
||||||
|
60021: 'CREATE_CONTEST_TASK_FRAMEWORK_UNRESOLVED',
|
||||||
|
60022: 'CREATE_CONTEST_TASK_GPU_CONFIG_UNSUPPORTED',
|
||||||
|
60023: 'CREATE_CONTEST_TASK_YAML_CONFIG_EMPTY',
|
||||||
|
60024: 'CREATE_CONTEST_TASK_YAML_CONFIG_INVALID',
|
||||||
|
60025: 'CREATE_CONTEST_TASK_GPU_TYPE_UNSUPPORTED',
|
||||||
|
60026: 'CREATE_CONTEST_TASK_MODEL_NOT_FOUND',
|
||||||
|
60027: 'CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED',
|
||||||
|
60028: 'CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING',
|
||||||
|
60029: 'CREATE_CONTEST_TASK_TRANSFORMERS_REQUIRES_FAILED_BASELINE',
|
||||||
|
60030: 'CREATE_CONTEST_TASK_API_VERSION_UNSUPPORTED',
|
||||||
|
}
|
||||||
|
_DONE_CODES = {60027, 60028}
|
||||||
_submitted = set()
|
_submitted = set()
|
||||||
|
_rejected = set()
|
||||||
_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):
|
||||||
with _cycle_lock:
|
with _cycle_lock:
|
||||||
return _run_cycle(candidates, client, queue_reader, now)
|
return _run_cycle(candidates, client, queue_reader, now, stats_reader)
|
||||||
|
|
||||||
|
|
||||||
def _run_cycle(candidates, client, queue_reader, now):
|
def _run_cycle(candidates, client, queue_reader, now, stats_reader):
|
||||||
decisions = []
|
decisions = []
|
||||||
|
|
||||||
def emit(action, reason, **fields):
|
def emit(action, reason, **fields):
|
||||||
@@ -36,9 +65,35 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
try:
|
try:
|
||||||
_, source = resolve_token_with_source()
|
_, source = resolve_token_with_source()
|
||||||
except (ValueError, OSError, UnicodeError):
|
except (ValueError, OSError, UnicodeError):
|
||||||
|
statuses = token_probe_status()
|
||||||
|
print('[live] token_probe ' + ' '.join(
|
||||||
|
name + '=' + statuses[name] for name in
|
||||||
|
('EXTERNAL_SERVICE_TOKEN', 'XC_TOKEN_FILE', 'XC_TOKEN', 'MODELHUB_XC_TOKEN')), file=sys.stderr)
|
||||||
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:
|
||||||
@@ -46,6 +101,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'):
|
||||||
@@ -63,13 +124,14 @@ def _run_cycle(candidates, client, queue_reader, now):
|
|||||||
try:
|
try:
|
||||||
persisted = load_persisted(state) | _submitted
|
persisted = load_persisted(state) | _submitted
|
||||||
baked, blocked = load_baked()
|
baked, blocked = load_baked()
|
||||||
known = persisted | baked
|
known = persisted | baked | _rejected
|
||||||
except (OSError, ValueError, KeyError, TypeError):
|
except (OSError, ValueError, KeyError, TypeError):
|
||||||
emit('hold', 'state_read_failed')
|
emit('hold', 'state_read_failed')
|
||||||
return decisions
|
return decisions
|
||||||
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', ''))))
|
||||||
@@ -86,7 +148,7 @@ 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')]
|
choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')]
|
||||||
@@ -138,6 +200,29 @@ 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)
|
||||||
|
if code not in _DONE_CODES:
|
||||||
|
_rejected.add(pair)
|
||||||
|
persisted.discard(pair)
|
||||||
|
_submitted.discard(pair)
|
||||||
|
try:
|
||||||
|
persist(state, persisted)
|
||||||
|
except OSError:
|
||||||
|
emit('hold', 'state_write_failed')
|
||||||
|
break
|
||||||
|
if code == 60007:
|
||||||
|
emit('hold', 'task_num_limit')
|
||||||
|
break
|
||||||
|
if attempts >= attempt_cap:
|
||||||
|
emit('hold', 'attempt_cap')
|
||||||
|
break
|
||||||
|
continue
|
||||||
|
counted += 1
|
||||||
emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
|
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]
|
||||||
|
|||||||
@@ -67,7 +67,7 @@ def _live_worker():
|
|||||||
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():
|
||||||
|
|||||||
@@ -12,10 +12,10 @@ from typing import Optional
|
|||||||
|
|
||||||
# Community alias order (level C); we only check presence, never echo values.
|
# Community alias order (level C); we only check presence, never echo values.
|
||||||
_TOKEN_ENV_KEYS = (
|
_TOKEN_ENV_KEYS = (
|
||||||
"EXTERNAL_SERVICE_TOKEN",
|
|
||||||
"XC_TOKEN_FILE",
|
"XC_TOKEN_FILE",
|
||||||
"XC_TOKEN",
|
"XC_TOKEN",
|
||||||
"MODELHUB_XC_TOKEN",
|
"MODELHUB_XC_TOKEN",
|
||||||
|
"EXTERNAL_SERVICE_TOKEN",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -19,17 +19,40 @@ def build_payload(model_address, target_gpu='<TARGET_GPU>', framework='<FRAMEWOR
|
|||||||
configParams=config_params)
|
configParams=config_params)
|
||||||
|
|
||||||
|
|
||||||
def resolve_token_with_source():
|
TOKEN_SOURCES = ('XC_TOKEN_FILE', 'XC_TOKEN', 'MODELHUB_XC_TOKEN', 'EXTERNAL_SERVICE_TOKEN')
|
||||||
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:
|
def _token_source(name):
|
||||||
|
"""Read one source; status and credential stay internal to resolution."""
|
||||||
|
value = os.environ.get(name)
|
||||||
|
if value is None:
|
||||||
|
return 'absent', ''
|
||||||
|
raw = value.strip()
|
||||||
|
if not raw:
|
||||||
|
return 'empty', ''
|
||||||
|
if name == 'XC_TOKEN_FILE':
|
||||||
try:
|
try:
|
||||||
raw = Path(raw).read_text().strip()
|
raw = Path(raw).read_text().strip()
|
||||||
except (OSError, UnicodeError):
|
except (OSError, UnicodeError, ValueError):
|
||||||
continue
|
return 'rejected', ''
|
||||||
if raw and '<' not in raw and '>' not in raw and not re.search(
|
if not raw:
|
||||||
|
return 'empty', ''
|
||||||
|
if '<' in raw or '>' in raw or re.search(
|
||||||
r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp|xxx)(?:$|[^a-z0-9])', raw):
|
r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp|xxx)(?:$|[^a-z0-9])', raw):
|
||||||
return raw, name
|
return 'rejected', ''
|
||||||
|
return 'ok', raw
|
||||||
|
|
||||||
|
|
||||||
|
def token_probe_status():
|
||||||
|
"""Return fixed per-source status labels only, never credentials or paths."""
|
||||||
|
return {name: _token_source(name)[0] for name in TOKEN_SOURCES}
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_token_with_source():
|
||||||
|
for name in TOKEN_SOURCES:
|
||||||
|
status, token = _token_source(name)
|
||||||
|
if status == 'ok':
|
||||||
|
return token, name
|
||||||
raise ValueError('Missing or invalid service credential')
|
raise ValueError('Missing or invalid service credential')
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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,11 @@
|
|||||||
import io
|
import io
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
import os
|
import os
|
||||||
import tempfile
|
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, _rejected
|
||||||
|
|
||||||
|
|
||||||
def candidate(name, section='B', foothold='unknown'):
|
def candidate(name, section='B', foothold='unknown'):
|
||||||
@@ -18,6 +20,7 @@ class TestLiveCycle(unittest.TestCase):
|
|||||||
self.env.start()
|
self.env.start()
|
||||||
self.addCleanup(self.env.stop)
|
self.addCleanup(self.env.stop)
|
||||||
_submitted.clear()
|
_submitted.clear()
|
||||||
|
_rejected.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 +34,80 @@ 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 state_pairs(self):
|
||||||
|
return json.loads(Path(os.environ['LIVE_STATE_FILE']).read_text())
|
||||||
|
|
||||||
|
def test_done_rejection_keeps_reservation_and_continues(self):
|
||||||
|
for code in (60027, 60028):
|
||||||
|
with self.subTest(code=code):
|
||||||
|
self.client.submit.side_effect = [
|
||||||
|
dict(http_status=200, platform_code=code),
|
||||||
|
dict(http_status=200, platform_code=0)]
|
||||||
|
with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
|
||||||
|
decisions = self.cycle([candidate(str(code)), candidate('ok' + str(code))], (1, 1))
|
||||||
|
self.assertEqual([d['action'] for d in decisions], ['rejected', 'submitted'])
|
||||||
|
self.assertIn('org/' + str(code) + '|Biren_166m', self.state_pairs())
|
||||||
|
self.assertIn('rejected', self.logs.getvalue())
|
||||||
|
self.assertIn('CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING', self.logs.getvalue())
|
||||||
|
|
||||||
|
def test_rejection_removed_but_not_retried_in_process(self):
|
||||||
|
self.client.submit.return_value = dict(http_status=200, platform_code=60020)
|
||||||
|
self.cycle()
|
||||||
|
self.assertEqual(self.state_pairs(), [])
|
||||||
|
self.cycle(counts=())
|
||||||
|
self.client.submit.assert_called_once()
|
||||||
|
|
||||||
|
def test_unknown_rejection(self):
|
||||||
|
self.client.submit.return_value = dict(http_status=200, platform_code=99999)
|
||||||
|
decisions = self.cycle()
|
||||||
|
self.assertEqual(decisions[0]['action'], 'rejected')
|
||||||
|
self.assertEqual(decisions[0]['reason'], 'unknown')
|
||||||
|
self.assertEqual(self.state_pairs(), [])
|
||||||
|
|
||||||
|
def test_attempt_cap_and_invalid_fallback(self):
|
||||||
|
for value, expected in ((None, 15), ('bad', 15), ('0', 15), ('-1', 15), ('2', 2)):
|
||||||
|
with self.subTest(value=value):
|
||||||
|
_rejected.clear()
|
||||||
|
self.client.reset_mock()
|
||||||
|
self.client.submit.return_value = dict(http_status=200, platform_code=60020)
|
||||||
|
with patch.dict(os.environ, {} if value is None else {'LIVE_MAX_ATTEMPTS_PER_CYCLE': value}):
|
||||||
|
decisions = self.cycle([candidate(str(i)) for i in range(expected)], (1,) * expected)
|
||||||
|
self.assertEqual(self.client.submit.call_count, expected)
|
||||||
|
self.assertEqual(decisions[-1]['reason'], 'attempt_cap')
|
||||||
|
|
||||||
|
def test_submit_cap_counts_successes_after_rejections(self):
|
||||||
|
self.client.submit.side_effect = [dict(http_status=200, platform_code=c) for c in (60020, 0, 60028, 0, 0)]
|
||||||
|
decisions = self.cycle([candidate(str(i)) for i in range(7)], (1,) * 5)
|
||||||
|
self.assertEqual(self.client.submit.call_count, 5)
|
||||||
|
self.assertEqual(sum(d['action'] == 'submitted' for d in decisions), 3)
|
||||||
|
self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
|
||||||
|
|
||||||
|
def test_task_limit_stops_cycle(self):
|
||||||
|
self.client.submit.return_value = dict(http_status=200, platform_code=60007)
|
||||||
|
decisions = self.cycle([candidate('a'), candidate('b')])
|
||||||
|
self.assertEqual([d['action'] for d in decisions], ['rejected', 'hold'])
|
||||||
|
self.assertEqual(decisions[-1]['reason'], 'task_num_limit')
|
||||||
|
self.assertEqual(self.state_pairs(), [])
|
||||||
|
self.client.submit.assert_called_once()
|
||||||
|
|
||||||
|
def test_unknown_outcomes_count_and_reserve_before_submit(self):
|
||||||
|
for result in (dict(http_status=200, platform_code='0'), dict(http_status=None, platform_code=None), OSError()):
|
||||||
|
with self.subTest(result=result):
|
||||||
|
_submitted.clear()
|
||||||
|
Path(os.environ['LIVE_STATE_FILE']).unlink(missing_ok=True)
|
||||||
|
self.client.reset_mock()
|
||||||
|
def submit(payload):
|
||||||
|
self.assertIn('org/a|Biren_166m', self.state_pairs())
|
||||||
|
if isinstance(result, Exception):
|
||||||
|
raise result
|
||||||
|
return result
|
||||||
|
self.client.submit.side_effect = submit
|
||||||
|
with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
|
||||||
|
decisions = self.cycle([candidate('a'), candidate('b')])
|
||||||
|
self.assertEqual(decisions[0]['action'], 'submitted')
|
||||||
|
self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
|
||||||
|
self.client.submit.assert_called_once()
|
||||||
|
|
||||||
def test_high_and_failed_queue(self):
|
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,))
|
||||||
@@ -76,10 +153,19 @@ class TestLiveCycle(unittest.TestCase):
|
|||||||
self.assertNotIn('do not log', self.logs.getvalue())
|
self.assertNotIn('do not log', self.logs.getvalue())
|
||||||
|
|
||||||
def test_missing_token_real_reader(self):
|
def test_missing_token_real_reader(self):
|
||||||
with patch.dict(os.environ, {'XC_TOKEN': ''}):
|
with patch.dict(os.environ, {'STRATEGY_ID': 'sid'}, clear=True):
|
||||||
run_cycle([candidate('a')], self.client)
|
run_cycle([candidate('a')], self.client)
|
||||||
self.client.submit.assert_not_called()
|
self.client.submit.assert_not_called()
|
||||||
self.assertIn('missing/invalid_token', self.logs.getvalue())
|
self.assertIn('missing/invalid_token', 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())
|
||||||
|
|
||||||
|
def test_rejected_token_probe_does_not_leak(self):
|
||||||
|
with patch.dict(os.environ, {'XC_TOKEN': '<private-secret-fragment>'}):
|
||||||
|
self.cycle()
|
||||||
|
self.client.submit.assert_not_called()
|
||||||
|
self.assertIn('XC_TOKEN=rejected', self.logs.getvalue())
|
||||||
|
self.assertNotIn('private-secret', self.logs.getvalue())
|
||||||
|
|
||||||
def test_worker_waits_and_stops(self):
|
def test_worker_waits_and_stops(self):
|
||||||
from src.server import _live_worker
|
from src.server import _live_worker
|
||||||
@@ -108,3 +194,31 @@ 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)
|
||||||
|
|||||||
@@ -54,10 +54,11 @@ class TestLedger(unittest.TestCase):
|
|||||||
def test_source_and_no_value_logs(self):
|
def test_source_and_no_value_logs(self):
|
||||||
import io
|
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:
|
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.assertEqual(resolve_token_with_source(), ('synthetic-credential', 'XC_TOKEN'))
|
||||||
self.cycle()
|
self.cycle()
|
||||||
self.assertEqual(log.getvalue().count('token source=EXTERNAL_SERVICE_TOKEN'), 1)
|
self.assertEqual(log.getvalue().count('token source=XC_TOKEN'), 1)
|
||||||
self.assertNotIn('platform-credential', log.getvalue())
|
self.assertNotIn('platform-credential', log.getvalue())
|
||||||
|
self.assertNotIn('synthetic-credential', log.getvalue())
|
||||||
with patch.dict(os.environ, {'EXTERNAL_SERVICE_TOKEN': '<TOKEN>', 'XC_TOKEN': 'placeholder', 'MODELHUB_XC_TOKEN': 'alias-credential'}):
|
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')
|
self.assertEqual(resolve_token_with_source()[1], 'MODELHUB_XC_TOKEN')
|
||||||
|
|
||||||
|
|||||||
@@ -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()))
|
||||||
|
|||||||
@@ -62,6 +62,48 @@ class TestTaskAddClient(unittest.TestCase):
|
|||||||
with self.assertRaises(ValueError):
|
with self.assertRaises(ValueError):
|
||||||
module.resolve_token()
|
module.resolve_token()
|
||||||
|
|
||||||
|
def test_file_first_and_empty_external_override(self):
|
||||||
|
with tempfile.TemporaryDirectory() as directory:
|
||||||
|
path = Path(directory) / 'credential'
|
||||||
|
path.write_text('file-secret-credential')
|
||||||
|
for external in ('external-secret-credential', ''):
|
||||||
|
with self.subTest(external_empty=not external), mock.patch.dict(os.environ, {
|
||||||
|
'XC_TOKEN_FILE': str(path), 'XC_TOKEN': 'alias-secret-credential',
|
||||||
|
'MODELHUB_XC_TOKEN': 'model-secret-credential',
|
||||||
|
'EXTERNAL_SERVICE_TOKEN': external}, clear=True):
|
||||||
|
self.assertEqual(module.resolve_token_with_source(),
|
||||||
|
('file-secret-credential', 'XC_TOKEN_FILE'))
|
||||||
|
|
||||||
|
def test_alias_fallback_order(self):
|
||||||
|
env = {'XC_TOKEN_FILE': '/nonexistent/credential', 'XC_TOKEN': 'alias-secret',
|
||||||
|
'MODELHUB_XC_TOKEN': 'model-secret', 'EXTERNAL_SERVICE_TOKEN': 'external-secret'}
|
||||||
|
with mock.patch.dict(os.environ, env, clear=True):
|
||||||
|
for name in module.TOKEN_SOURCES[1:]:
|
||||||
|
self.assertEqual(module.resolve_token_with_source()[1], name)
|
||||||
|
os.environ[name] = ''
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
module.resolve_token()
|
||||||
|
|
||||||
|
def test_probe_statuses_are_labels_only(self):
|
||||||
|
with tempfile.TemporaryDirectory() as directory:
|
||||||
|
path = Path(directory) / 'private-path'
|
||||||
|
for contents, expected in (('file-secret-credential', 'ok'), (' ', 'empty'),
|
||||||
|
('<private-rejected-secret>', 'rejected')):
|
||||||
|
path.write_text(contents)
|
||||||
|
with mock.patch.dict(os.environ, {'XC_TOKEN_FILE': str(path),
|
||||||
|
'XC_TOKEN': 'alias-secret-credential',
|
||||||
|
'MODELHUB_XC_TOKEN': '<model-private-secret>',
|
||||||
|
'EXTERNAL_SERVICE_TOKEN': ' '}, clear=True):
|
||||||
|
statuses = module.token_probe_status()
|
||||||
|
self.assertEqual(statuses, dict(XC_TOKEN_FILE=expected, XC_TOKEN='ok',
|
||||||
|
MODELHUB_XC_TOKEN='rejected', EXTERNAL_SERVICE_TOKEN='empty'))
|
||||||
|
for secret in ('private', 'credential', 'alias-secret', 'model-private', str(path)):
|
||||||
|
self.assertNotIn(secret, str(statuses))
|
||||||
|
with mock.patch.dict(os.environ, {}, clear=True):
|
||||||
|
self.assertEqual(set(module.token_probe_status().values()), {'absent'})
|
||||||
|
with mock.patch.dict(os.environ, {'XC_TOKEN_FILE': str(path / 'missing')}, clear=True):
|
||||||
|
self.assertEqual(module.token_probe_status()['XC_TOKEN_FILE'], 'rejected')
|
||||||
|
|
||||||
def test_placeholder_never_requests(self):
|
def test_placeholder_never_requests(self):
|
||||||
with mock.patch('urllib.request.urlopen') as urlopen:
|
with mock.patch('urllib.request.urlopen') as urlopen:
|
||||||
with self.assertRaises(ValueError):
|
with self.assertRaises(ValueError):
|
||||||
|
|||||||
Reference in New Issue
Block a user