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

V3.5 支持Advanced模式，用户提供python命令

Browse Source
This commit is contained in:
yuyr
2026-01-04 15:28:41 +08:00
parent e3dcfe526f
commit 5d0d849c28
22 changed files with 1355 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/mvp/py/argus/ray/builders.py
View File

@@ -97,3 +97,12 @@ def build_training_argv(spec: JobSpec, submission_id: str, job_dir: str) -> Buil
return BuiltCommand(argv=argv)
raise ValueError(f"unsupported workload: {spec.workload}")
def build_advanced_argv(command: str) -> BuiltCommand:
"""
Execute an arbitrary user-provided command under bash for maximum shell compatibility
(multiline, env vars, line continuations, etc.).
"""
return BuiltCommand(argv=["bash", "-lc", command])

2
src/mvp/py/argus/ray/driver_entrypoint.py
View File

@@ -35,6 +35,8 @@ def main() -> int:
job_dir = Path(args.job_dir)
job_dir.mkdir(parents=True, exist_ok=True)
os.environ.setdefault("MVP_JOB_DIR", str(job_dir))
os.chdir(job_dir)
_preflight()

55
src/mvp/py/argus/ray/models.py
View File

@@ -18,6 +18,7 @@ class RayConfig:
entrypoint_resources: dict[str, float]
runtime_env_env_vars: dict[str, str]
user_code_path: str
verl_code_path: str
@staticmethod
def from_dict(d: dict[str, Any]) -> "RayConfig":
@@ -42,6 +43,7 @@ class RayConfig:
entrypoint_resources={str(k): float(v) for k, v in entrypoint_resources.items()},
runtime_env_env_vars={str(k): str(v) for k, v in env_vars.items()},
user_code_path=str(d.get("user_code_path", f"{_require(d, 'shared_root')}/user/code")),
verl_code_path=str(d.get("verl_code_path", f"{_require(d, 'shared_root')}/common/code/verl/verl_repo")),
)
def to_public_dict(self) -> dict[str, Any]:
@@ -52,6 +54,7 @@ class RayConfig:
"entrypoint_resources": self.entrypoint_resources,
"runtime_env": {"env_vars": self.runtime_env_env_vars},
"user_code_path": self.user_code_path,
"verl_code_path": self.verl_code_path,
}
@@ -126,3 +129,55 @@ class JobSpec:
if self.workload == "sft":
out["trainer_device"] = self.trainer_device or "cpu"
return out
@dataclass(frozen=True)
class AdvancedTaskSpec:
kind: str # advanced
workload: str # always "advanced" for classification/ID prefix
submission_id: str | None
nnodes: int
n_gpus_per_node: int
command: str
@staticmethod
def from_dict(d: dict[str, Any]) -> "AdvancedTaskSpec":
kind = str(_require(d, "kind"))
if kind != "advanced":
raise ValueError(f"unsupported taskspec kind: {kind}")
command = str(_require(d, "command"))
return AdvancedTaskSpec(
kind=kind,
workload="advanced",
submission_id=(str(d["submission_id"]) if d.get("submission_id") else None),
nnodes=int(_require(d, "nnodes")),
n_gpus_per_node=int(_require(d, "n_gpus_per_node")),
command=command,
)
def to_public_dict(self) -> dict[str, Any]:
return {
"kind": self.kind,
"workload": self.workload,
"submission_id": self.submission_id or "",
"nnodes": self.nnodes,
"n_gpus_per_node": self.n_gpus_per_node,
"command": self.command,
}
def parse_taskspec(d: dict[str, Any]) -> JobSpec | AdvancedTaskSpec:
"""
v3.x compatibility rule:
- If no "kind": treat as legacy/basic JobSpec.
- If kind == "advanced": parse AdvancedTaskSpec.
"""
kind = d.get("kind")
if kind in (None, ""):
return JobSpec.from_dict(d)
if kind == "advanced":
return AdvancedTaskSpec.from_dict(d)
raise ValueError(f"unsupported taskspec kind: {kind}")

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

@@ -10,8 +10,8 @@ from typing import Any
import ray
from ray.job_submission import JobSubmissionClient
from .builders import build_training_argv
from .models import JobSpec, RayConfig
from .builders import build_advanced_argv, build_training_argv
from .models import AdvancedTaskSpec, JobSpec, RayConfig
from .yaml_io import dump_yaml
@@ -45,6 +45,9 @@ class RayJobTool:
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_for(self, *, code_path: str, workload: str, extra_pythonpath: list[str] | None = None) -> dict[str, Any]:
env_vars = dict(self.cfg.runtime_env_env_vars)
# Default HF cache
@@ -57,18 +60,18 @@ class RayJobTool:
# Place it before verl code to avoid interfering with verl import priority.
tool_code_path = os.environ.get("MVP_TOOL_CODE_PATH", "/workspace/mvp/py")
user_code_path = self.cfg.user_code_path
code_path = spec.code_path
existing = env_vars.get("PYTHONPATH", "")
prefix = f"{tool_code_path}:{code_path}:{user_code_path}"
base_parts = [tool_code_path, code_path, self.cfg.user_code_path]
if extra_pythonpath:
base_parts.extend(extra_pythonpath)
prefix = ":".join(base_parts)
env_vars["PYTHONPATH"] = f"{prefix}:{existing}" if existing else prefix
# For debugging / log visibility
env_vars["MVP_CODE_PATH"] = code_path
# SFT: ensure ray.init() connects to the cluster
if spec.workload == "sft":
if workload == "sft":
env_vars.setdefault("RAY_ADDRESS", "auto")
return {"env_vars": env_vars}
@@ -146,6 +149,91 @@ class RayJobTool:
return submitted
def submit_advanced(
self,
spec: AdvancedTaskSpec,
no_wait: bool,
job_dir: str | None = None,
user_id: 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_advanced_argv(spec.command)
entrypoint_argv = [
"python3",
"-m",
"argus.ray.driver_entrypoint",
"--job-dir",
job_dir,
*built.argv,
]
entrypoint = " ".join(shlex.quote(x) for x in entrypoint_argv)
extra_pythonpath: list[str] = []
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)
# Prepare job artifacts directory
job_root = Path(job_dir)
_mkdir(job_root / "config")
_mkdir(job_root / "logs")
_mkdir(job_root / "debug")
_mkdir(job_root / "checkpoints")
_write_text(job_root / "config" / "ray_config.yaml", dump_yaml(self.cfg.to_public_dict()))
_write_text(job_root / "config" / "taskspec.yaml", dump_yaml(spec.to_public_dict()))
_write_json(
job_root / "config" / "submit_payload.json",
{
"submission_id": submission_id,
"address": self.cfg.address,
"entrypoint": entrypoint,
"entrypoint_num_cpus": self.cfg.entrypoint_num_cpus,
"entrypoint_resources": self.cfg.entrypoint_resources,
"runtime_env": runtime_env,
},
)
# Pre-submit debug snapshot (ray cluster resources via ray.init)
try:
ray.init(address="auto", ignore_reinit_error=True, log_to_driver=False)
_write_json(job_root / "debug" / "ray_cluster_resources_pre.json", ray.cluster_resources())
_write_json(job_root / "debug" / "ray_available_resources_pre.json", ray.available_resources())
except Exception as e:
_write_text(job_root / "debug" / "ray_resources_pre.error.txt", repr(e) + "\n")
try:
submitted = self.client.submit_job(
entrypoint=entrypoint,
submission_id=submission_id,
runtime_env=runtime_env,
entrypoint_num_cpus=self.cfg.entrypoint_num_cpus,
entrypoint_resources=self.cfg.entrypoint_resources,
)
except Exception as e:
_write_text(job_root / "logs" / "submit.error.txt", repr(e) + "\n")
raise
_write_text(job_root / "config" / "ray_submission_id.txt", submitted + "\n")
# Post-submit debug snapshot via SDK
try:
jobs = self.client.list_jobs()
_write_text(
job_root / "debug" / "ray_job_list_post.json",
json.dumps([_job_details_to_dict(j) for j in jobs], indent=2) + "\n",
)
except Exception as e:
_write_text(job_root / "debug" / "ray_job_list_post.error.txt", repr(e) + "\n")
if not no_wait:
pass
return submitted
def status(self, submission_id: str) -> str:
return str(self.client.get_job_status(submission_id))

75
src/mvp/py/argus/service/advanced_command.py Normal file
View File

@@ -0,0 +1,75 @@
from __future__ import annotations
import re
_RE_PRIVATE_USER = re.compile(r"/private/users/([^/\s\"']+)")
_RE_TRAIN_FILES = re.compile(r"data\.train_files=([^\s\\]+)")
_RE_VAL_FILES = re.compile(r"data\.val_files=([^\s\\]+)")
_RE_REWARD_PATH = re.compile(r"custom_reward_function\.path=([^\s\\]+)")
def _strip_quotes(s: str) -> str:
s = s.strip()
if len(s) >= 2 and s[0] == s[-1] and s[0] in ("'", '"'):
return s[1:-1]
return s
def expand_advanced_command(*, user_id: str, command: str) -> str:
"""
Expand user-facing macros into real container paths.
Important: `$HOME/common/...` is a SFTPGo virtual folder; the training process
cannot see it. We map those paths to real shared mounts.
"""
home = f"/private/users/{user_id}"
expanded = command
expanded = expanded.replace("$HOME/common/datasets", "/private/datasets")
expanded = expanded.replace("$HOME/common/hf", "/private/hf")
expanded = expanded.replace("$HOME", home)
return expanded
def validate_advanced_command(*, user_id: str, expanded_command: str) -> None:
"""
Best-effort validation for advanced commands.
Strong constraints (raise ValueError on failure):
- Must include a python3 VERL trainer entrypoint
- Must not reference other users' /private/users/<other>/... paths
- If train/val files are present, must be under allowed roots
- If reward path is present, must be under /private/users/<me>/code/...
"""
if "python3" not in expanded_command or "-m verl.trainer." not in expanded_command:
raise ValueError("command must include 'python3' and '-m verl.trainer.'")
for other in _RE_PRIVATE_USER.findall(expanded_command):
if other != user_id:
raise ValueError(f"command references another user's path: /private/users/{other}/...")
allowed_dataset_prefixes = (
f"/private/users/{user_id}/datasets/",
"/private/datasets/",
# Backward compatible with v3.0 paths used by existing templates/users.
"/private/common/datasets/",
)
for m in _RE_TRAIN_FILES.finditer(expanded_command):
path = _strip_quotes(m.group(1))
if not path.startswith(allowed_dataset_prefixes):
raise ValueError(f"train_file must be under allowed roots, got: {path}")
for m in _RE_VAL_FILES.finditer(expanded_command):
path = _strip_quotes(m.group(1))
if not path.startswith(allowed_dataset_prefixes):
raise ValueError(f"val_file must be under allowed roots, got: {path}")
for m in _RE_REWARD_PATH.finditer(expanded_command):
path = _strip_quotes(m.group(1))
allow = f"/private/users/{user_id}/code/"
if not path.startswith(allow):
raise ValueError(f"reward path must be under {allow}, got: {path}")

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

@@ -9,8 +9,9 @@ import yaml
from fastapi import FastAPI, HTTPException, Request, Response
from argus.core.ids import new_task_id
from argus.ray.models import JobSpec, RayConfig
from argus.ray.models import AdvancedTaskSpec, JobSpec, RayConfig, parse_taskspec
from .advanced_command import expand_advanced_command, validate_advanced_command
from .config import V2Config
from .db import Db
from .janitor import JobsJanitor
@@ -316,23 +317,52 @@ def create_app(config_path: str) -> FastAPI:
async def submit_task(req: Request) -> dict[str, Any]:
subject = _auth(req)
body = (await req.body()).decode("utf-8")
obj = yaml.safe_load(body) or {}
try:
obj = yaml.safe_load(body) or {}
except Exception as e:
# yaml.YAMLError and similar should be a user-facing 400, not a 500.
raise HTTPException(status_code=400, detail=f"invalid YAML: {e!r}")
if not isinstance(obj, dict):
raise HTTPException(status_code=400, detail="jobspec must be a YAML mapping")
try:
spec = JobSpec.from_dict(obj)
spec = parse_taskspec(obj)
except Exception as e:
raise HTTPException(status_code=400, detail=f"invalid jobspec: {e!r}")
# v3.0 path policy:
root = ray_cfg.shared_root.rstrip("/")
user_id = str(subject["user_id"]).strip()
task_id = new_task_id(str(spec.workload), user_id=user_id)
# v3.5: Advanced TaskSpec (custom command)
if isinstance(spec, AdvancedTaskSpec):
expanded = expand_advanced_command(user_id=user_id, command=spec.command)
try:
validate_advanced_command(user_id=user_id, expanded_command=expanded)
except Exception as e:
raise HTTPException(status_code=400, detail=f"invalid command: {e!r}")
stored = spec.to_public_dict()
stored["command"] = expanded
stored_yaml = yaml.safe_dump(stored, sort_keys=False)
db.create_task_v25(
task_id=task_id,
user_id=user_id,
workload=spec.workload,
jobspec_yaml=stored_yaml,
nnodes=spec.nnodes,
n_gpus_per_node=spec.n_gpus_per_node,
)
return {"task_id": task_id, "state": "QUEUED"}
# v3.0 path policy (Basic JobSpec):
# - code_path: only allow /private/common/...
# - train/val: allow /private/common/datasets/... OR /private/users/<me>/datasets/...
# - model_id: if it looks like a local path (/private/...), allow only models dirs:
# /private/common/models/... OR /private/users/<me>/models/...
root = ray_cfg.shared_root.rstrip("/")
common_prefix = f"{root}/common/"
user_prefix = f"{root}/users/{str(subject['user_id']).strip()}/"
user_prefix = f"{root}/users/{user_id}/"
common_datasets_prefix = f"{common_prefix}datasets/"
user_datasets_prefix = f"{user_prefix}datasets/"
@@ -360,15 +390,17 @@ def create_app(config_path: str) -> FastAPI:
detail=f"model_id local path must start with {common_models_prefix} or {user_models_prefix}",
)
task_id = new_task_id(spec.workload, user_id=str(subject["user_id"]))
db.create_task_v25(
task_id=task_id,
user_id=str(subject["user_id"]),
workload=spec.workload,
jobspec_yaml=body,
nnodes=spec.nnodes,
n_gpus_per_node=spec.n_gpus_per_node,
)
try:
db.create_task_v25(
task_id=task_id,
user_id=user_id,
workload=spec.workload,
jobspec_yaml=body,
nnodes=spec.nnodes,
n_gpus_per_node=spec.n_gpus_per_node,
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"db create task failed: {e!r}")
return {"task_id": task_id, "state": "QUEUED"}
@app.get("/api/v2/tasks/{task_id}")
@@ -428,7 +460,7 @@ def create_app(config_path: str) -> FastAPI:
obj = yaml.safe_load(raw) or {}
if not isinstance(obj, dict):
raise ValueError("jobspec must be a YAML mapping")
spec = JobSpec.from_dict(obj)
spec = parse_taskspec(obj)
resolved = spec.to_public_dict()
attempts = db.list_attempts(task_id)
if attempts:

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

@@ -9,7 +9,7 @@ from typing import Any
import yaml
from argus.core.ids import attempt_submission_id
from argus.ray.models import JobSpec, RayConfig
from argus.ray.models import AdvancedTaskSpec, JobSpec, RayConfig, parse_taskspec
from argus.ray.ray_job_tool import RayJobTool
from .config import V2Config
@@ -48,11 +48,11 @@ class Scheduler:
required = float(nnodes * n_gpus_per_node)
return avail.total_available_gpus >= required
def _parse_jobspec(self, jobspec_yaml: str) -> JobSpec:
def _parse_taskspec(self, jobspec_yaml: str) -> JobSpec | AdvancedTaskSpec:
obj = yaml.safe_load(jobspec_yaml) or {}
if not isinstance(obj, dict):
raise ValueError("jobspec must be a YAML mapping")
return JobSpec.from_dict(obj)
return parse_taskspec(obj)
def _submit_one(self, task_row: dict[str, Any]) -> None:
task_id = str(task_row["task_id"])
@@ -60,7 +60,7 @@ class Scheduler:
user_id = task_row.get("user_id")
user_id_s = str(user_id) if user_id not in (None, "") else None
spec = self._parse_jobspec(jobspec_yaml)
spec = self._parse_taskspec(jobspec_yaml)
attempt_no = int(task_row.get("latest_attempt_no", 0)) + 1
ray_sid = attempt_submission_id(task_id, attempt_no)
job_dir = self._job_dir_for_task(user_id=user_id_s, ray_submission_id=ray_sid)
@@ -69,13 +69,20 @@ class Scheduler:
self.db.create_attempt(task_id=task_id, attempt_no=attempt_no, ray_submission_id=ray_sid)
self.db.set_task_state(task_id=task_id, state="SUBMITTING", latest_attempt_no=attempt_no)
# Override submission_id in jobspec (v1.1 compatible)
d = spec.to_public_dict()
d["submission_id"] = ray_sid
spec2 = JobSpec.from_dict(d)
# Override submission_id in taskspec (v1.1 compatible)
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)
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)
try:
submitted = self.tool.submit(spec2, no_wait=True, job_dir=job_dir)
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.set_task_state(task_id=task_id, state="SUBMITTED")

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

@@ -329,16 +329,71 @@ 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
kind: advanced # 任务类型（必填）：advanced（自定义 command）
# 说明：v3.5 中 Advanced 任务不会按 ppo/grpo/sft 分类；平台统一按 "advanced" 做任务分类与 task_id 命名。
nnodes: 2 # 训练节点数（必填）：用于平台队列调度与资源预检查
n_gpus_per_node: 4 # 每节点 GPU 数（必填）：用于平台队列调度与资源预检查
# 自定义训练命令（必填）：平台会做 $HOME 宏替换：
# - $HOME -> /private/users/<user>
# - $HOME/common/datasets -> /private/datasets（共享只读数据）
# - $HOME/common/hf -> /private/hf（共享只读 HF cache）
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 \
data.val_files=$HOME/common/datasets/gsm8k/test.parquet \
data.train_batch_size=256 \
data.max_prompt_length=512 \
data.max_response_length=512 \
actor_rollout_ref.model.path=Qwen/Qwen2.5-0.5B-Instruct \
actor_rollout_ref.actor.optim.lr=1e-6 \
actor_rollout_ref.actor.ppo_mini_batch_size=64 \
actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=4 \
actor_rollout_ref.rollout.name=sglang \
actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=8 \
actor_rollout_ref.rollout.tensor_model_parallel_size=1 \
actor_rollout_ref.rollout.gpu_memory_utilization=0.4 \
actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=4 \
critic.optim.lr=1e-5 \
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.val_before_train=False \
trainer.nnodes=2 \
trainer.n_gpus_per_node=4 \
trainer.total_epochs=1 \
trainer.total_training_steps=10 \
trainer.save_freq=10 \
trainer.test_freq=-1 \
trainer.resume_mode=disable \
trainer.default_local_dir=checkpoints \
+ray_kwargs.ray_init.address=auto \
hydra.run.dir=logs/hydra
# 可选：自定义 reward（方式 A：直接写在 command 里）
# command 里增加如下 overrides：
# custom_reward_function.path=$HOME/code/reward.py
# custom_reward_function.name=compute_score
""".strip()
body = f"""
<h1>New Task</h1>
<div class="card">
<div class="muted">Paste TaskSpec YAML and submit to API server. Note: <code>code_path</code> is required (v3.0 does not execute user code; use the common snapshot).</div>
<div class="muted">
Paste TaskSpec YAML and submit to API server.
Basic tasks require <code>code_path</code>; Advanced tasks use <code>kind: advanced</code> with a custom <code>command</code>.
</div>
<div style="height:10px"></div>
<div class="row">
<button class="btn" id="tpl-ppo">PPO example</button>
<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>
</div>
<div style="height:10px"></div>
<textarea id="yaml" rows="16">{html.escape(ppo)}</textarea>
@@ -354,6 +409,7 @@ val_file: /private/common/datasets/gsm8k_sft/test.parquet # 验证数据（必
tpl_ppo = json.dumps(ppo)
tpl_grpo = json.dumps(grpo)
tpl_sft = json.dumps(sft)
tpl_adv = json.dumps(adv)
script = (
"""
const msg = document.getElementById("msg");
@@ -361,9 +417,11 @@ 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__;
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("submit").onclick = async () => {
msg.textContent = "Submitting...";
const body = yamlEl.value;
@@ -378,6 +436,7 @@ document.getElementById("submit").onclick = async () => {
.replace("__TPL_PPO__", tpl_ppo)
.replace("__TPL_GRPO__", tpl_grpo)
.replace("__TPL_SFT__", tpl_sft)
.replace("__TPL_ADV__", tpl_adv)
)
return HTMLResponse(content=_page("New Task", "new", body, script))

74
src/mvp/py/tests/test_advanced_command.py Normal file
View File

@@ -0,0 +1,74 @@
from __future__ import annotations
import pytest
from argus.service.advanced_command import expand_advanced_command, validate_advanced_command
def test_expand_maps_home_common_datasets_to_private_datasets():
cmd = "python3 -m verl.trainer.main_ppo data.train_files=$HOME/common/datasets/gsm8k/train.parquet"
expanded = expand_advanced_command(user_id="alice", command=cmd)
assert "/private/datasets/gsm8k/train.parquet" in expanded
assert "$HOME/common/datasets" not in expanded
def test_expand_maps_home_common_hf_to_private_hf():
cmd = "python3 -m verl.trainer.main_ppo HF_HOME=$HOME/common/hf"
expanded = expand_advanced_command(user_id="alice", command=cmd)
assert "HF_HOME=/private/hf" in expanded
def test_expand_maps_home_to_private_users():
cmd = "python3 -m verl.trainer.main_ppo data.train_files=$HOME/datasets/x.parquet"
expanded = expand_advanced_command(user_id="alice", command=cmd)
assert "data.train_files=/private/users/alice/datasets/x.parquet" in expanded
def test_validate_requires_verl_trainer_entry():
with pytest.raises(ValueError, match="verl\\.trainer"):
validate_advanced_command(user_id="alice", expanded_command="echo hi")
def test_validate_rejects_cross_user_paths():
cmd = "python3 -m verl.trainer.main_ppo x=/private/users/bob/datasets/x"
with pytest.raises(ValueError, match="another user's path"):
validate_advanced_command(user_id="alice", expanded_command=cmd)
def test_validate_allows_train_val_under_allowed_roots():
cmd = (
"python3 -m verl.trainer.main_ppo "
"data.train_files=/private/datasets/gsm8k/train.parquet "
"data.val_files=/private/users/alice/datasets/gsm8k/test.parquet"
)
validate_advanced_command(user_id="alice", expanded_command=cmd)
def test_validate_rejects_train_val_outside_allowed_roots():
cmd = (
"python3 -m verl.trainer.main_ppo "
"data.train_files=/etc/passwd "
"data.val_files=/private/datasets/gsm8k/test.parquet"
)
with pytest.raises(ValueError, match="train_file must be under allowed roots"):
validate_advanced_command(user_id="alice", expanded_command=cmd)
def test_validate_allows_reward_path_under_user_code():
cmd = (
"python3 -m verl.trainer.main_ppo "
"custom_reward_function.path=/private/users/alice/code/reward.py "
"custom_reward_function.name=compute_score"
)
validate_advanced_command(user_id="alice", expanded_command=cmd)
def test_validate_rejects_reward_path_outside_user_code():
cmd = (
"python3 -m verl.trainer.main_ppo "
"custom_reward_function.path=/private/datasets/reward.py "
"custom_reward_function.name=compute_score"
)
with pytest.raises(ValueError, match="reward path must be under"):
validate_advanced_command(user_id="alice", expanded_command=cmd)

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

@@ -185,6 +185,29 @@ def test_submit_rejects_non_mapping_jobspec(tmp_path: Path, monkeypatch):
assert r.status_code == 400
def test_submit_rejects_invalid_yaml_returns_400(tmp_path: Path, monkeypatch):
from argus.service import app as app_mod
cfg_path = _write_config(tmp_path)
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:
# YAML parse error: unclosed quote
r = c.post("/api/v2/tasks", headers={"authorization": "Bearer token1"}, data="workload: \"ppo\n")
assert r.status_code == 400
assert "invalid YAML" in r.text
def test_submit_rejects_invalid_jobspec(tmp_path: Path, monkeypatch):
from argus.service import app as app_mod
@@ -206,6 +229,99 @@ def test_submit_rejects_invalid_jobspec(tmp_path: Path, monkeypatch):
assert r.status_code == 400
def test_submit_advanced_ok_and_command_expanded(tmp_path: Path, monkeypatch):
from argus.service import app as app_mod
cfg_path = _write_config(tmp_path)
monkeypatch.setenv("MVP_INTERNAL_TOKEN", "admin-token")
monkeypatch.setattr(app_mod, "new_task_id", lambda workload, **kwargs: "tid_adv")
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))
# seed user + token
from argus.service.config import V2Config
from argus.service.db import Db
root = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
v2_cfg = V2Config.from_root_dict(root)
db = Db(v2_cfg.sqlite.db_path)
db.create_user(user_id="u1", display_name=None)
token = db.issue_token(user_id="u1")
spec = """
kind: advanced
workload: ppo
nnodes: 2
n_gpus_per_node: 4
command: |
python3 -m verl.trainer.main_ppo \\
data.train_files=$HOME/common/datasets/gsm8k/train.parquet \\
data.val_files=$HOME/datasets/gsm8k/test.parquet \\
+ray_kwargs.ray_init.address=auto
""".lstrip()
with TestClient(app) as c:
r = c.post("/api/v2/tasks", headers={"authorization": f"Bearer {token}"}, data=spec)
assert r.status_code == 200
assert r.json()["task_id"] == "tid_adv"
r2 = c.get("/api/v2/tasks/tid_adv/spec", headers={"authorization": f"Bearer {token}"})
assert r2.status_code == 200
assert "kind: advanced" in r2.text
# Expanded mapping for SFTPGo "virtual common" folders:
assert "/private/datasets/gsm8k/train.parquet" in r2.text
assert "/private/users/u1/datasets/gsm8k/test.parquet" in r2.text
def test_submit_advanced_rejects_invalid_command(tmp_path: Path, monkeypatch):
from argus.service import app as app_mod
cfg_path = _write_config(tmp_path)
monkeypatch.setenv("MVP_INTERNAL_TOKEN", "admin-token")
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))
# seed user + token
from argus.service.config import V2Config
from argus.service.db import Db
root = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
v2_cfg = V2Config.from_root_dict(root)
db = Db(v2_cfg.sqlite.db_path)
db.create_user(user_id="u1", display_name=None)
token = db.issue_token(user_id="u1")
bad = """
kind: advanced
workload: ppo
nnodes: 2
n_gpus_per_node: 4
command: |
echo hello
""".lstrip()
with TestClient(app) as c:
r = c.post("/api/v2/tasks", headers={"authorization": f"Bearer {token}"}, data=bad)
assert r.status_code == 400
assert "invalid command" in r.text
def test_me_sftp_reset_password_disabled_returns_400(tmp_path: Path, monkeypatch):
from argus.service import app as app_mod

7
src/mvp/py/tests/test_builders.py
View File

@@ -56,3 +56,10 @@ def test_build_training_argv_unsupported_raises():
spec = _base_spec("bad")
with pytest.raises(ValueError, match="unsupported workload"):
build_training_argv(spec, submission_id="sid", job_dir="/job")
def test_build_advanced_argv_uses_bash_lc():
from argus.ray.builders import build_advanced_argv
built = build_advanced_argv("echo hi")
assert built.argv[:2] == ["bash", "-lc"]

4
src/mvp/py/tests/test_driver_entrypoint.py
View File

@@ -18,8 +18,10 @@ def test_driver_entrypoint_strips_double_dash_and_returns_code(monkeypatch, tmp_
returncode = 7
monkeypatch.setattr(mod.subprocess, "run", lambda cmd, check: _Proc())
got = {}
monkeypatch.setattr(mod.os, "chdir", lambda p: got.update({"cwd": str(p)}))
monkeypatch.setattr(sys, "argv", ["x", "--job-dir", str(tmp_path), "--", "echo", "hi"])
assert mod.main() == 7
assert (tmp_path).exists()
assert got["cwd"] == str(tmp_path)

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

@@ -31,6 +31,7 @@ def test_ray_config_from_dict_new_format_and_defaults():
assert cfg.entrypoint_resources["worker_node"] == 1.0
assert cfg.runtime_env_env_vars["HF_ENDPOINT"] == "x"
assert cfg.user_code_path == "/private/user/code"
assert cfg.verl_code_path == "/private/common/code/verl/verl_repo"
public = cfg.to_public_dict()
assert public["runtime_env"]["env_vars"]["HF_ENDPOINT"] == "x"
@@ -106,3 +107,50 @@ def test_jobspec_unsupported_workload():
JobSpec.from_dict(
{"workload": "nope", "code_path": "x", "model_id": "m", "train_file": "t", "val_file": "v"}
)
def test_parse_taskspec_basic_no_kind_compat():
from argus.ray.models import JobSpec, parse_taskspec
got = parse_taskspec(
{
"workload": "ppo",
"code_path": "/code",
"model_id": "m",
"train_file": "train.jsonl",
"val_file": "val.jsonl",
}
)
assert isinstance(got, JobSpec)
assert got.workload == "ppo"
def test_parse_taskspec_advanced_smoke():
from argus.ray.models import AdvancedTaskSpec, parse_taskspec
got = parse_taskspec(
{
"kind": "advanced",
"nnodes": 2,
"n_gpus_per_node": 4,
"command": "python3 -m verl.trainer.main_ppo +ray_kwargs.ray_init.address=auto",
}
)
assert isinstance(got, AdvancedTaskSpec)
assert got.kind == "advanced"
assert got.workload == "advanced"
assert got.nnodes == 2
assert got.n_gpus_per_node == 4
def test_parse_taskspec_advanced_requires_command_nnodes_gpus():
from argus.ray.models import parse_taskspec
with pytest.raises(ValueError, match="missing required field: command"):
parse_taskspec({"kind": "advanced", "nnodes": 1, "n_gpus_per_node": 1})
with pytest.raises(ValueError, match="missing required field: nnodes"):
parse_taskspec({"kind": "advanced", "n_gpus_per_node": 1, "command": "python3 -m verl.trainer.main_ppo"})
with pytest.raises(ValueError, match="missing required field: n_gpus_per_node"):
parse_taskspec({"kind": "advanced", "nnodes": 1, "command": "python3 -m verl.trainer.main_ppo"})

55
src/mvp/py/tests/test_ray_job_tool.py
View File

@@ -161,3 +161,58 @@ def test_submit_error_writes_file_then_reraises(tmp_path: Path, monkeypatch):
err = tmp_path / "jobs" / "sid2" / "logs" / "submit.error.txt"
assert err.exists()
def test_submit_advanced_writes_artifacts_and_returns_submission_id(tmp_path: Path, monkeypatch):
from argus.ray import ray_job_tool as mod
class _FakeClient:
def __init__(self, address: str):
self.address = address
self.last_submit_kwargs = None
def submit_job(self, **kwargs):
self.last_submit_kwargs = dict(kwargs)
return str(kwargs["submission_id"])
def list_jobs(self):
class X:
def dict(self):
return {"ok": True}
return [X()]
monkeypatch.setattr(mod, "JobSubmissionClient", _FakeClient)
monkeypatch.setattr(mod.ray, "init", lambda **kwargs: (_ for _ in ()).throw(RuntimeError("no ray")))
cfg = RayConfig.from_dict(
{
"address": "http://127.0.0.1:8265",
"shared_root": str(tmp_path),
"entrypoint_resources": {"worker_node": 1},
"runtime_env": {"env_vars": {}},
}
)
spec = mod.AdvancedTaskSpec.from_dict(
{
"kind": "advanced",
"submission_id": "sid3",
"nnodes": 2,
"n_gpus_per_node": 4,
"command": "echo hi",
}
)
tool = mod.RayJobTool(cfg)
submitted = tool.submit_advanced(spec, no_wait=True, user_id="alice")
assert submitted == "sid3"
job_root = tmp_path / "jobs" / "sid3"
assert (job_root / "config" / "ray_config.yaml").exists()
assert (job_root / "config" / "taskspec.yaml").exists()
assert (job_root / "config" / "ray_submission_id.txt").read_text(encoding="utf-8").strip() == "sid3"
payload = json.loads((job_root / "config" / "submit_payload.json").read_text(encoding="utf-8"))
assert payload["submission_id"] == "sid3"
assert "argus.ray.driver_entrypoint" in payload["entrypoint"]
assert (job_root / "debug" / "ray_resources_pre.error.txt").exists()

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

@@ -85,6 +85,67 @@ def test_tick_submits_one_task(monkeypatch, tmp_path: Path):
assert s.tool.job_dirs[-1] == "/private/users/alice/jobs/t1--a01"
def test_tick_submits_one_task_advanced(monkeypatch, tmp_path: Path):
from argus.service import scheduler as sched_mod
ray_cfg, v2_cfg = _mk_cfg(tmp_path)
db = Db(v2_cfg.sqlite.db_path)
db.init()
db.create_task_v25(
task_id="t1",
user_id="alice",
workload="advanced",
jobspec_yaml=yaml.safe_dump(
{
"kind": "advanced",
"nnodes": 2,
"n_gpus_per_node": 4,
"command": "python3 -m verl.trainer.main_ppo +ray_kwargs.ray_init.address=auto",
}
),
nnodes=2,
n_gpus_per_node=4,
)
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.submitted = []
self.job_dirs = []
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):
self.submitted.append(spec.submission_id)
self.job_dirs.append(job_dir)
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()
row = db.get_task("t1")
assert row and row["state"] == "SUBMITTED"
attempts = db.list_attempts("t1")
assert len(attempts) == 1
assert attempts[0]["ray_submission_id"] == "t1--a01"
assert s.tool.job_dirs[-1] == "/private/users/alice/jobs/t1--a01"
def test_tick_marks_pending_resources(monkeypatch, tmp_path: Path):
from argus.service import scheduler as sched_mod

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

@@ -78,3 +78,18 @@ def test_ui_task_detail_shows_ids(tmp_path, monkeypatch):
assert f"/ui/tasks/{task_id}/logs" in r.text
assert "TaskSpec (YAML)" in r.text
assert "/api/v2/tasks/" in r.text
def test_ui_new_task_contains_advanced_example_snippet(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
# Ensure Advanced example includes required PPO micro batch override (common failure mode if omitted).
assert "kind: advanced" in r.text
# 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