NASP/argus-cluster
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

V3.6 增加W&B server，以及配置任务默认都开wandb

Browse Source
This commit is contained in:
yuyr
2026-01-05 17:04:10 +08:00
parent 70f52a907b
commit a6c66995e3
25 changed files with 5852 additions and 2223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/mvp/configs/dev.yaml
View File

@@ -32,6 +32,16 @@ service:
retry_interval_s: 60
max_running_tasks: 1
tracking:
wandb:
# v3.6: enable Weights & Biases integration (best-effort).
# - If WANDB_API_KEY is missing, the platform degrades to console logging and records a warning.
enabled: true
# For dev compose, use docker bridge gateway + published port for stability.
base_url: "http://172.22.0.1:8090"
api_key_env: "WANDB_API_KEY"
project_suffix: "_project"
# v3.0: user data management (filesystem + SFTPGo)
data:
# All user writable data is placed under this root:

9
src/mvp/configs/dev_v30.yaml
View File

@@ -32,6 +32,14 @@ service:
retry_interval_s: 60
max_running_tasks: 1
tracking:
wandb:
enabled: true
# For dev compose, recommend docker bridge gateway + host published port for stability.
base_url: "http://172.22.0.1:8090"
api_key_env: "WANDB_API_KEY"
project_suffix: "_project"
data:
user_root: "/private/users"
sftpgo:
@@ -48,4 +56,3 @@ data:
jobs_trash_after_days: 3
jobs_purge_after_days: 7
janitor_interval_s: 3600

21
src/mvp/docker-compose.yaml
View File

@@ -74,6 +74,27 @@ services:
SFTPGO_HTTPD__BINDINGS__0__PORT: "8080"
SFTPGO_SFTPD__BINDINGS__0__PORT: "2022"
# v3.6: Weights & Biases local server (shared single account; per-user projects).
#
# - UI: 8080 in container, mapped to host 8090
# - Data: persist under /private/common/wandb (NFS/shared)
#
# NOTE:
# - The license/initial setup is done via the W&B UI.
# - Ray jobs use WANDB_BASE_URL from config (recommended: docker bridge gateway + host published port).
wandb:
image: wandb/local:latest
container_name: argus-wandb
ports:
- "8090:8080"
volumes:
- ../../shared/common/wandb:/vol
networks:
argus-ray-net:
aliases:
- wandb
- argus-wandb
ray_worker_0:
image: argus/argus-ray-node:v2.5
container_name: argus-ray-worker-0

44
src/mvp/py/argus/ray/ray_job_tool.py
View File

@@ -10,7 +10,7 @@ from typing import Any
import ray
from ray.job_submission import JobSubmissionClient
from .builders import build_advanced_argv, build_training_argv
from .builders import BuiltCommand, build_advanced_argv, build_training_argv
from .models import AdvancedTaskSpec, JobSpec, RayConfig
from .yaml_io import dump_yaml
@@ -44,10 +44,17 @@ class RayJobTool:
def _job_dir(self, submission_id: str) -> str:
return f"{self.cfg.shared_root}/jobs/{submission_id}"
def _runtime_env(self, spec: JobSpec) -> dict[str, Any]:
return self._runtime_env_for(code_path=spec.code_path, workload=spec.workload)
def _runtime_env(self, spec: JobSpec, *, extra_env_vars: dict[str, str] | None = None) -> dict[str, Any]:
return self._runtime_env_for(code_path=spec.code_path, workload=spec.workload, extra_env_vars=extra_env_vars)
def _runtime_env_for(self, *, code_path: str, workload: str, extra_pythonpath: list[str] | None = None) -> dict[str, Any]:
def _runtime_env_for(
self,
*,
code_path: str,
workload: str,
extra_pythonpath: list[str] | None = None,
extra_env_vars: dict[str, str] | None = None,
) -> dict[str, Any]:
env_vars = dict(self.cfg.runtime_env_env_vars)
# Default HF cache
@@ -74,13 +81,29 @@ class RayJobTool:
if workload == "sft":
env_vars.setdefault("RAY_ADDRESS", "auto")
# Per-job extra env vars (e.g. W&B). Do not override existing keys.
if extra_env_vars:
for k, v in dict(extra_env_vars).items():
if k not in env_vars:
env_vars[k] = str(v)
return {"env_vars": env_vars}
def submit(self, spec: JobSpec, no_wait: bool, job_dir: str | None = None) -> str:
def submit(
self,
spec: JobSpec,
no_wait: bool,
job_dir: str | None = None,
*,
extra_env_vars: dict[str, str] | None = None,
extra_overrides: list[str] | None = None,
) -> str:
submission_id = spec.submission_id or f"mvp11_{spec.workload}_{_ts()}_{os.getpid()}"
job_dir = job_dir or self._job_dir(submission_id)
built = build_training_argv(spec, submission_id=submission_id, job_dir=job_dir)
if extra_overrides:
built = BuiltCommand(argv=[*built.argv, *list(extra_overrides)])
entrypoint_argv = [
"python3",
"-m",
@@ -91,7 +114,7 @@ class RayJobTool:
]
entrypoint = " ".join(shlex.quote(x) for x in entrypoint_argv)
runtime_env = self._runtime_env(spec)
runtime_env = self._runtime_env(spec, extra_env_vars=extra_env_vars)
# Prepare job artifacts directory
job_root = Path(job_dir)
@@ -155,6 +178,8 @@ class RayJobTool:
no_wait: bool,
job_dir: str | None = None,
user_id: str | None = None,
*,
extra_env_vars: dict[str, str] | None = None,
) -> str:
submission_id = spec.submission_id or f"mvp11_{spec.workload}_{_ts()}_{os.getpid()}"
job_dir = job_dir or self._job_dir(submission_id)
@@ -174,7 +199,12 @@ class RayJobTool:
if user_id:
extra_pythonpath.append(f"{self.cfg.shared_root}/users/{user_id}/code")
runtime_env = self._runtime_env_for(code_path=self.cfg.verl_code_path, workload=spec.workload, extra_pythonpath=extra_pythonpath)
runtime_env = self._runtime_env_for(
code_path=self.cfg.verl_code_path,
workload=spec.workload,
extra_pythonpath=extra_pythonpath,
extra_env_vars=extra_env_vars,
)
# Prepare job artifacts directory
job_root = Path(job_dir)

20
src/mvp/py/argus/service/app.py
View File

@@ -276,6 +276,14 @@ def create_app(config_path: str) -> FastAPI:
"jobs_purge_after_days": int(v2_cfg.data.retention.jobs_purge_after_days),
},
}
if v2_cfg.tracking.wandb.enabled:
suffix = v2_cfg.tracking.wandb.project_suffix or "_project"
out["wandb"] = {
"enabled": True,
# Training uses base_url; UI can also open the host port directly.
"base_url": v2_cfg.tracking.wandb.base_url,
"project_name": f"{user_id}{suffix}",
}
if _sftpgo_enabled():
out["sftp"] = {
"host": v2_cfg.data.sftpgo.host,
@@ -429,7 +437,7 @@ def create_app(config_path: str) -> FastAPI:
"error_summary": row.get("error_summary"),
}
if latest_attempt:
out["latest_attempt"] = {
latest_out: dict[str, Any] = {
"attempt_no": latest_attempt["attempt_no"],
"ray_submission_id": latest_attempt["ray_submission_id"],
"ray_status": latest_attempt.get("ray_status"),
@@ -438,6 +446,16 @@ def create_app(config_path: str) -> FastAPI:
"failure_kind": latest_attempt.get("failure_kind"),
"message": latest_attempt.get("message"),
}
if v2_cfg.tracking.wandb.enabled:
suffix = v2_cfg.tracking.wandb.project_suffix or "_project"
# For v3.6: shared W&B account, user-level projects.
# Use attempt's ray_submission_id as run name for stable mapping.
latest_out["wandb"] = {
"base_url": v2_cfg.tracking.wandb.base_url,
"project_name": f"{str(row.get('user_id') or subject.get('user_id'))}{suffix}",
"run_name": str(latest_attempt.get("ray_submission_id") or ""),
}
out["latest_attempt"] = latest_out
return out
@app.get("/api/v2/tasks/{task_id}/spec")

34
src/mvp/py/argus/service/config.py
View File

@@ -34,6 +34,19 @@ class V2RetentionConfig:
janitor_interval_s: int = 3600
@dataclass(frozen=True)
class V2WandbConfig:
enabled: bool = False
base_url: str = ""
api_key_env: str = "WANDB_API_KEY"
project_suffix: str = "_project"
@dataclass(frozen=True)
class V2TrackingConfig:
wandb: V2WandbConfig
@dataclass(frozen=True)
class V2SFTPGoConfig:
enabled: bool = False
@@ -57,6 +70,7 @@ class V2Config:
auth: V2AuthConfig
sqlite: V2SqliteConfig
scheduler: V2SchedulerConfig
tracking: V2TrackingConfig
data: V2DataConfig
@staticmethod
@@ -83,6 +97,13 @@ class V2Config:
else:
shared_root = str(root.get("shared_root") or "/private")
tracking = root.get("tracking") or {}
if not isinstance(tracking, dict):
raise ValueError("config.tracking must be a mapping")
wandb = tracking.get("wandb") or {}
if not isinstance(wandb, dict):
raise ValueError("config.tracking.wandb must be a mapping")
data = root.get("data") or {}
if not isinstance(data, dict):
raise ValueError("config.data must be a mapping")
@@ -96,6 +117,11 @@ class V2Config:
user_root = str(data.get("user_root") or f"{shared_root}/users")
wandb_enabled = bool(wandb.get("enabled") or False)
wandb_base_url = str(wandb.get("base_url") or "")
if wandb_enabled and not wandb_base_url:
raise ValueError("tracking.wandb.base_url is required when tracking.wandb.enabled=true")
return V2Config(
api=V2ApiConfig(
host=str(api.get("host") or "0.0.0.0"),
@@ -108,6 +134,14 @@ class V2Config:
retry_interval_s=int(scheduler.get("retry_interval_s") or 60),
max_running_tasks=int(scheduler.get("max_running_tasks") or 1),
),
tracking=V2TrackingConfig(
wandb=V2WandbConfig(
enabled=wandb_enabled,
base_url=wandb_base_url,
api_key_env=str(wandb.get("api_key_env") or "WANDB_API_KEY"),
project_suffix=str(wandb.get("project_suffix") or "_project"),
)
),
data=V2DataConfig(
user_root=user_root,
sftpgo=V2SFTPGoConfig(

92
src/mvp/py/argus/service/scheduler.py
View File

@@ -1,5 +1,6 @@
from __future__ import annotations
import os
import re
import time
from dataclasses import dataclass
@@ -48,6 +49,74 @@ class Scheduler:
required = float(nnodes * n_gpus_per_node)
return avail.total_available_gpus >= required
def _wandb_env_vars(self, *, ray_submission_id: str, job_dir: str) -> tuple[dict[str, str], str | None]:
"""
Returns (env_vars, warn_message).
env_vars is always non-empty so that AdvancedTaskSpec examples can safely
reference injected env vars without requiring the user to edit their command:
- MVP_TRAINER_LOGGER: "console" or "[console,wandb]"
- MVP_WANDB_PROJECT: "<user_id>_project"
- MVP_WANDB_RUN: "<ray_submission_id>"
When W&B is effectively enabled (config enabled + API key present),
WANDB_* env vars are included as well.
"""
env, warn, _enabled = self._tracking_env_vars(user_id=None, ray_submission_id=ray_submission_id, job_dir=job_dir)
return env, warn
def _tracking_env_vars(
self, *, user_id: str | None, ray_submission_id: str, job_dir: str
) -> tuple[dict[str, str], str | None, bool]:
"""
Returns (env_vars, warn_message, wandb_enabled_effective).
- env_vars always includes MVP_* vars so advanced commands can use them.
- wandb_enabled_effective is True only when tracking.wandb.enabled AND api key exists.
"""
cfg = self.v2_cfg.tracking.wandb
project_suffix = cfg.project_suffix or "_project"
project_name = f"{user_id}{project_suffix}" if user_id else f"mvp{project_suffix}"
api_key = os.environ.get(cfg.api_key_env, "").strip()
wandb_enabled_effective = bool(cfg.enabled and api_key)
warn: str | None = None
if cfg.enabled and not api_key:
warn = f"WARN_WANDB_DISABLED_MISSING_ENV:{cfg.api_key_env}"
env: dict[str, str] = {
"MVP_TRAINER_LOGGER": ("[console,wandb]" if wandb_enabled_effective else "console"),
"MVP_WANDB_PROJECT": project_name,
"MVP_WANDB_RUN": ray_submission_id,
}
if wandb_enabled_effective:
env.update(
{
"WANDB_BASE_URL": cfg.base_url,
"WANDB_API_KEY": api_key,
"WANDB_MODE": "online",
"WANDB_DIR": f"{job_dir.rstrip('/')}/wandb",
}
)
return env, warn, wandb_enabled_effective
def _wandb_training_overrides(self, *, user_id: str | None, ray_submission_id: str) -> list[str] | None:
"""
Returns hydra overrides to enable W&B logging for built-in training workloads.
Only call this when W&B is effectively enabled for the job (i.e. API key exists).
"""
cfg = self.v2_cfg.tracking.wandb
project_suffix = cfg.project_suffix or "_project"
project_name = f"{user_id}{project_suffix}" if user_id else f"mvp{project_suffix}"
return [
"trainer.logger=[console,wandb]",
f"trainer.project_name={project_name}",
f"trainer.experiment_name={ray_submission_id}",
]
def _parse_taskspec(self, jobspec_yaml: str) -> JobSpec | AdvancedTaskSpec:
obj = yaml.safe_load(jobspec_yaml) or {}
if not isinstance(obj, dict):
@@ -70,21 +139,38 @@ class Scheduler:
self.db.set_task_state(task_id=task_id, state="SUBMITTING", latest_attempt_no=attempt_no)
# Override submission_id in taskspec (v1.1 compatible)
tracking_env, wandb_warn, wandb_enabled = self._tracking_env_vars(
user_id=user_id_s, ray_submission_id=ray_sid, job_dir=job_dir
)
if isinstance(spec, JobSpec):
d = spec.to_public_dict()
d["submission_id"] = ray_sid
spec2 = JobSpec.from_dict(d)
submit = lambda: self.tool.submit(spec2, no_wait=True, job_dir=job_dir)
wandb_overrides = self._wandb_training_overrides(user_id=user_id_s, ray_submission_id=ray_sid) if wandb_enabled else None
submit = lambda: self.tool.submit(
spec2,
no_wait=True,
job_dir=job_dir,
extra_env_vars=tracking_env,
extra_overrides=wandb_overrides,
)
else:
d = spec.to_public_dict()
d["submission_id"] = ray_sid
spec2 = AdvancedTaskSpec.from_dict(d)
submit = lambda: self.tool.submit_advanced(spec2, no_wait=True, job_dir=job_dir, user_id=user_id_s)
submit = lambda: self.tool.submit_advanced(
spec2, no_wait=True, job_dir=job_dir, user_id=user_id_s, extra_env_vars=tracking_env
)
try:
submitted = submit()
# submitted should equal ray_sid; keep as source of truth.
self.db.update_attempt(task_id=task_id, attempt_no=attempt_no, ray_status="SUBMITTED")
self.db.update_attempt(
task_id=task_id,
attempt_no=attempt_no,
ray_status="SUBMITTED",
message=wandb_warn,
)
self.db.set_task_state(task_id=task_id, state="SUBMITTED")
if submitted != ray_sid:
self.db.set_task_state(task_id=task_id, state="SUBMITTED", event_type="WARN_SUBMISSION_ID_MISMATCH")

139
src/mvp/py/argus/service/ui.py
View File

@@ -188,11 +188,29 @@ def register_ui_routes(app: FastAPI) -> None:
<div style="height:10px"></div>
<pre id="me" class="muted">(not loaded)</pre>
</div>
<div style="height:12px"></div>
<div class="card">
<h3 style="margin-top:0">W&B</h3>
<div class="muted">Weights &amp; Biases local server (v3.6). Metrics are written by training jobs; this UI is for viewing.</div>
<div style="height:10px"></div>
<div class="row">
<a class="btn" id="wandb-open" target="_blank" rel="noopener" href="#">Open W&amp;B (:8090)</a>
<button class="btn" id="wandb-copy-project">Copy project</button>
</div>
<div style="height:10px"></div>
<div class="muted">project: <code id="wandb-project">(unknown)</code></div>
<div class="muted">base_url (job runtime): <code id="wandb-base-url">(unknown)</code></div>
</div>
""".strip()
script = """
const tokEl = document.getElementById("tok");
const msg = document.getElementById("msg");
const me = document.getElementById("me");
const wandbOpen = document.getElementById("wandb-open");
const wandbProject = document.getElementById("wandb-project");
const wandbBaseUrl = document.getElementById("wandb-base-url");
document.getElementById("wandb-copy-project").onclick = async () => { await copyText(wandbProject.textContent || ""); };
wandbOpen.href = curOriginWithPort(8090);
tokEl.value = mvpTokenGet();
async function refreshMe() {
@@ -200,8 +218,17 @@ async function refreshMe() {
try {
const obj = await apiJson("/api/v2/me");
me.textContent = fmtJson(obj);
if (obj.wandb && obj.wandb.enabled) {
wandbProject.textContent = obj.wandb.project_name || "(unknown)";
wandbBaseUrl.textContent = obj.wandb.base_url || "(unknown)";
} else {
wandbProject.textContent = "(disabled)";
wandbBaseUrl.textContent = "(disabled)";
}
} catch (e) {
me.textContent = "Error: " + (e.status || "") + "\\n" + (e.body || String(e));
wandbProject.textContent = "(error)";
wandbBaseUrl.textContent = "(error)";
}
}
@@ -358,19 +385,26 @@ val_file: /private/common/datasets/gsm8k_sft/test.parquet # 验证数据（必
# trainer_device: cpu # 仅 SFT 生效：driver 侧 device（可选，默认：cpu）
# submission_id: "" # Ray submission_id（可选，默认空；通常由服务自动生成，无需填写）
""".strip()
adv = """# Advanced TaskSpec (YAML) - v3.5
adv = """# Advanced TaskSpec (YAML) - v3.6
kind: advanced # 任务类型（必填）：advanced（自定义 command）
# 说明：v3.5 中 Advanced 任务不会按 ppo/grpo/sft 分类；平台统一按 "advanced" 做任务分类与 task_id 命名。
nnodes: 2 # 训练节点数（必填）：用于平台队列调度与资源预检查
n_gpus_per_node: 4 # 每节点 GPU 数（必填）：用于平台队列调度与资源预检查
# 自定义训练命令（必填）：平台会做 $HOME 宏替换：
# 说明：平台统一按 "advanced" 做任务分类与 task_id 命名（不按 ppo/grpo/sft 细分）。
#
# 自定义训练命令：平台会做 $HOME 宏替换：
# - $HOME -> /private/users/<user>
# - $HOME/common/datasets -> /private/datasets（共享只读数据）
# - $HOME/common/hf -> /private/hf（共享只读 HF cache）
#
# W&B（v3.6）：
# - 平台会在 runtime_env 注入 WANDB_BASE_URL/WANDB_API_KEY/WANDB_DIR
# - 以及注入以下 env，供 Advanced command 使用（无需用户手动修改 project）：
# - MVP_TRAINER_LOGGER: "console" 或 "[console,wandb]"
# - MVP_WANDB_PROJECT: "<user_id>_project"
# - MVP_WANDB_RUN: "<ray_submission_id>"
#
nnodes: 2 # 训练节点数（必填）：用于平台队列调度与资源预检查
n_gpus_per_node: 4 # 每节点 GPU 数（必填）：用于平台队列调度与资源预检查
command: |
# 注意：PPO 需要一些关键参数，否则 VERL 会在启动前 fail-fast（例如 actor 的 micro batch）。
PYTHONUNBUFFERED=1 \
python3 -m verl.trainer.main_ppo \
data.train_files=$HOME/common/datasets/gsm8k/train.parquet \
@@ -391,7 +425,9 @@ command: |
critic.model.path=Qwen/Qwen2.5-0.5B-Instruct \
critic.ppo_micro_batch_size_per_gpu=4 \
algorithm.kl_ctrl.kl_coef=0.001 \
trainer.logger=console \
trainer.logger=${MVP_TRAINER_LOGGER} \
trainer.project_name=${MVP_WANDB_PROJECT} \
trainer.experiment_name=${MVP_WANDB_RUN} \
trainer.val_before_train=False \
trainer.nnodes=2 \
trainer.n_gpus_per_node=4 \
@@ -426,6 +462,32 @@ command: |
--backend fsdp \
--local_dir $HOME/jobs/<RAY_SUBMISSION_ID>/checkpoints/<GLOBAL_STEP>/actor \
--target_dir $HOME/jobs/<RAY_SUBMISSION_ID>/checkpoints/<GLOBAL_STEP>/actor/huggingface
""".strip()
evaluation = """# Evaluation (YAML) - v3.6 (Advanced command)
# 用途：对“已生成的结果 parquet”做离线评估（基于 custom reward function），示例使用 VERL 内置 main_eval。
#
# 说明：
# - 你需要准备一个包含生成 responses 的 parquet，并替换 <EVAL_PARQUET>（示例放在某个 job 目录下）。
# - main_eval 会在 Ray 上并发计算，因此这里也用 +ray_kwargs 连接已有 Ray 集群。
#
kind: advanced
nnodes: 1
n_gpus_per_node: 0
command: |
PYTHONUNBUFFERED=1 \
python3 -m verl.trainer.main_eval \
data.path=$HOME/jobs/<RAY_SUBMISSION_ID>/outputs/<EVAL_PARQUET>.parquet \
+ray_kwargs.ray_init.address=auto
# 可选：指定 parquet schema（按你的 parquet 列名调整）
# data.response_key=responses
# data.data_source_key=data_source
# data.reward_model_key=reward_model
#
# 可选：自定义 reward（方式 A：用户自行提供 reward.py）
# custom_reward_function.path=$HOME/code/reward.py
# custom_reward_function.name=compute_score
""".strip()
body = f"""
<h1>New Task</h1>
@@ -447,6 +509,7 @@ command: |
<button class="btn" id="tpl-grpo">GRPO example</button>
<button class="btn" id="tpl-sft">SFT example</button>
<button class="btn" id="tpl-adv">Advanced example</button>
<button class="btn" id="tpl-eval">Evaluation example</button>
<button class="btn" id="tpl-merge">Model merge example</button>
</div>
<div style="height:10px"></div>
@@ -603,21 +666,24 @@ command: |
tpl_grpo = json.dumps(grpo)
tpl_sft = json.dumps(sft)
tpl_adv = json.dumps(adv)
tpl_eval = json.dumps(evaluation)
tpl_merge = json.dumps(merge)
script = (
"""
const msg = document.getElementById("msg");
const yamlEl = document.getElementById("yaml");
const TPL_PPO = __TPL_PPO__;
const TPL_GRPO = __TPL_GRPO__;
const TPL_SFT = __TPL_SFT__;
const TPL_ADV = __TPL_ADV__;
const TPL_MERGE = __TPL_MERGE__;
document.getElementById("tpl-ppo").onclick = () => { yamlEl.value = TPL_PPO; msg.textContent = ""; };
document.getElementById("tpl-grpo").onclick = () => { yamlEl.value = TPL_GRPO; msg.textContent = ""; };
document.getElementById("tpl-sft").onclick = () => { yamlEl.value = TPL_SFT; msg.textContent = ""; };
document.getElementById("tpl-adv").onclick = () => { yamlEl.value = TPL_ADV; msg.textContent = ""; };
document.getElementById("tpl-merge").onclick = () => { yamlEl.value = TPL_MERGE; msg.textContent = ""; };
const msg = document.getElementById("msg");
const yamlEl = document.getElementById("yaml");
const TPL_PPO = __TPL_PPO__;
const TPL_GRPO = __TPL_GRPO__;
const TPL_SFT = __TPL_SFT__;
const TPL_ADV = __TPL_ADV__;
const TPL_EVAL = __TPL_EVAL__;
const TPL_MERGE = __TPL_MERGE__;
document.getElementById("tpl-ppo").onclick = () => { yamlEl.value = TPL_PPO; msg.textContent = ""; };
document.getElementById("tpl-grpo").onclick = () => { yamlEl.value = TPL_GRPO; msg.textContent = ""; };
document.getElementById("tpl-sft").onclick = () => { yamlEl.value = TPL_SFT; msg.textContent = ""; };
document.getElementById("tpl-adv").onclick = () => { yamlEl.value = TPL_ADV; msg.textContent = ""; };
document.getElementById("tpl-eval").onclick = () => { yamlEl.value = TPL_EVAL; msg.textContent = ""; };
document.getElementById("tpl-merge").onclick = () => { yamlEl.value = TPL_MERGE; msg.textContent = ""; };
function yamlQuote(s) {
s = String(s ?? "");
@@ -802,11 +868,12 @@ document.getElementById("submit").onclick = async () => {
msg.textContent = "OK: " + fmtJson(obj);
if (obj.task_id) window.location.href = "/ui/tasks/" + obj.task_id;
};
""".strip()
""".strip()
.replace("__TPL_PPO__", tpl_ppo)
.replace("__TPL_GRPO__", tpl_grpo)
.replace("__TPL_SFT__", tpl_sft)
.replace("__TPL_ADV__", tpl_adv)
.replace("__TPL_EVAL__", tpl_eval)
.replace("__TPL_MERGE__", tpl_merge)
)
return HTMLResponse(content=_page("New Task", "new", body, script))
@@ -827,6 +894,19 @@ document.getElementById("submit").onclick = async () => {
<pre id="out" class="muted">Loading...</pre>
</div>
<div style="height:12px"></div>
<div class="card">
<h3 style="margin-top:0">W&amp;B</h3>
<div class="muted">This is a best-effort hint. Run name maps to <code>ray_submission_id</code> (attempt).</div>
<div style="height:10px"></div>
<div class="row">
<a class="btn" id="wandb-open" target="_blank" rel="noopener" href="#">Open W&amp;B (:8090)</a>
<button class="btn" id="wandb-copy-run">Copy run</button>
</div>
<div style="height:10px"></div>
<div class="muted">project: <code id="wandb-project">(unknown)</code></div>
<div class="muted">run: <code id="wandb-run">(unknown)</code></div>
</div>
<div style="height:12px"></div>
<div class="card">
<h3 style="margin-top:0">TaskSpec (YAML)</h3>
<div class="muted">Resolved TaskSpec (includes default values; submission_id reflects latest attempt when available).</div>
@@ -836,6 +916,8 @@ document.getElementById("submit").onclick = async () => {
""".strip()
script = f"""
document.getElementById("nav-ray-dashboard").href = curOriginWithPort(8265);
document.getElementById("wandb-open").href = curOriginWithPort(8090);
document.getElementById("wandb-copy-run").onclick = async () => {{ await copyText((document.getElementById("wandb-run").textContent || \"\").trim()); }};
const out = document.getElementById("out");
const spec = document.getElementById("spec");
async function refresh() {{
@@ -844,7 +926,18 @@ async function refresh() {{
const resp = await apiFetch("/api/v2/tasks/{task_id}");
const text = await resp.text();
if (!resp.ok) {{ out.textContent = "Error: " + resp.status + "\\n" + text; return; }}
out.textContent = fmtJson(JSON.parse(text));
const obj = JSON.parse(text);
out.textContent = fmtJson(obj);
const p = document.getElementById("wandb-project");
const r = document.getElementById("wandb-run");
const w = (obj.latest_attempt && obj.latest_attempt.wandb) ? obj.latest_attempt.wandb : null;
if (w) {{
p.textContent = w.project_name || "(unknown)";
r.textContent = w.run_name || "(unknown)";
}} else {{
p.textContent = "(not available)";
r.textContent = "(not available)";
}}
const resp2 = await apiFetch("/api/v2/tasks/{task_id}/spec");
const text2 = await resp2.text();

31
src/mvp/py/tests/test_app.py
View File

@@ -164,6 +164,37 @@ def test_task_submit_get_cancel_logs_queue(tmp_path: Path, monkeypatch):
assert r5.text.strip() == "c"
def test_me_includes_wandb_when_enabled(tmp_path: Path, monkeypatch):
from argus.service import app as app_mod
cfg_path = _write_config(tmp_path)
# Enable tracking.wandb
cfg = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
cfg["tracking"] = {"wandb": {"enabled": True, "base_url": "http://172.22.0.1:8090"}}
cfg_path.write_text(yaml.safe_dump(cfg), encoding="utf-8")
monkeypatch.setenv("MVP_INTERNAL_TOKEN", "token1")
class _Scheduler:
def __init__(self, **kwargs):
self.tool = object()
def run_forever(self, stop_flag):
return None
monkeypatch.setattr(app_mod, "Scheduler", _Scheduler)
app = app_mod.create_app(str(cfg_path))
with TestClient(app) as c:
r = c.get("/api/v2/me", headers={"authorization": "Bearer token1"})
assert r.status_code == 200
obj = r.json()
assert obj["user_id"] == "admin"
assert obj.get("wandb", {}).get("enabled") is True
assert obj["wandb"]["base_url"] == "http://172.22.0.1:8090"
assert obj["wandb"]["project_name"] == "admin_project"
def test_submit_rejects_non_mapping_jobspec(tmp_path: Path, monkeypatch):
from argus.service import app as app_mod

276
src/mvp/py/tests/test_scheduler.py
View File

@@ -60,10 +60,21 @@ def test_tick_submits_one_task(monkeypatch, tmp_path: Path):
def __init__(self, cfg):
self.submitted = []
self.job_dirs = []
self.extra_env = []
self.extra_overrides = []
def submit(self, spec, no_wait: bool, job_dir: str | None = None):
def submit(
self,
spec,
no_wait: bool,
job_dir: str | None = None,
extra_env_vars: dict | None = None,
extra_overrides: list[str] | None = None,
):
self.submitted.append(spec.submission_id)
self.job_dirs.append(job_dir)
self.extra_env.append(extra_env_vars)
self.extra_overrides.append(extra_overrides)
return str(spec.submission_id)
def status(self, submission_id: str):
@@ -83,6 +94,11 @@ def test_tick_submits_one_task(monkeypatch, tmp_path: Path):
assert len(attempts) == 1
assert attempts[0]["ray_submission_id"] == "t1--a01"
assert s.tool.job_dirs[-1] == "/private/users/alice/jobs/t1--a01"
env = s.tool.extra_env[-1]
assert env and env["MVP_WANDB_PROJECT"] == "alice_project"
assert env["MVP_WANDB_RUN"] == "t1--a01"
assert env["MVP_TRAINER_LOGGER"] == "console"
assert s.tool.extra_overrides[-1] is None
def test_tick_submits_one_task_advanced(monkeypatch, tmp_path: Path):
@@ -118,13 +134,23 @@ def test_tick_submits_one_task_advanced(monkeypatch, tmp_path: Path):
def __init__(self, cfg):
self.submitted = []
self.job_dirs = []
self.extra_env = []
def submit(self, spec, no_wait: bool, job_dir: str | None = None):
raise AssertionError("should not call submit() for advanced")
def submit_advanced(self, spec, no_wait: bool, job_dir: str | None = None, user_id: str | None = None):
def submit_advanced(
self,
spec,
no_wait: bool,
job_dir: str | None = None,
user_id: str | None = None,
extra_env_vars: dict | None = None,
extra_overrides: list[str] | None = None,
):
self.submitted.append(spec.submission_id)
self.job_dirs.append(job_dir)
self.extra_env.append(extra_env_vars)
return str(spec.submission_id)
def status(self, submission_id: str):
@@ -144,6 +170,252 @@ def test_tick_submits_one_task_advanced(monkeypatch, tmp_path: Path):
assert len(attempts) == 1
assert attempts[0]["ray_submission_id"] == "t1--a01"
assert s.tool.job_dirs[-1] == "/private/users/alice/jobs/t1--a01"
env = s.tool.extra_env[-1]
assert env and env["MVP_WANDB_PROJECT"] == "alice_project"
assert env["MVP_WANDB_RUN"] == "t1--a01"
assert env["MVP_TRAINER_LOGGER"] == "console"
def test_tick_injects_wandb_env_when_enabled(monkeypatch, tmp_path: Path):
from argus.service import scheduler as sched_mod
root = {
"ray": {
"address": "http://127.0.0.1:8265",
"shared_root": "/private",
"entrypoint_resources": {"worker_node": 1},
"runtime_env": {"env_vars": {}},
},
"service": {
"sqlite": {"db_path": str(tmp_path / "mvp.sqlite3")},
"scheduler": {"tick_s": 1, "retry_interval_s": 1, "max_running_tasks": 1},
},
"tracking": {"wandb": {"enabled": True, "base_url": "http://172.22.0.1:8090"}},
}
ray_cfg, v2_cfg = RayConfig.from_dict(root), V2Config.from_root_dict(root)
db = Db(v2_cfg.sqlite.db_path)
db.init()
db.create_task_v25(
task_id="t1",
user_id="alice",
workload="ppo",
jobspec_yaml=yaml.safe_dump(
{
"workload": "ppo",
"code_path": "/private/common/code/verl",
"model_id": "m",
"train_file": "/private/common/datasets/t",
"val_file": "/private/common/datasets/v",
}
),
nnodes=2,
n_gpus_per_node=4,
)
monkeypatch.setenv("WANDB_API_KEY", "k")
monkeypatch.setattr(sched_mod, "ensure_ray_connected", lambda: None)
monkeypatch.setattr(
sched_mod,
"get_cluster_available",
lambda: SimpleNamespace(total_available_gpus=999.0, total_available_npus=0.0),
)
class _Tool:
def __init__(self, cfg):
self.extra_env = []
self.job_dirs = []
def submit(
self,
spec,
no_wait: bool,
job_dir: str | None = None,
extra_env_vars: dict | None = None,
extra_overrides: list[str] | None = None,
):
self.job_dirs.append(job_dir)
self.extra_env.append(extra_env_vars)
return str(spec.submission_id)
def status(self, submission_id: str):
return "RUNNING"
def logs(self, submission_id: str):
return ""
monkeypatch.setattr(sched_mod, "RayJobTool", _Tool)
s = sched_mod.Scheduler(db=db, ray_cfg=ray_cfg, v2_cfg=v2_cfg)
s.tick()
assert s.tool.job_dirs[-1] == "/private/users/alice/jobs/t1--a01"
env = s.tool.extra_env[-1]
assert env and env["WANDB_API_KEY"] == "k"
assert env["WANDB_BASE_URL"] == "http://172.22.0.1:8090"
assert env["WANDB_MODE"] == "online"
assert env["WANDB_DIR"] == "/private/users/alice/jobs/t1--a01/wandb"
assert env["MVP_WANDB_PROJECT"] == "alice_project"
assert env["MVP_WANDB_RUN"] == "t1--a01"
assert env["MVP_TRAINER_LOGGER"] == "[console,wandb]"
def test_tick_wandb_enabled_missing_key_degrades(monkeypatch, tmp_path: Path):
from argus.service import scheduler as sched_mod
root = {
"ray": {
"address": "http://127.0.0.1:8265",
"shared_root": "/private",
"entrypoint_resources": {"worker_node": 1},
"runtime_env": {"env_vars": {}},
},
"service": {
"sqlite": {"db_path": str(tmp_path / "mvp.sqlite3")},
"scheduler": {"tick_s": 1, "retry_interval_s": 1, "max_running_tasks": 1},
},
"tracking": {"wandb": {"enabled": True, "base_url": "http://172.22.0.1:8090"}},
}
ray_cfg, v2_cfg = RayConfig.from_dict(root), V2Config.from_root_dict(root)
db = Db(v2_cfg.sqlite.db_path)
db.init()
db.create_task_v25(
task_id="t1",
user_id="alice",
workload="ppo",
jobspec_yaml=yaml.safe_dump(
{
"workload": "ppo",
"code_path": "/private/common/code/verl",
"model_id": "m",
"train_file": "/private/common/datasets/t",
"val_file": "/private/common/datasets/v",
}
),
nnodes=2,
n_gpus_per_node=4,
)
monkeypatch.delenv("WANDB_API_KEY", raising=False)
monkeypatch.setattr(sched_mod, "ensure_ray_connected", lambda: None)
monkeypatch.setattr(
sched_mod,
"get_cluster_available",
lambda: SimpleNamespace(total_available_gpus=999.0, total_available_npus=0.0),
)
class _Tool:
def __init__(self, cfg):
self.extra_env = []
def submit(
self,
spec,
no_wait: bool,
job_dir: str | None = None,
extra_env_vars: dict | None = None,
extra_overrides: list[str] | None = None,
):
self.extra_env.append(extra_env_vars)
return str(spec.submission_id)
def status(self, submission_id: str):
return "RUNNING"
def logs(self, submission_id: str):
return ""
monkeypatch.setattr(sched_mod, "RayJobTool", _Tool)
s = sched_mod.Scheduler(db=db, ray_cfg=ray_cfg, v2_cfg=v2_cfg)
s.tick()
env = s.tool.extra_env[-1]
assert env and env["MVP_WANDB_PROJECT"] == "alice_project"
assert env["MVP_WANDB_RUN"] == "t1--a01"
assert env["MVP_TRAINER_LOGGER"] == "console"
attempts = db.list_attempts("t1")
assert attempts and "WANDB_API_KEY" in str(attempts[0].get("message") or "")
def test_tick_wandb_enabled_updates_jobspec_logger(monkeypatch, tmp_path: Path):
from argus.service import scheduler as sched_mod
root = {
"ray": {
"address": "http://127.0.0.1:8265",
"shared_root": "/private",
"entrypoint_resources": {"worker_node": 1},
"runtime_env": {"env_vars": {}},
},
"service": {
"sqlite": {"db_path": str(tmp_path / "mvp.sqlite3")},
"scheduler": {"tick_s": 1, "retry_interval_s": 1, "max_running_tasks": 1},
},
"tracking": {"wandb": {"enabled": True, "base_url": "http://172.22.0.1:8090"}},
}
ray_cfg, v2_cfg = RayConfig.from_dict(root), V2Config.from_root_dict(root)
db = Db(v2_cfg.sqlite.db_path)
db.init()
db.create_task_v25(
task_id="t1",
user_id="alice",
workload="ppo",
jobspec_yaml=yaml.safe_dump(
{
"workload": "ppo",
"code_path": "/private/common/code/verl",
"model_id": "m",
"train_file": "/private/common/datasets/t",
"val_file": "/private/common/datasets/v",
}
),
nnodes=2,
n_gpus_per_node=4,
)
monkeypatch.setenv("WANDB_API_KEY", "k")
monkeypatch.setattr(sched_mod, "ensure_ray_connected", lambda: None)
monkeypatch.setattr(
sched_mod,
"get_cluster_available",
lambda: SimpleNamespace(total_available_gpus=999.0, total_available_npus=0.0),
)
class _Tool:
def __init__(self, cfg):
self.specs = []
self.extra_overrides = []
def submit(
self,
spec,
no_wait: bool,
job_dir: str | None = None,
extra_env_vars: dict | None = None,
extra_overrides: list[str] | None = None,
):
self.specs.append(spec)
self.extra_overrides.append(extra_overrides)
return str(spec.submission_id)
def status(self, submission_id: str):
return "RUNNING"
def logs(self, submission_id: str):
return ""
monkeypatch.setattr(sched_mod, "RayJobTool", _Tool)
s = sched_mod.Scheduler(db=db, ray_cfg=ray_cfg, v2_cfg=v2_cfg)
s.tick()
spec = s.tool.specs[-1]
# v3.6: built-in training should be configured to use wandb when enabled and API key exists.
assert spec.workload == "ppo"
ov = s.tool.extra_overrides[-1]
assert ov and "trainer.logger=[console,wandb]" in ov
assert ov and "trainer.project_name=alice_project" in ov
assert ov and "trainer.experiment_name=t1--a01" in ov
def test_tick_marks_pending_resources(monkeypatch, tmp_path: Path):

48
src/mvp/py/tests/test_service_config.py
View File

@@ -22,6 +22,7 @@ def test_v2_config_from_root_dict_new_format_defaults():
assert cfg.auth.token_env == "X"
assert cfg.sqlite.db_path.endswith(".sqlite3")
assert cfg.scheduler.max_running_tasks == 3
assert cfg.tracking.wandb.enabled is False
def test_v2_config_backward_compat_v2_section_and_default_db_path():
@@ -29,6 +30,7 @@ def test_v2_config_backward_compat_v2_section_and_default_db_path():
cfg = V2Config.from_root_dict({"shared_root": "/private", "v2": {"sqlite": {}}})
assert cfg.sqlite.db_path == "/private/common/db/mvp.sqlite3"
assert cfg.tracking.wandb.enabled is False
def test_v2_config_requires_mappings():
@@ -53,3 +55,49 @@ def test_v2_config_requires_data_mappings():
with pytest.raises(ValueError, match="config\\.data\\.\\{sftpgo,retention\\} must be mappings"):
V2Config.from_root_dict({**base, "data": {"sftpgo": ["x"], "retention": {}}})
def test_tracking_wandb_defaults_disabled():
from argus.service.config import V2Config
cfg = V2Config.from_root_dict(
{
"ray": {"shared_root": "/private"},
"service": {"api": {}, "auth": {}, "sqlite": {}, "scheduler": {}},
"data": {"sftpgo": {}, "retention": {}},
}
)
assert cfg.tracking.wandb.enabled is False
assert cfg.tracking.wandb.base_url == ""
assert cfg.tracking.wandb.api_key_env == "WANDB_API_KEY"
assert cfg.tracking.wandb.project_suffix == "_project"
def test_tracking_wandb_enabled_parses():
from argus.service.config import V2Config
cfg = V2Config.from_root_dict(
{
"ray": {"shared_root": "/private"},
"service": {"api": {}, "auth": {}, "sqlite": {}, "scheduler": {}},
"tracking": {"wandb": {"enabled": True, "base_url": "http://127.0.0.1:8090", "api_key_env": "X"}},
"data": {"sftpgo": {}, "retention": {}},
}
)
assert cfg.tracking.wandb.enabled is True
assert cfg.tracking.wandb.base_url == "http://127.0.0.1:8090"
assert cfg.tracking.wandb.api_key_env == "X"
def test_tracking_wandb_enabled_requires_base_url():
from argus.service.config import V2Config
with pytest.raises(ValueError, match="tracking\\.wandb\\.base_url is required"):
V2Config.from_root_dict(
{
"ray": {"shared_root": "/private"},
"service": {"api": {}, "auth": {}, "sqlite": {}, "scheduler": {}},
"tracking": {"wandb": {"enabled": True, "base_url": ""}},
"data": {"sftpgo": {}, "retention": {}},
}
)

30
src/mvp/py/tests/test_ui.py
View File

@@ -93,5 +93,35 @@ def test_ui_new_task_contains_advanced_example_snippet(tmp_path, monkeypatch):
# workload is not needed for advanced in v3.5.
assert "# workload:" not in r.text
assert "actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu" in r.text
# v3.6: Advanced example uses platform-injected env vars so users don't need to edit W&B project/run.
assert "trainer.logger=${MVP_TRAINER_LOGGER}" in r.text
assert "trainer.project_name=${MVP_WANDB_PROJECT}" in r.text
assert "trainer.experiment_name=${MVP_WANDB_RUN}" in r.text
assert "Model merge example" in r.text
assert "verl.model_merger" in r.text
def test_ui_new_task_contains_evaluation_example(tmp_path, monkeypatch):
cfg = _write_config(tmp_path)
monkeypatch.setenv("MVP_INTERNAL_TOKEN", "admin-token")
app = create_app(str(cfg))
c = TestClient(app)
r = c.get("/ui/tasks/new")
assert r.status_code == 200
assert "Evaluation example" in r.text
assert "verl.trainer.main_eval" in r.text
assert "data.path=$HOME/jobs/<RAY_SUBMISSION_ID>/outputs/<EVAL_PARQUET>.parquet" in r.text
assert "+ray_kwargs.ray_init.address=auto" in r.text
def test_ui_login_contains_wandb_section(tmp_path, monkeypatch):
cfg = _write_config(tmp_path)
monkeypatch.setenv("MVP_INTERNAL_TOKEN", "admin-token")
app = create_app(str(cfg))
c = TestClient(app)
r = c.get("/ui/login")
assert r.status_code == 200
assert "W&amp;B" in r.text
assert "Open W&amp;B" in r.text

4
src/mvp/scripts/00_prereq_check.sh
View File

@@ -31,6 +31,10 @@ echo "[host] ensure shared dirs exist under ../../shared"
mkdir -p "${ROOT_DIR}/../../shared"/{datasets,hf,jobs,outputs,ray,common,user}
mkdir -p "${ROOT_DIR}/../../shared/common"/{code,datasets,models}
mkdir -p "${ROOT_DIR}/../../shared/user"/{code}
# v3.6: W&B local server persists metadata under /private/common/wandb (mounted as /vol).
# The wandb container runs as a non-root user; ensure the bind mount is writable.
mkdir -p "${ROOT_DIR}/../../shared/common/wandb"
chmod 777 "${ROOT_DIR}/../../shared/common/wandb" 2>/dev/null || true
echo "[host] ensure verl repo exists under ../../verl (required by prepare scripts)"
if [[ ! -d "${ROOT_DIR}/../../verl" ]]; then

3
src/mvp/scripts/60_start_api.sh
View File

@@ -25,6 +25,9 @@ fi
env_args=(-e "MVP_INTERNAL_TOKEN=${MVP_INTERNAL_TOKEN}")
# If host does not provide it, fall back to the dev default used by docker-compose (kept in sync).
env_args+=(-e "SFTPGO_ADMIN_PASSWORD=${SFTPGO_ADMIN_PASSWORD:-my-dev-sftpgo-admin}")
if [[ -n "${WANDB_API_KEY:-}" ]]; then
env_args+=(-e "WANDB_API_KEY=${WANDB_API_KEY}")
fi
docker exec -d "${env_args[@]}" "${HEAD_CONTAINER}" bash -lc "nohup python3 /workspace/mvp/py/server.py --config '${CONFIG_IN_CONTAINER}' >>'${LOG_PATH}' 2>&1 & echo \$! >'${PID_PATH}'"