v3.0 开发验证完成，满足webui和最小数据管理需求，任务模版目前偏简单，可以扩展一下，再release开放使用

2025-12-30 17:29:50 +08:00
parent 7b89d60b3b
commit 6d3fefc7a6
30 changed files with 3764 additions and 16 deletions
--- a/src/mvp/README.md
+++ b/src/mvp/README.md
@@ -16,3 +16,11 @@
 - CLI 提交流程：`scripts/run_all_cli.sh`
 - API 提交流程：`scripts/run_all_api.sh`
 - v2.5（Stateless worker + user 隔离 jobs）E2E：`scripts/run_all_v25_api.sh`
+- v3.0（WebUI + SFTPGo + user datasets/models + jobs retention）E2E：`scripts/run_all_v30_api.sh`
+
+v3.0 访问入口（dev/h1）：
+- WebUI：`http://127.0.0.1:8080/ui`
+- Ray Dashboard：`http://127.0.0.1:8265`
+- SFTPGo：
+  - SFTP：`127.0.0.1:2022`
+  - Admin API/UI：`http://127.0.0.1:8081`（容器内 8080，host 映射到 8081 避免与 API server 冲突）
--- a/src/mvp/configs/dev.yaml
+++ b/src/mvp/configs/dev.yaml
@@ -32,3 +32,24 @@ service:
    retry_interval_s: 60
    max_running_tasks: 1

+# v3.0: user data management (filesystem + SFTPGo)
+data:
+  # All user writable data is placed under this root:
+  #   /private/users/<user_id>/{datasets,models,code,jobs,trash/jobs}
+  user_root: "/private/users"
+
+  # SFTPGo is optional in dev; when enabled, admin endpoints will call SFTPGo admin API.
+  # Admin password is provided by env var `data.sftpgo.admin_password_env`.
+  sftpgo:
+    enabled: false
+    host: ""              # shown to users via GET /api/v2/me
+    sftp_port: 2022
+    admin_api_base: ""    # e.g. http://argus-sftpgo:8080
+    admin_user: "admin"
+    admin_password_env: "SFTPGO_ADMIN_PASSWORD"
+
+  # Jobs retention policy (v3.0 janitor): move to trash after 3d, purge after 7d.
+  retention:
+    jobs_trash_after_days: 3
+    jobs_purge_after_days: 7
+    janitor_interval_s: 3600
--- a/src/mvp/configs/dev_v30.yaml
+++ b/src/mvp/configs/dev_v30.yaml
@@ -0,0 +1,51 @@
+ray:
+  # Ray Job server address (head 容器内视角)
+  address: "http://127.0.0.1:8265"
+
+  # 共享根路径（容器内统一 /private，对齐生产）
+  shared_root: "/private"
+
+  # 强制 driver 落 worker（head 不跑训练）
+  entrypoint_num_cpus: 1
+  entrypoint_resources:
+    worker_node: 1
+
+  # 所有 job 通用 runtime env
+  runtime_env:
+    env_vars:
+      HF_ENDPOINT: "https://hf-mirror.com"
+      PYTHONUNBUFFERED: "1"
+
+  # v3.0 先不支持 user code 执行
+  user_code_path: "/private/user/code"
+
+service:
+  api:
+    host: "0.0.0.0"
+    port: 8080
+  auth:
+    token_env: "MVP_INTERNAL_TOKEN"
+  sqlite:
+    db_path: "/private/common/db/mvp.sqlite3"
+  scheduler:
+    tick_s: 5
+    retry_interval_s: 60
+    max_running_tasks: 1
+
+data:
+  user_root: "/private/users"
+  sftpgo:
+    enabled: true
+    # Returned to users by GET /api/v2/me. For h1 E2E, usually connect to the host IP.
+    host: "127.0.0.1"
+    sftp_port: 2022
+    # Admin API base should include /api/v2 (SFTPGo OpenAPI server base).
+    # From head container, access SFTPGo by service name on the compose network.
+    admin_api_base: "http://argus-sftpgo:8080/api/v2"
+    admin_user: "admin"
+    admin_password_env: "SFTPGO_ADMIN_PASSWORD"
+  retention:
+    jobs_trash_after_days: 3
+    jobs_purge_after_days: 7
+    janitor_interval_s: 3600
+
--- a/src/mvp/docker-compose.yaml
+++ b/src/mvp/docker-compose.yaml
@@ -38,6 +38,36 @@ services:
      HF_ENDPOINT: "https://hf-mirror.com"
      PYTHONUNBUFFERED: "1"

+  # v3.0: Data management service (SFTPGo).
+  # - SFTP: 2022
+  # - Admin API/UI: 8080 (mapped to host 8081 to avoid collision with MVP API server on 8080)
+  #
+  # NOTE: This is for dev / h1 E2E. In prod you may use an internal mirror image/tag and different ports.
+  sftpgo:
+    image: drakkan/sftpgo:latest
+    container_name: argus-sftpgo
+    user: "0:0"
+    ports:
+      - "2022:2022"
+      - "8081:8080"
+    volumes:
+      - ../../shared:/private
+      - ../../shared/common/sftpgo:/var/lib/sftpgo
+    networks:
+      - argus-ray-net
+    environment:
+      # Create a default admin on first start (used by API server to manage users).
+      # Override on host as needed:
+      #   export SFTPGO_ADMIN_PASSWORD=...
+      SFTPGO_DATA_PROVIDER__CREATE_DEFAULT_ADMIN: "true"
+      # Persist the sqlite DB under the mounted metadata dir; otherwise it defaults to a relative path.
+      SFTPGO_DATA_PROVIDER__NAME: "/var/lib/sftpgo/sftpgo.db"
+      SFTPGO_DEFAULT_ADMIN_USERNAME: "admin"
+      SFTPGO_DEFAULT_ADMIN_PASSWORD: "${SFTPGO_ADMIN_PASSWORD:-my-dev-sftpgo-admin}"
+      # Explicitly pin default ports via env-var schema (double-underscore for nesting).
+      SFTPGO_HTTPD__BINDINGS__0__PORT: "8080"
+      SFTPGO_SFTPD__BINDINGS__0__PORT: "2022"
+
  ray_worker_0:
    image: argus/argus-ray-node:v2.5
    container_name: argus-ray-worker-0
--- a/src/mvp/py/argus/service/app.py
+++ b/src/mvp/py/argus/service/app.py
@@ -1,6 +1,7 @@
 from __future__ import annotations

 import os
+import secrets
 import threading
 from typing import Any

@@ -12,7 +13,10 @@ from argus.ray.models import JobSpec, RayConfig

 from .config import V2Config
 from .db import Db
+from .janitor import JobsJanitor
 from .scheduler import Scheduler
+from .sftpgo import SFTPGoAdminClient, SFTPGoError
+from .ui import register_ui_routes


 def _utc_now_iso() -> str:
@@ -38,11 +42,48 @@ def create_app(config_path: str) -> FastAPI:
    db.init()

    scheduler = Scheduler(db=db, ray_cfg=ray_cfg, v2_cfg=v2_cfg)
+    janitor = JobsJanitor(
+        db=db,
+        user_root=v2_cfg.data.user_root,
+        trash_after_days=v2_cfg.data.retention.jobs_trash_after_days,
+        purge_after_days=v2_cfg.data.retention.jobs_purge_after_days,
+        interval_s=v2_cfg.data.retention.janitor_interval_s,
+    )
    stop_flag = threading.Event()
    tool = scheduler.tool

    app = FastAPI(title="mvp-v2", version="2.0")

+    def _user_home(user_id: str) -> str:
+        base = v2_cfg.data.user_root.rstrip("/")
+        return f"{base}/{user_id}"
+
+    def _ensure_user_dirs(user_id: str) -> None:
+        home = _user_home(user_id)
+        # Note: create a physical /common directory under each user's home so that
+        # SFTPGo WebClient shows a "common" entry at the root. The actual shared
+        # content is exposed via SFTPGo virtual folders under /common/{datasets,models}.
+        for rel in ("datasets", "models", "code", "jobs", "trash/jobs", "common"):
+            os.makedirs(f"{home}/{rel}", exist_ok=True)
+
+    def _sftpgo_enabled() -> bool:
+        return bool(v2_cfg.data.sftpgo.enabled)
+
+    def _sftpgo_client() -> SFTPGoAdminClient:
+        cfg = v2_cfg.data.sftpgo
+        if not cfg.admin_api_base:
+            raise HTTPException(status_code=500, detail="sftpgo enabled but data.sftpgo.admin_api_base is empty")
+        pw = os.environ.get(cfg.admin_password_env, "")
+        if not pw:
+            raise HTTPException(status_code=500, detail=f"missing env: {cfg.admin_password_env}")
+        shared_root = ray_cfg.shared_root.rstrip("/")
+        return SFTPGoAdminClient(
+            admin_api_base=str(cfg.admin_api_base),
+            admin_user=str(cfg.admin_user),
+            admin_password=pw,
+            common_root=f"{shared_root}/common",
+        )
+
    def _auth(req: Request) -> dict[str, Any]:
        token_env = v2_cfg.auth.token_env
        admin_token = os.environ.get(token_env, "")
@@ -74,6 +115,9 @@ def create_app(config_path: str) -> FastAPI:
    def _startup() -> None:
        t = threading.Thread(target=scheduler.run_forever, args=(stop_flag,), daemon=True)
        t.start()
+        if int(janitor.interval_s) > 0:
+            tj = threading.Thread(target=janitor.run_forever, args=(stop_flag,), daemon=True)
+            tj.start()

    @app.on_event("shutdown")
    def _shutdown() -> None:
@@ -93,7 +137,42 @@ def create_app(config_path: str) -> FastAPI:
            row = db.create_user(user_id=user_id, display_name=str(display_name) if display_name is not None else None)
        except Exception as e:
            raise HTTPException(status_code=409, detail=f"user create failed: {e!r}")
-        return {"user_id": row.get("user_id", user_id), "state": row.get("state", "ACTIVE")}
+
+        # v3.0: create user dir structure (datasets/models/code/jobs/trash).
+        try:
+            _ensure_user_dirs(user_id)
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"failed to create user dirs: {e!r}")
+
+        out: dict[str, Any] = {"user_id": row.get("user_id", user_id), "state": row.get("state", "ACTIVE")}
+
+        # v3.0: scheme A (password) SFTPGo integration.
+        if _sftpgo_enabled():
+            pw = secrets.token_urlsafe(12)
+            try:
+                _sftpgo_client().create_user(username=user_id, password=pw, home_dir=_user_home(user_id))
+            except SFTPGoError as e:
+                # Make create user idempotent for retries:
+                # If the user already exists in SFTPGo, reset password and enable it.
+                if "http error: 409" in str(e):
+                    try:
+                        _sftpgo_client().reset_password(username=user_id, new_password=pw, home_dir=_user_home(user_id))
+                        _sftpgo_client().enable_user(username=user_id, home_dir=_user_home(user_id))
+                    except Exception as e2:
+                        raise HTTPException(status_code=502, detail=f"sftpgo upsert user failed: {e2!r}")
+                else:
+                    raise HTTPException(status_code=502, detail=f"sftpgo create user failed: {e!r}")
+            except HTTPException:
+                raise
+            except Exception as e:
+                raise HTTPException(status_code=502, detail=f"sftpgo create user failed: {e!r}")
+            out["sftp"] = {
+                "username": user_id,
+                "password": pw,  # one-time return to admin; do not persist plaintext
+                "host": v2_cfg.data.sftpgo.host,
+                "port": int(v2_cfg.data.sftpgo.sftp_port),
+            }
+        return out

    @app.post("/api/v2/users/{user_id}/tokens")
    async def issue_token(user_id: str, req: Request) -> dict[str, Any]:
@@ -104,6 +183,12 @@ def create_app(config_path: str) -> FastAPI:
        token = db.issue_token(user_id=user_id)
        return {"user_id": user_id, "token": token}

+    @app.get("/api/v2/users")
+    async def list_users(req: Request, limit: int = 200) -> dict[str, Any]:
+        _require_admin(req)
+        lim = max(1, min(int(limit), 1000))
+        return {"users": db.list_users(limit=lim), "limit": lim}
+
    @app.post("/api/v2/users/{user_id}:disable")
    async def disable_user(user_id: str, req: Request) -> dict[str, Any]:
        _require_admin(req)
@@ -111,8 +196,122 @@ def create_app(config_path: str) -> FastAPI:
        if not u:
            raise HTTPException(status_code=404, detail="user not found")
        db.disable_user(user_id=user_id)
+        if _sftpgo_enabled():
+            try:
+                _sftpgo_client().disable_user(username=user_id, home_dir=_user_home(user_id))
+            except Exception:
+                # Best-effort: DB state is source of truth for API auth; SFTPGo sync can lag.
+                pass
        return {"user_id": user_id, "state": "DISABLED"}

+    @app.post("/api/v2/users/{user_id}/sftp:reset_password")
+    async def reset_sftp_password(user_id: str, req: Request) -> dict[str, Any]:
+        _require_admin(req)
+        u = db.get_user(user_id)
+        if not u:
+            raise HTTPException(status_code=404, detail="user not found")
+        if not _sftpgo_enabled():
+            raise HTTPException(status_code=400, detail="sftpgo is not enabled")
+        pw = secrets.token_urlsafe(12)
+        try:
+            _sftpgo_client().reset_password(username=user_id, new_password=pw, home_dir=_user_home(user_id))
+        except HTTPException:
+            raise
+        except Exception as e:
+            raise HTTPException(status_code=502, detail=f"sftpgo reset password failed: {e!r}")
+        return {"user_id": user_id, "password": pw}
+
+    @app.post("/api/v2/me/sftp:reset_password")
+    async def reset_my_sftp_password(req: Request) -> dict[str, Any]:
+        """
+        v3.0 WebUI: allow a user to rotate their own SFTPGo password and copy it.
+        Note: SFTPGo does not allow reading the current password, so this endpoint returns a new one-time password.
+        """
+        subject = _auth(req)
+        user_id = str(subject["user_id"])
+        u = db.get_user(user_id)
+        if not u:
+            raise HTTPException(status_code=404, detail="user not found")
+        if not _sftpgo_enabled():
+            raise HTTPException(status_code=400, detail="sftpgo is not enabled")
+        pw = secrets.token_urlsafe(12)
+        try:
+            _sftpgo_client().reset_password(username=user_id, new_password=pw, home_dir=_user_home(user_id))
+            _sftpgo_client().enable_user(username=user_id, home_dir=_user_home(user_id))
+        except Exception as e:
+            raise HTTPException(status_code=502, detail=f"sftpgo reset password failed: {e!r}")
+        return {"user_id": user_id, "password": pw}
+
+    @app.get("/api/v2/me")
+    async def me(req: Request) -> dict[str, Any]:
+        subject = _auth(req)
+        user_id = str(subject["user_id"])
+        try:
+            _ensure_user_dirs(user_id)
+        except Exception:
+            # Best-effort: user may still be able to use API even if FS init fails.
+            pass
+        # Best-effort: reconcile SFTPGo user to include /common read-only mounts.
+        # This makes the virtual folders visible without requiring a password reset.
+        if _sftpgo_enabled() and not subject.get("is_admin"):
+            try:
+                _sftpgo_client().enable_user(username=user_id, home_dir=_user_home(user_id))
+            except Exception:
+                pass
+        home = _user_home(user_id)
+        out: dict[str, Any] = {
+            "user_id": user_id,
+            "is_admin": bool(subject.get("is_admin")),
+            "paths": {
+                "home": home,
+                "datasets": f"{home}/datasets",
+                "models": f"{home}/models",
+                "code": f"{home}/code",
+                "jobs": f"{home}/jobs",
+                "trash_jobs": f"{home}/trash/jobs",
+            },
+            "retention": {
+                "jobs_trash_after_days": int(v2_cfg.data.retention.jobs_trash_after_days),
+                "jobs_purge_after_days": int(v2_cfg.data.retention.jobs_purge_after_days),
+            },
+        }
+        if _sftpgo_enabled():
+            out["sftp"] = {
+                "host": v2_cfg.data.sftpgo.host,
+                "port": int(v2_cfg.data.sftpgo.sftp_port),
+                "username": user_id,
+            }
+        return out
+
+    @app.get("/api/v2/tasks")
+    async def list_tasks(req: Request, limit: int = 200, offset: int = 0, states: str | None = None) -> dict[str, Any]:
+        subject = _auth(req)
+        lim = max(1, min(int(limit), 1000))
+        off = max(0, int(offset))
+        state_list: list[str] | None = None
+        if states:
+            raw = [s.strip() for s in str(states).split(",") if s.strip()]
+            # Keep it strict to avoid typos silently returning empty results.
+            allowed = {
+                "QUEUED",
+                "PENDING_RESOURCES",
+                "SUBMITTING",
+                "SUBMITTED",
+                "RUNNING",
+                "SUCCEEDED",
+                "FAILED",
+                "CANCELED",
+            }
+            for s in raw:
+                if s not in allowed:
+                    raise HTTPException(status_code=400, detail=f"invalid state: {s}")
+            state_list = raw or None
+        if subject.get("is_admin"):
+            tasks = db.list_tasks(user_id=None, states=state_list, limit=lim, offset=off)
+        else:
+            tasks = db.list_tasks(user_id=str(subject["user_id"]), states=state_list, limit=lim, offset=off)
+        return {"tasks": tasks, "limit": lim, "offset": off, "has_more": bool(len(tasks) == lim)}
+
    @app.post("/api/v2/tasks")
    async def submit_task(req: Request) -> dict[str, Any]:
        subject = _auth(req)
@@ -126,13 +325,40 @@ def create_app(config_path: str) -> FastAPI:
        except Exception as e:
            raise HTTPException(status_code=400, detail=f"invalid jobspec: {e!r}")

-        # v2.5 constraint: training inputs must come from /private/common (dev/prod统一)。
-        common_prefix = ray_cfg.shared_root.rstrip("/") + "/common/"
-        for k, v in (("code_path", spec.code_path), ("train_file", spec.train_file), ("val_file", spec.val_file)):
+        # v3.0 path policy:
+        # - 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()}/"
+
+        common_datasets_prefix = f"{common_prefix}datasets/"
+        user_datasets_prefix = f"{user_prefix}datasets/"
+        common_models_prefix = f"{common_prefix}models/"
+        user_models_prefix = f"{user_prefix}models/"
+
+        if not str(spec.code_path).startswith(common_prefix):
+            raise HTTPException(status_code=400, detail=f"code_path must start with {common_prefix}")
+
+        for k, v in (("train_file", spec.train_file), ("val_file", spec.val_file)):
            if v is None:
                continue
-            if not str(v).startswith(common_prefix):
-                raise HTTPException(status_code=400, detail=f"{k} must start with {common_prefix}")
+            sv = str(v)
+            if not (sv.startswith(common_datasets_prefix) or sv.startswith(user_datasets_prefix)):
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"{k} must start with {common_datasets_prefix} or {user_datasets_prefix}",
+                )
+
+        model_id = str(spec.model_id)
+        if model_id.startswith(f"{root}/"):
+            if not (model_id.startswith(common_models_prefix) or model_id.startswith(user_models_prefix)):
+                raise HTTPException(
+                    status_code=400,
+                    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(
@@ -257,4 +483,7 @@ def create_app(config_path: str) -> FastAPI:
            return db.list_queue()
        return db.list_queue(user_id=str(subject["user_id"]))

+    # v3.0: minimal WebUI (no server-side session; token stored in browser localStorage).
+    register_ui_routes(app)
+
    return app
--- a/src/mvp/py/argus/service/config.py
+++ b/src/mvp/py/argus/service/config.py
@@ -27,12 +27,37 @@ class V2SchedulerConfig:
    max_running_tasks: int = 1


+@dataclass(frozen=True)
+class V2RetentionConfig:
+    jobs_trash_after_days: int = 3
+    jobs_purge_after_days: int = 7
+    janitor_interval_s: int = 3600
+
+
+@dataclass(frozen=True)
+class V2SFTPGoConfig:
+    enabled: bool = False
+    host: str = ""
+    sftp_port: int = 2022
+    admin_api_base: str = ""
+    admin_user: str = "admin"
+    admin_password_env: str = "SFTPGO_ADMIN_PASSWORD"
+
+
+@dataclass(frozen=True)
+class V2DataConfig:
+    user_root: str
+    sftpgo: V2SFTPGoConfig
+    retention: V2RetentionConfig
+
+
@dataclass(frozen=True)
 class V2Config:
    api: V2ApiConfig
    auth: V2AuthConfig
    sqlite: V2SqliteConfig
    scheduler: V2SchedulerConfig
+    data: V2DataConfig

    @staticmethod
    def from_root_dict(root: dict[str, Any]) -> "V2Config":
@@ -58,9 +83,19 @@ class V2Config:
        else:
            shared_root = str(root.get("shared_root") or "/private")

+        data = root.get("data") or {}
+        if not isinstance(data, dict):
+            raise ValueError("config.data must be a mapping")
+        sftpgo = data.get("sftpgo") or {}
+        retention = data.get("retention") or {}
+        if not isinstance(sftpgo, dict) or not isinstance(retention, dict):
+            raise ValueError("config.data.{sftpgo,retention} must be mappings")
+
        default_db_path = f"{shared_root}/common/db/mvp.sqlite3"
        db_path = str(sqlite.get("db_path") or default_db_path)

+        user_root = str(data.get("user_root") or f"{shared_root}/users")
+
        return V2Config(
            api=V2ApiConfig(
                host=str(api.get("host") or "0.0.0.0"),
@@ -73,4 +108,20 @@ class V2Config:
                retry_interval_s=int(scheduler.get("retry_interval_s") or 60),
                max_running_tasks=int(scheduler.get("max_running_tasks") or 1),
            ),
+            data=V2DataConfig(
+                user_root=user_root,
+                sftpgo=V2SFTPGoConfig(
+                    enabled=bool(sftpgo.get("enabled") or False),
+                    host=str(sftpgo.get("host") or ""),
+                    sftp_port=int(sftpgo.get("sftp_port") or 2022),
+                    admin_api_base=str(sftpgo.get("admin_api_base") or ""),
+                    admin_user=str(sftpgo.get("admin_user") or "admin"),
+                    admin_password_env=str(sftpgo.get("admin_password_env") or "SFTPGO_ADMIN_PASSWORD"),
+                ),
+                retention=V2RetentionConfig(
+                    jobs_trash_after_days=int(retention.get("jobs_trash_after_days") or 3),
+                    jobs_purge_after_days=int(retention.get("jobs_purge_after_days") or 7),
+                    janitor_interval_s=int(retention.get("janitor_interval_s") or 3600),
+                ),
+            ),
        )
--- a/src/mvp/py/argus/service/db.py
+++ b/src/mvp/py/argus/service/db.py
@@ -40,7 +40,8 @@ class Db:
                  user_id TEXT PRIMARY KEY,
                  display_name TEXT,
                  state TEXT NOT NULL,
-                  created_at TEXT NOT NULL
+                  created_at TEXT NOT NULL,
+                  updated_at TEXT NOT NULL
                )
                """
            )
@@ -49,6 +50,7 @@ class Db:
                CREATE TABLE IF NOT EXISTS api_tokens (
                  token_hash TEXT PRIMARY KEY,
                  user_id TEXT NOT NULL,
+                  token_plain TEXT NOT NULL,
                  created_at TEXT NOT NULL,
                  last_used_at TEXT,
                  FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
@@ -78,6 +80,15 @@ class Db:
                conn.execute("ALTER TABLE tasks ADD COLUMN user_id TEXT")
            except sqlite3.OperationalError:
                pass
+            # Best-effort: add missing columns (forward compatibility).
+            try:
+                conn.execute("ALTER TABLE users ADD COLUMN updated_at TEXT")
+            except sqlite3.OperationalError:
+                pass
+            try:
+                conn.execute("ALTER TABLE api_tokens ADD COLUMN token_plain TEXT")
+            except sqlite3.OperationalError:
+                pass
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS attempts (
@@ -168,10 +179,10 @@ class Db:
        with self.tx() as conn:
            conn.execute(
                """
-                INSERT INTO users (user_id, display_name, state, created_at)
-                VALUES (?, ?, 'ACTIVE', ?)
+                INSERT INTO users (user_id, display_name, state, created_at, updated_at)
+                VALUES (?, ?, 'ACTIVE', ?, ?)
                """,
-                (user_id, display_name, now),
+                (user_id, display_name, now, now),
            )
            conn.execute(
                "INSERT INTO events (task_id, ts, event_type, payload_json) VALUES (NULL, ?, 'USER_CREATED', ?)",
@@ -183,7 +194,7 @@ class Db:
    def disable_user(self, *, user_id: str) -> None:
        now = _utc_now_iso()
        with self.tx() as conn:
-            conn.execute("UPDATE users SET state = 'DISABLED' WHERE user_id = ?", (user_id,))
+            conn.execute("UPDATE users SET state = 'DISABLED', updated_at = ? WHERE user_id = ?", (now, user_id))
            conn.execute(
                "INSERT INTO events (task_id, ts, event_type, payload_json) VALUES (NULL, ?, 'USER_DISABLED', ?)",
                (now, user_id),
@@ -195,15 +206,18 @@ class Db:
            return dict(row) if row else None

    def issue_token(self, *, user_id: str) -> str:
-        # Returns plaintext token once; stores hash only.
+        # Returns plaintext token once.
+        # Note: For v3.0 WebUI admin convenience we persist the plaintext token so admins
+        # can re-copy it later. This is acceptable only for internal/dev deployments.
        now = _utc_now_iso()
        token = f"mvp_u_{user_id}_{secrets.token_urlsafe(18)}"
        token_hash = self._hash_token(token)
        with self.tx() as conn:
            conn.execute(
-                "INSERT INTO api_tokens (token_hash, user_id, created_at) VALUES (?, ?, ?)",
-                (token_hash, user_id, now),
+                "INSERT INTO api_tokens (token_hash, user_id, token_plain, created_at) VALUES (?, ?, ?, ?)",
+                (token_hash, user_id, token, now),
            )
+            conn.execute("UPDATE users SET updated_at = ? WHERE user_id = ?", (now, user_id))
            conn.execute(
                "INSERT INTO events (task_id, ts, event_type, payload_json) VALUES (NULL, ?, 'TOKEN_ISSUED', ?)",
                (now, user_id),
@@ -226,8 +240,48 @@ class Db:
                return None
            now = _utc_now_iso()
            conn.execute("UPDATE api_tokens SET last_used_at = ? WHERE token_hash = ?", (now, token_hash))
+            conn.execute("UPDATE users SET updated_at = ? WHERE user_id = ?", (now, str(row["user_id"])))
            return {"user_id": row["user_id"], "state": row["state"]}

+    def list_users(self, limit: int = 200) -> list[dict[str, Any]]:
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT
+                  u.user_id,
+                  u.display_name,
+                  u.state,
+                  u.created_at,
+                  u.updated_at,
+                  (
+                    SELECT t.token_plain
+                    FROM api_tokens t
+                    WHERE t.user_id = u.user_id
+                    ORDER BY t.created_at DESC
+                    LIMIT 1
+                  ) AS token,
+                  (
+                    SELECT t.created_at
+                    FROM api_tokens t
+                    WHERE t.user_id = u.user_id
+                    ORDER BY t.created_at DESC
+                    LIMIT 1
+                  ) AS token_created_at,
+                  (
+                    SELECT t.last_used_at
+                    FROM api_tokens t
+                    WHERE t.user_id = u.user_id
+                    ORDER BY t.created_at DESC
+                    LIMIT 1
+                  ) AS token_last_used_at
+                FROM users u
+                ORDER BY u.created_at DESC
+                LIMIT ?
+                """,
+                (int(limit),),
+            ).fetchall()
+            return [dict(r) for r in rows]
+
    def get_task(self, task_id: str) -> dict[str, Any] | None:
        with self._connect() as conn:
            row = conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone()
@@ -269,6 +323,40 @@ class Db:
            running = conn.execute(running_sql, tuple(params)).fetchall()
            return {"pending": [dict(r) for r in pending], "running": [dict(r) for r in running]}

+    def list_tasks(
+        self,
+        *,
+        user_id: str | None = None,
+        states: list[str] | None = None,
+        limit: int = 200,
+        offset: int = 0,
+    ) -> list[dict[str, Any]]:
+        """
+        Returns recent tasks (including terminal tasks).
+        """
+        with self._connect() as conn:
+            params: list[Any] = []
+            where_clauses: list[str] = []
+            if user_id is not None:
+                where_clauses.append("user_id = ?")
+                params.append(user_id)
+            if states:
+                placeholders = ",".join(["?"] * len(states))
+                where_clauses.append(f"state IN ({placeholders})")
+                params.extend(states)
+            where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
+            sql = (
+                "SELECT task_id, user_id, workload, state, nnodes, n_gpus_per_node, latest_attempt_no, created_at, updated_at, error_summary "
+                "FROM tasks"
+                f"{where_sql} "
+                "ORDER BY created_at DESC "
+                "LIMIT ? OFFSET ?"
+            )
+            params.append(int(limit))
+            params.append(max(0, int(offset)))
+            rows = conn.execute(sql, tuple(params)).fetchall()
+            return [dict(r) for r in rows]
+
    def count_running(self) -> int:
        with self._connect() as conn:
            row = conn.execute(
@@ -383,3 +471,25 @@ class Db:
                "INSERT INTO events (task_id, ts, event_type, payload_json) VALUES (?, ?, 'ATTEMPT_UPDATE', ?)",
                (task_id, now, None),
            )
+
+    def list_ended_attempts_before(self, *, end_time_le: str, limit: int = 2000) -> list[dict[str, Any]]:
+        """
+        Used by the jobs janitor:
+        - Only considers tasks with a non-null user_id (v2.5+).
+        - Returns attempts that have end_time <= end_time_le.
+        """
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT t.task_id, t.user_id, a.ray_submission_id, a.end_time
+                FROM attempts a
+                JOIN tasks t ON t.task_id = a.task_id
+                WHERE t.user_id IS NOT NULL
+                  AND a.end_time IS NOT NULL
+                  AND a.end_time <= ?
+                ORDER BY a.end_time ASC
+                LIMIT ?
+                """,
+                (str(end_time_le), int(limit)),
+            ).fetchall()
+            return [dict(r) for r in rows]
--- a/src/mvp/py/argus/service/janitor.py
+++ b/src/mvp/py/argus/service/janitor.py
@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import os
+import shutil
+import time
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from .db import Db
+
+
+def _parse_iso_z(ts: str) -> datetime:
+    # Stored as "YYYY-MM-DDTHH:MM:SSZ" (naive UTC). Parse into aware UTC.
+    return datetime.fromisoformat(ts.replace("Z", "+00:00")).astimezone(timezone.utc)
+
+
+def _iso_z(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+@dataclass
+class JobsJanitor:
+    db: Db
+    user_root: str
+    trash_after_days: int = 3
+    purge_after_days: int = 7
+    interval_s: int = 3600
+
+    def __post_init__(self) -> None:
+        if int(self.trash_after_days) < 0 or int(self.purge_after_days) < 0:
+            raise ValueError("retention days must be non-negative")
+        if int(self.purge_after_days) and int(self.purge_after_days) < int(self.trash_after_days):
+            raise ValueError("purge_after_days must be >= trash_after_days")
+
+    def _job_dir(self, *, user_id: str, ray_submission_id: str) -> str:
+        base = self.user_root.rstrip("/")
+        return f"{base}/{user_id}/jobs/{ray_submission_id}"
+
+    def _trash_dir(self, *, user_id: str, ray_submission_id: str) -> str:
+        base = self.user_root.rstrip("/")
+        return f"{base}/{user_id}/trash/jobs/{ray_submission_id}"
+
+    def tick_once(self, *, now: datetime | None = None, limit: int = 2000) -> None:
+        if int(self.trash_after_days) <= 0 and int(self.purge_after_days) <= 0:
+            return
+
+        now_dt = now or datetime.now(timezone.utc)
+        move_cutoff = now_dt - timedelta(days=int(self.trash_after_days))
+        move_cutoff_iso = _iso_z(move_cutoff)
+
+        rows = self.db.list_ended_attempts_before(end_time_le=move_cutoff_iso, limit=int(limit))
+        for r in rows:
+            user_id = str(r.get("user_id") or "").strip()
+            sid = str(r.get("ray_submission_id") or "").strip()
+            end_time = str(r.get("end_time") or "").strip()
+            if not user_id or not sid or not end_time:
+                continue
+
+            try:
+                ended_at = _parse_iso_z(end_time)
+            except Exception:
+                continue
+
+            age_days = (now_dt - ended_at).total_seconds() / 86400.0
+            src = self._job_dir(user_id=user_id, ray_submission_id=sid)
+            dst = self._trash_dir(user_id=user_id, ray_submission_id=sid)
+            dst_parent = os.path.dirname(dst)
+
+            # Between trash and purge: ensure in trash.
+            if age_days >= float(self.trash_after_days) and age_days < float(self.purge_after_days or 10**9):
+                if os.path.exists(src) and not os.path.exists(dst):
+                    os.makedirs(dst_parent, exist_ok=True)
+                    try:
+                        shutil.move(src, dst)
+                    except Exception:
+                        pass
+                continue
+
+            # Purge: move to trash (if still in jobs) then delete from trash.
+            if int(self.purge_after_days) > 0 and age_days >= float(self.purge_after_days):
+                if os.path.exists(src) and not os.path.exists(dst):
+                    os.makedirs(dst_parent, exist_ok=True)
+                    try:
+                        shutil.move(src, dst)
+                    except Exception:
+                        pass
+                if os.path.exists(dst):
+                    try:
+                        shutil.rmtree(dst)
+                    except Exception:
+                        pass
+
+    def run_forever(self, stop_flag: object) -> None:
+        try:
+            is_set = stop_flag.is_set  # type: ignore[attr-defined]
+        except Exception:
+            raise ValueError("stop_flag must be a threading.Event-like object with is_set()")
+
+        while not is_set():
+            try:
+                self.tick_once()
+            except Exception:
+                pass
+            time.sleep(max(1, int(self.interval_s)))
+
--- a/src/mvp/py/argus/service/sftpgo.py
+++ b/src/mvp/py/argus/service/sftpgo.py
@@ -0,0 +1,221 @@
+from __future__ import annotations
+
+import base64
+import json
+from dataclasses import dataclass
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+
+class SFTPGoError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class SFTPGoAdminClient:
+    """
+    Minimal SFTPGo admin API client (v3.0).
+
+    SFTPGo OpenAPI documents the admin token flow:
+      - GET /api/v2/token with HTTP BasicAuth -> returns {"access_token": "..."}.
+      - Use Authorization: Bearer <token> for admin endpoints like POST /api/v2/users.
+
+    See upstream OpenAPI for details:
+      https://raw.githubusercontent.com/drakkan/sftpgo/main/openapi/openapi.yaml
+    """
+
+    admin_api_base: str
+    admin_user: str
+    admin_password: str
+    common_root: str = "/private/common"
+
+    def _url(self, path: str) -> str:
+        base = self.admin_api_base.rstrip("/")
+        p = path if path.startswith("/") else f"/{path}"
+        return f"{base}{p}"
+
+    def _basic_auth_header(self) -> str:
+        raw = f"{self.admin_user}:{self.admin_password}".encode("utf-8")
+        return "Basic " + base64.b64encode(raw).decode("ascii")
+
+    def _get_json(self, url: str, headers: dict[str, str]) -> dict:
+        req = Request(url, headers=headers, method="GET")
+        try:
+            with urlopen(req, timeout=10) as resp:
+                return json.loads(resp.read().decode("utf-8"))
+        except HTTPError as e:
+            raise SFTPGoError(f"sftpgo http error: {e.code} {e.reason}") from e
+        except URLError as e:
+            raise SFTPGoError(f"sftpgo connection error: {e!r}") from e
+
+    def _post_json(self, url: str, payload: dict, headers: dict[str, str]) -> None:
+        data = json.dumps(payload).encode("utf-8")
+        req = Request(url, data=data, headers=headers, method="POST")
+        try:
+            with urlopen(req, timeout=10) as resp:
+                _ = resp.read()
+        except HTTPError as e:
+            raise SFTPGoError(f"sftpgo http error: {e.code} {e.reason}") from e
+        except URLError as e:
+            raise SFTPGoError(f"sftpgo connection error: {e!r}") from e
+
+    def _put_json(self, url: str, payload: dict, headers: dict[str, str]) -> None:
+        data = json.dumps(payload).encode("utf-8")
+        req = Request(url, data=data, headers=headers, method="PUT")
+        try:
+            with urlopen(req, timeout=10) as resp:
+                _ = resp.read()
+        except HTTPError as e:
+            raise SFTPGoError(f"sftpgo http error: {e.code} {e.reason}") from e
+        except URLError as e:
+            raise SFTPGoError(f"sftpgo connection error: {e!r}") from e
+
+    def _admin_token(self) -> str:
+        url = self._url("/token")
+        obj = self._get_json(url, headers={"Authorization": self._basic_auth_header()})
+        tok = str(obj.get("access_token") or "").strip()
+        if not tok:
+            raise SFTPGoError("sftpgo token response missing access_token")
+        return tok
+
+    def _auth_headers(self, tok: str) -> dict[str, str]:
+        return {"Authorization": f"Bearer {tok}", "Content-Type": "application/json"}
+
+    def _ensure_folder(self, *, tok: str, name: str, mapped_path: str) -> None:
+        url = self._url("/folders")
+        try:
+            self._post_json(url, {"name": name, "mapped_path": mapped_path}, headers=self._auth_headers(tok))
+        except SFTPGoError as e:
+            # Idempotent + self-healing:
+            # If it already exists, update mapped_path to the desired value.
+            if "409" in str(e):
+                self._put_json(
+                    self._url(f"/folders/{name}"),
+                    {"name": name, "mapped_path": mapped_path},
+                    headers=self._auth_headers(tok),
+                )
+                return
+            raise
+
+    def _ensure_common_folders(self, *, tok: str) -> None:
+        # Important: do NOT map datasets to /private/common/datasets because that path is
+        # a symlink farm (e.g. gsm8k -> /private/datasets/gsm8k) and SFTPGo WebClient
+        # will treat symlink traversal as escaping the virtual folder root, resulting
+        # in "permission denied". Map directly to the real datasets root.
+        common = self.common_root.rstrip("/")
+        if common.endswith("/common"):
+            shared_root = common[: -len("/common")] or "/private"
+        else:
+            shared_root = "/private"
+
+        self._ensure_folder(tok=tok, name="common_datasets", mapped_path=f"{shared_root}/datasets")
+        # Expose HF cache read-only so users can inspect downloaded models/datasets.
+        self._ensure_folder(tok=tok, name="common_hf", mapped_path=f"{shared_root}/hf")
+
+    def _get_user(self, *, tok: str, username: str) -> dict:
+        url = self._url(f"/users/{username}")
+        return self._get_json(url, headers={"Authorization": f"Bearer {tok}"})
+
+    def _put_user(self, *, tok: str, username: str, payload: dict) -> None:
+        url = self._url(f"/users/{username}")
+        self._put_json(url, payload, headers=self._auth_headers(tok))
+
+    def _apply_common_readonly(self, user_payload: dict) -> dict:
+        # Path-based permissions: make /common/* read-only while keeping home writeable.
+        perms = dict(user_payload.get("permissions") or {"/": ["*"]})
+        # Ensure /common is visible as a directory and can be traversed.
+        perms["/common"] = ["list"]
+        perms["/common/datasets"] = ["list", "download"]
+        perms["/common/hf"] = ["list", "download"]
+        user_payload["permissions"] = perms
+
+        desired_vf = [
+            {"name": "common_datasets", "virtual_path": "/common/datasets"},
+            {"name": "common_hf", "virtual_path": "/common/hf"},
+        ]
+        existing = user_payload.get("virtual_folders") or []
+        if not isinstance(existing, list):
+            existing = []
+        seen = {(vf.get("name"), vf.get("virtual_path")) for vf in existing if isinstance(vf, dict)}
+        merged = list(existing)
+        for vf in desired_vf:
+            key = (vf["name"], vf["virtual_path"])
+            if key not in seen:
+                merged.append(vf)
+        user_payload["virtual_folders"] = merged
+        return user_payload
+
+    def create_user(self, *, username: str, password: str, home_dir: str) -> None:
+        tok = self._admin_token()
+        self._ensure_common_folders(tok=tok)
+        url = self._url("/users")
+        payload: dict = {
+            "status": 1,
+            "username": username,
+            "password": password,
+            "home_dir": home_dir,
+            "permissions": {
+                "/": ["*"],
+                "/common": ["list"],
+                "/common/datasets": ["list", "download"],
+                "/common/hf": ["list", "download"],
+            },
+            "virtual_folders": [
+                {"name": "common_datasets", "virtual_path": "/common/datasets"},
+                {"name": "common_hf", "virtual_path": "/common/hf"},
+            ],
+        }
+        self._post_json(
+            url,
+            payload,
+            headers=self._auth_headers(tok),
+        )
+
+    def disable_user(self, *, username: str, home_dir: str) -> None:
+        tok = self._admin_token()
+        self._ensure_common_folders(tok=tok)
+        cur = self._get_user(tok=tok, username=username)
+        payload: dict = {
+            "username": username,
+            "status": 0,
+            "home_dir": home_dir,
+            "uid": cur.get("uid", 0),
+            "gid": cur.get("gid", 0),
+            "permissions": cur.get("permissions") or {"/": ["*"]},
+            "virtual_folders": cur.get("virtual_folders") or [],
+        }
+        self._apply_common_readonly(payload)
+        self._put_user(tok=tok, username=username, payload=payload)
+
+    def enable_user(self, *, username: str, home_dir: str) -> None:
+        tok = self._admin_token()
+        self._ensure_common_folders(tok=tok)
+        cur = self._get_user(tok=tok, username=username)
+        payload: dict = {
+            "username": username,
+            "status": 1,
+            "home_dir": home_dir,
+            "uid": cur.get("uid", 0),
+            "gid": cur.get("gid", 0),
+            "permissions": cur.get("permissions") or {"/": ["*"]},
+            "virtual_folders": cur.get("virtual_folders") or [],
+        }
+        self._apply_common_readonly(payload)
+        self._put_user(tok=tok, username=username, payload=payload)
+
+    def reset_password(self, *, username: str, new_password: str, home_dir: str) -> None:
+        tok = self._admin_token()
+        self._ensure_common_folders(tok=tok)
+        cur = self._get_user(tok=tok, username=username)
+        payload: dict = {
+            "username": username,
+            "status": 1,
+            "home_dir": home_dir,
+            "uid": cur.get("uid", 0),
+            "gid": cur.get("gid", 0),
+            "permissions": cur.get("permissions") or {"/": ["*"]},
+            "virtual_folders": cur.get("virtual_folders") or [],
+            "password": new_password,
+        }
+        self._apply_common_readonly(payload)
+        self._put_user(tok=tok, username=username, payload=payload)
--- a/src/mvp/py/argus/service/ui.py
+++ b/src/mvp/py/argus/service/ui.py
@@ -0,0 +1,629 @@
+from __future__ import annotations
+
+import html
+import json
+
+from fastapi import FastAPI
+from fastapi.responses import HTMLResponse, RedirectResponse
+
+
+_BASE_CSS = """
+:root { --bg:#0b1020; --panel:#111a33; --muted:#95a3c6; --fg:#e8eeff; --accent:#7aa2ff; --danger:#ff6b6b; --ok:#3ddc97; }
+* { box-sizing: border-box; }
+body { margin:0; font-family: ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, Helvetica, Arial; background:var(--bg); color:var(--fg); }
+a { color:var(--accent); text-decoration:none; }
+.layout { display:flex; min-height:100vh; }
+.nav { width: 240px; padding:16px; background: linear-gradient(180deg, #0e1630, #0b1020); border-right: 1px solid rgba(255,255,255,0.06); }
+.brand { font-weight: 700; letter-spacing: .2px; margin-bottom: 12px; }
+.nav a { display:block; padding:10px 10px; border-radius:10px; color: var(--fg); opacity: .9; }
+.nav a.active { background: rgba(122,162,255,0.14); border: 1px solid rgba(122,162,255,0.22); }
+.nav a:hover { background: rgba(255,255,255,0.06); }
+.main { flex:1; padding: 20px 24px; }
+.card { background: rgba(255,255,255,0.04); border: 1px solid rgba(255,255,255,0.08); border-radius: 14px; padding: 16px; }
+.row { display:flex; gap: 12px; align-items:center; flex-wrap: wrap; }
+.muted { color: var(--muted); }
+.btn { border: 1px solid rgba(255,255,255,0.16); background: rgba(255,255,255,0.06); color: var(--fg); padding: 10px 12px; border-radius: 10px; cursor: pointer; }
+.btn:hover { background: rgba(255,255,255,0.10); }
+.btn.danger { border-color: rgba(255,107,107,0.35); background: rgba(255,107,107,0.10); }
+.pill { display:inline-block; padding: 2px 10px; border-radius: 999px; border: 1px solid rgba(255,255,255,0.16); font-size: 12px; }
+.pill.ok { border-color: rgba(61,220,151,0.35); background: rgba(61,220,151,0.12); }
+.pill.bad { border-color: rgba(255,107,107,0.35); background: rgba(255,107,107,0.12); }
+textarea, input { width: 100%; color: var(--fg); background: rgba(255,255,255,0.06); border: 1px solid rgba(255,255,255,0.12); border-radius: 12px; padding: 10px 12px; outline: none; }
+button:disabled { opacity: .45; cursor: not-allowed; }
+pre { white-space: pre-wrap; word-break: break-word; }
+table { width:100%; border-collapse: collapse; }
+th, td { padding: 10px 8px; border-bottom: 1px solid rgba(255,255,255,0.08); text-align:left; }
+""".strip()
+
+
+_BASE_JS = """
+function mvpTokenGet() {
+  return (localStorage.getItem("mvp_token") || "").trim();
+}
+function mvpTokenSet(v) {
+  localStorage.setItem("mvp_token", (v || "").trim());
+}
+function mvpSftpPasswordGet() {
+  return (localStorage.getItem("mvp_sftp_password") || "").trim();
+}
+function mvpSftpPasswordSet(v) {
+  localStorage.setItem("mvp_sftp_password", (v || "").trim());
+}
+async function apiFetch(path, opts) {
+  opts = opts || {};
+  opts.headers = opts.headers || {};
+  const tok = mvpTokenGet();
+  if (tok) opts.headers["Authorization"] = "Bearer " + tok;
+  return fetch(path, opts);
+}
+async function apiJson(path, opts) {
+  const resp = await apiFetch(path, opts);
+  const text = await resp.text();
+  if (!resp.ok) {
+    const err = new Error("HTTP " + resp.status);
+    err.status = resp.status;
+    err.body = text;
+    throw err;
+  }
+  return JSON.parse(text);
+}
+function fmtJson(obj) {
+  try { return JSON.stringify(obj, null, 2); } catch (e) { return String(obj); }
+}
+function curOriginWithPort(port) {
+  const proto = window.location.protocol;
+  const host = window.location.hostname;
+  return proto + "//" + host + ":" + port;
+}
+async function copyText(v) {
+  if (!v) return false;
+  try {
+    await navigator.clipboard.writeText(v);
+    return true;
+  } catch (e) {
+    // Fallback for non-secure contexts (http) or older browsers.
+    try {
+      const ta = document.createElement("textarea");
+      ta.value = v;
+      ta.style.position = "fixed";
+      ta.style.opacity = "0";
+      document.body.appendChild(ta);
+      ta.focus();
+      ta.select();
+      const ok = document.execCommand("copy");
+      document.body.removeChild(ta);
+      return ok;
+    } catch (e2) {
+      return false;
+    }
+  }
+}
+document.addEventListener("DOMContentLoaded", () => {
+  const el = document.getElementById("nav-ray-dashboard");
+  if (el) el.href = curOriginWithPort(8265);
+});
+""".strip()
+
+
+def _nav(active: str) -> str:
+    links = [
+        ("login", "/ui/login", "Login"),
+        ("tasks", "/ui/tasks", "Tasks"),
+        ("new", "/ui/tasks/new", "New Task"),
+        ("data", "/ui/data", "Data"),
+        ("admin", "/ui/admin", "Admin"),
+        ("ray", "#", "Ray Dashboard"),
+    ]
+    items = []
+    for key, href, label in links:
+        cls = "active" if key == active else ""
+        extra = ""
+        if key == "ray":
+            extra = ' id="nav-ray-dashboard" target="_blank" rel="noopener"'
+        items.append(f'<a class="{cls}" href="{href}"{extra}>{html.escape(label)}</a>')
+    return "\n".join(items)
+
+
+def _page(title: str, active: str, body: str, script: str = "") -> str:
+    return f"""<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <title>{html.escape(title)}</title>
+  <style>{_BASE_CSS}</style>
+</head>
+<body>
+  <div class="layout">
+    <nav class="nav">
+      <div class="brand">Argus MVP</div>
+      {_nav(active)}
+      <div style="margin-top:12px" class="muted">
+        Token stored in browser localStorage as <code>mvp_token</code>.
+      </div>
+    </nav>
+    <main class="main">
+      {body}
+    </main>
+  </div>
+  <script>{_BASE_JS}</script>
+  <script>{script}</script>
+</body>
+</html>"""
+
+
+def register_ui_routes(app: FastAPI) -> None:
+    @app.get("/ui")
+    async def ui_root() -> RedirectResponse:
+        return RedirectResponse(url="/ui/tasks")
+
+    @app.get("/ui/login")
+    async def ui_login() -> HTMLResponse:
+        body = """
+<h1>Login</h1>
+<div class="card">
+  <div class="muted">Paste your API token (without the <code>Bearer</code> prefix).</div>
+  <div style="height:10px"></div>
+  <input id="tok" placeholder="token..." />
+  <div style="height:10px"></div>
+  <div class="row">
+    <button class="btn" id="save">Save</button>
+    <button class="btn" id="clear">Clear</button>
+    <a href="/ui/tasks" class="btn" style="display:inline-block">Go to Tasks</a>
+  </div>
+  <div style="height:10px"></div>
+  <div id="msg" class="muted"></div>
+</div>
+""".strip()
+        script = """
+const tokEl = document.getElementById("tok");
+const msg = document.getElementById("msg");
+tokEl.value = mvpTokenGet();
+document.getElementById("save").onclick = () => { mvpTokenSet(tokEl.value); msg.textContent = "Saved."; };
+document.getElementById("clear").onclick = () => { mvpTokenSet(""); tokEl.value = ""; msg.textContent = "Cleared."; };
+""".strip()
+        return HTMLResponse(content=_page("Login", "login", body, script))
+
+    @app.get("/ui/tasks")
+    async def ui_tasks() -> HTMLResponse:
+        body = """
+<h1>Tasks</h1>
+<div class="card">
+  <div class="row">
+    <button class="btn" id="refresh">Refresh</button>
+    <a class="btn" href="/ui/tasks/new" style="display:inline-block">New Task</a>
+  </div>
+  <div style="height:10px"></div>
+  <div id="out" class="muted">Loading...</div>
+</div>
+""".strip()
+        script = """
+const out = document.getElementById("out");
+async function refresh() {
+  out.textContent = "Loading...";
+  try {
+    const q = await apiJson("/api/v2/queue");
+    const completedLimit = 25;
+    const completedOffset = Number(localStorage.getItem("mvp_completed_offset") || "0") || 0;
+    const done = await apiJson("/api/v2/tasks?limit=" + completedLimit + "&offset=" + completedOffset + "&states=SUCCEEDED,FAILED,CANCELED");
+
+    function pill(state) {
+    const s = String(state || "");
+    if (s === "SUCCEEDED") return `<span class="pill ok">${s}</span>`;
+    if (s === "FAILED") return `<span class="pill bad">${s}</span>`;
+    if (s === "CANCELED") return `<span class="pill">${s}</span>`;
+    if (s === "RUNNING") return `<span class="pill ok">${s}</span>`;
+    if (s === "QUEUED" || s === "PENDING_RESOURCES" || s === "SUBMITTING" || s === "SUBMITTED") return `<span class="pill">${s}</span>`;
+    return `<span class="pill">${s}</span>`;
+  }
+  function row(t) {
+    const id = t.task_id;
+    return `<tr>
+      <td><a href="/ui/tasks/${id}">${id}</a></td>
+      <td>${t.workload}</td>
+      <td>${pill(t.state)}</td>
+      <td>${t.nnodes} x ${t.n_gpus_per_node} GPU</td>
+      <td>${t.updated_at || ""}</td>
+    </tr>`;
+  }
+
+  const running = (q.running || []).map(row).join("");
+  const pending = (q.pending || []).map(row).join("");
+  const doneRows = (done.tasks || []).map(row).join("");
+  const pageNo = Math.floor(completedOffset / completedLimit) + 1;
+  const prevDisabled = completedOffset <= 0;
+  const nextDisabled = !done.has_more;
+  out.innerHTML = `
+    <div class="muted">Tip: configure token in <a href="/ui/login">Login</a>.</div>
+    <div style="height:10px"></div>
+    <h3>Running</h3>
+    <table><thead><tr><th>Task</th><th>Workload</th><th>State</th><th>Resources</th><th>Updated</th></tr></thead><tbody>${running || "<tr><td colspan=5 class=muted>(none)</td></tr>"}</tbody></table>
+    <div style="height:12px"></div>
+    <h3>Pending</h3>
+    <table><thead><tr><th>Task</th><th>Workload</th><th>State</th><th>Resources</th><th>Updated</th></tr></thead><tbody>${pending || "<tr><td colspan=5 class=muted>(none)</td></tr>"}</tbody></table>
+    <div style="height:12px"></div>
+    <h3>Completed</h3>
+    <div class="row" style="justify-content: space-between; margin-bottom: 8px;">
+      <div class="muted">Page ${pageNo}</div>
+      <div class="row">
+        <button class="btn" id="done-prev" ${prevDisabled ? "disabled" : ""}>Prev</button>
+        <button class="btn" id="done-next" ${nextDisabled ? "disabled" : ""}>Next</button>
+      </div>
+    </div>
+    <table><thead><tr><th>Task</th><th>Workload</th><th>State</th><th>Resources</th><th>Updated</th></tr></thead><tbody>${doneRows || "<tr><td colspan=5 class=muted>(none)</td></tr>"}</tbody></table>
+  `;
+
+  const prevBtn = document.getElementById("done-prev");
+  const nextBtn = document.getElementById("done-next");
+  if (prevBtn) prevBtn.onclick = () => {
+    const cur = Number(localStorage.getItem("mvp_completed_offset") || "0") || 0;
+    const next = Math.max(0, cur - completedLimit);
+    localStorage.setItem("mvp_completed_offset", String(next));
+    refresh();
+  };
+  if (nextBtn) nextBtn.onclick = () => {
+    const cur = Number(localStorage.getItem("mvp_completed_offset") || "0") || 0;
+    const next = cur + completedLimit;
+    localStorage.setItem("mvp_completed_offset", String(next));
+    refresh();
+  };
+  } catch (e) {
+    out.textContent = "Error: " + (e.status || "") + "\\n" + (e.body || String(e));
+  }
+}
+document.getElementById("refresh").onclick = refresh;
+refresh();
+""".strip()
+        return HTMLResponse(content=_page("Tasks", "tasks", body, script))
+
+    @app.get("/ui/tasks/new")
+    async def ui_new_task() -> HTMLResponse:
+        ppo = """# PPO TaskSpec (YAML)
+workload: ppo
+nnodes: 2
+n_gpus_per_node: 4
+code_path: /private/common/code/verl/verl_repo
+train_file: /private/common/datasets/gsm8k/train.parquet
+val_file: /private/common/datasets/gsm8k/test.parquet
+model_id: Qwen/Qwen2.5-0.5B-Instruct
+""".strip()
+        grpo = """# GRPO TaskSpec (YAML)
+workload: grpo
+nnodes: 2
+n_gpus_per_node: 4
+code_path: /private/common/code/verl/verl_repo
+train_file: /private/common/datasets/gsm8k/train.parquet
+val_file: /private/common/datasets/gsm8k/test.parquet
+model_id: Qwen/Qwen2.5-0.5B-Instruct
+""".strip()
+        sft = """# SFT TaskSpec (YAML)
+workload: sft
+nnodes: 1
+n_gpus_per_node: 1
+code_path: /private/common/code/verl/verl_repo
+train_file: /private/common/datasets/gsm8k_sft/train.parquet
+val_file: /private/common/datasets/gsm8k_sft/test.parquet
+model_id: Qwen/Qwen2.5-0.5B-Instruct
+""".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 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>
+  </div>
+  <div style="height:10px"></div>
+  <textarea id="yaml" rows="16">{html.escape(ppo)}</textarea>
+  <div style="height:10px"></div>
+  <div class="row">
+    <button class="btn" id="submit">Submit</button>
+    <a class="btn" href="/ui/tasks" style="display:inline-block">Back</a>
+  </div>
+  <div style="height:10px"></div>
+  <pre id="msg" class="muted"></pre>
+</div>
+""".strip()
+        tpl_ppo = json.dumps(ppo)
+        tpl_grpo = json.dumps(grpo)
+        tpl_sft = json.dumps(sft)
+        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__;
+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("submit").onclick = async () => {
+  msg.textContent = "Submitting...";
+  const body = yamlEl.value;
+  const resp = await apiFetch("/api/v2/tasks", { method: "POST", headers: {"Content-Type":"text/plain"}, body });
+  const text = await resp.text();
+  if (!resp.ok) { msg.textContent = "Error: " + resp.status + "\\n" + text; return; }
+  const obj = JSON.parse(text);
+  msg.textContent = "OK: " + fmtJson(obj);
+  if (obj.task_id) window.location.href = "/ui/tasks/" + obj.task_id;
+};
+""".strip()
+            .replace("__TPL_PPO__", tpl_ppo)
+            .replace("__TPL_GRPO__", tpl_grpo)
+            .replace("__TPL_SFT__", tpl_sft)
+        )
+        return HTMLResponse(content=_page("New Task", "new", body, script))
+
+    @app.get("/ui/tasks/{task_id}")
+    async def ui_task_detail(task_id: str) -> HTMLResponse:
+        safe_id = html.escape(task_id)
+        body = f"""
+<h1>Task: <code>{safe_id}</code></h1>
+<div class="card">
+  <div class="row">
+    <a class="btn" href="/ui/tasks/{safe_id}/logs" style="display:inline-block">Logs</a>
+    <button class="btn" id="refresh">Refresh</button>
+    <button class="btn danger" id="cancel">Cancel</button>
+    <a class="btn" href="/ui/tasks" style="display:inline-block">Back</a>
+  </div>
+  <div style="height:10px"></div>
+  <pre id="out" class="muted">Loading...</pre>
+</div>
+""".strip()
+        script = f"""
+document.getElementById("nav-ray-dashboard").href = curOriginWithPort(8265);
+const out = document.getElementById("out");
+async function refresh() {{
+  out.textContent = "Loading...";
+  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));
+}}
+document.getElementById("refresh").onclick = refresh;
+document.getElementById("cancel").onclick = async () => {{
+  if (!confirm("Cancel this task?")) return;
+  const resp = await apiFetch("/api/v2/tasks/{task_id}:cancel", {{ method: "POST" }});
+  const text = await resp.text();
+  out.textContent = (resp.ok ? "Canceled.\\n" : "Error: " + resp.status + "\\n") + text;
+  setTimeout(refresh, 800);
+}};
+refresh();
+""".strip()
+        return HTMLResponse(content=_page(f"Task {task_id}", "tasks", body, script))
+
+    @app.get("/ui/tasks/{task_id}/logs")
+    async def ui_task_logs(task_id: str) -> HTMLResponse:
+        safe_id = html.escape(task_id)
+        body = f"""
+<h1>Logs: <code>{safe_id}</code></h1>
+<div class="card">
+  <div class="row">
+    <button class="btn" id="refresh">Refresh</button>
+    <label class="muted">Auto refresh <input type="checkbox" id="auto" /></label>
+    <a class="btn" href="/ui/tasks/{safe_id}" style="display:inline-block">Back</a>
+  </div>
+  <div style="height:10px"></div>
+  <pre id="out" class="muted">Loading...</pre>
+</div>
+""".strip()
+        script = f"""
+document.getElementById("nav-ray-dashboard").href = curOriginWithPort(8265);
+const out = document.getElementById("out");
+let timer = null;
+async function refresh() {{
+  const resp = await apiFetch("/api/v2/tasks/{task_id}/logs?tail=4000");
+  const text = await resp.text();
+  out.textContent = resp.ok ? text : ("Error: " + resp.status + "\\n" + text);
+}}
+document.getElementById("refresh").onclick = refresh;
+document.getElementById("auto").onchange = (e) => {{
+  if (e.target.checked) {{
+    timer = setInterval(refresh, 2000);
+  }} else {{
+    if (timer) clearInterval(timer);
+    timer = null;
+  }}
+}};
+refresh();
+""".strip()
+        return HTMLResponse(content=_page(f"Logs {task_id}", "tasks", body, script))
+
+    @app.get("/ui/data")
+    async def ui_data() -> HTMLResponse:
+        body = """
+<h1>Data</h1>
+<div class="card">
+  <div class="muted">User files live under your home directory. Keep long-term artifacts in <code>models/</code> or <code>datasets/</code>.</div>
+  <div style="height:14px"></div>
+
+  <div class="row">
+    <div style="flex:1; min-width:260px">
+      <div class="muted">Username</div>
+      <div style="height:6px"></div>
+      <div class="row" style="gap:8px">
+        <input id="u" readonly />
+        <button class="btn" id="copy-u">Copy</button>
+      </div>
+    </div>
+    <div style="flex:1; min-width:260px">
+      <div class="muted">SFTPGo password</div>
+      <div style="height:6px"></div>
+      <div class="row" style="gap:8px">
+        <input id="p" placeholder="Click Reset to generate..." />
+        <button class="btn" id="copy-p">Copy</button>
+        <button class="btn" id="reset-p">Reset</button>
+      </div>
+    </div>
+  </div>
+
+  <div style="height:12px"></div>
+  <div class="row">
+    <a class="btn" id="sftp-web" target="_blank" rel="noopener" href="#">Open SFTPGo Web Client (:8081)</a>
+  </div>
+
+  <div style="height:12px"></div>
+  <div class="muted">
+    You can also use an SFTP client (e.g. FileZilla) with the same username/password.
+    Host: <code id="sftp-host"></code>, Port: <code id="sftp-port"></code>.
+  </div>
+
+  <div style="height:14px"></div>
+  <pre id="out" class="muted">Loading...</pre>
+</div>
+""".strip()
+        script = """
+const out = document.getElementById("out");
+document.getElementById("nav-ray-dashboard").href = curOriginWithPort(8265);
+const u = document.getElementById("u");
+const p = document.getElementById("p");
+const sftpWeb = document.getElementById("sftp-web");
+const sftpHost = document.getElementById("sftp-host");
+const sftpPort = document.getElementById("sftp-port");
+document.getElementById("copy-u").onclick = async () => { await copyText(u.value || ""); };
+document.getElementById("copy-p").onclick = async () => { await copyText(p.value || ""); };
+
+async function refresh() {
+  const resp = await apiFetch("/api/v2/me");
+  const text = await resp.text();
+  if (!resp.ok) { out.textContent = "Error: " + resp.status + "\\n" + text; return; }
+  const obj = JSON.parse(text);
+  u.value = (obj.user_id || "");
+  const cached = mvpSftpPasswordGet();
+  if (cached) p.value = cached;
+  const host = curOriginWithPort(8081);
+  sftpWeb.href = host + "/web/client";
+  sftpHost.textContent = (obj.sftp && obj.sftp.host) ? obj.sftp.host : window.location.hostname;
+  sftpPort.textContent = (obj.sftp && obj.sftp.port) ? String(obj.sftp.port) : "2022";
+  out.textContent = fmtJson(obj);
+}
+document.getElementById("reset-p").onclick = async () => {
+  p.value = "";
+  const resp = await apiFetch("/api/v2/me/sftp:reset_password", { method: "POST" });
+  const text = await resp.text();
+  if (!resp.ok) { out.textContent = "Error: " + resp.status + "\\n" + text; return; }
+  const obj = JSON.parse(text);
+  p.value = obj.password || "";
+  mvpSftpPasswordSet(p.value);
+  out.textContent = "SFTPGo password rotated.\\n\\n" + fmtJson(obj);
+};
+refresh();
+""".strip()
+        return HTMLResponse(content=_page("Data", "data", body, script))
+
+    @app.get("/ui/admin")
+    async def ui_admin() -> HTMLResponse:
+        body = """
+<h1>Admin</h1>
+<div class="card">
+  <div class="muted">
+    This page requires the <code>admin</code> token (set it in <a href="/ui/login">Login</a>).
+  </div>
+  <div style="height:14px"></div>
+
+  <h3>Create user</h3>
+  <div class="row">
+    <input id="new-user-id" placeholder="user_id (e.g. alice)" style="max-width:320px" />
+    <input id="new-display-name" placeholder="display_name (optional)" style="max-width:320px" />
+    <button class="btn" id="create-user">Create</button>
+  </div>
+  <div style="height:10px"></div>
+  <pre id="create-msg" class="muted"></pre>
+
+  <div style="height:14px"></div>
+  <div class="row">
+    <button class="btn" id="refresh">Refresh</button>
+  </div>
+  <div style="height:10px"></div>
+  <div id="out" class="muted">Loading...</div>
+</div>
+""".strip()
+        script = """
+const out = document.getElementById("out");
+const createMsg = document.getElementById("create-msg");
+const userIdEl = document.getElementById("new-user-id");
+const displayNameEl = document.getElementById("new-display-name");
+
+function esc(s) {
+  s = String(s || "");
+  return s.replaceAll("&","&amp;").replaceAll("<","&lt;").replaceAll(">","&gt;");
+}
+
+async function refresh() {
+  out.textContent = "Loading...";
+  try {
+    const obj = await apiJson("/api/v2/users?limit=200");
+    const users = (obj.users || []);
+    function row(u) {
+      const uid = u.user_id;
+      const tok = u.token || "";
+      const tokShort = tok ? (tok.length > 18 ? (tok.slice(0, 18) + "…") : tok) : "";
+      const created = u.created_at || "";
+      const updated = u.updated_at || "";
+      const tCreated = u.token_created_at || "";
+      const tUsed = u.token_last_used_at || "";
+      return `<tr>
+        <td><code>${esc(uid)}</code></td>
+        <td class="muted">${esc(created)}</td>
+        <td class="muted">${esc(updated)}</td>
+        <td>
+          <div class="row" style="gap:8px">
+            <code title="${esc(tok)}">${esc(tokShort)}</code>
+            <button class="btn" data-copy="${esc(tok)}">Copy</button>
+            <button class="btn" data-issue="${esc(uid)}">Issue token</button>
+          </div>
+          <div class="muted" style="margin-top:6px">token_created_at: ${esc(tCreated)}; last_used_at: ${esc(tUsed)}</div>
+        </td>
+      </tr>`;
+    }
+    out.innerHTML = `
+      <table>
+        <thead><tr><th>User</th><th>Created</th><th>Updated</th><th>Token</th></tr></thead>
+        <tbody>${users.map(row).join("") || "<tr><td colspan=4 class=muted>(none)</td></tr>"}</tbody>
+      </table>
+    `;
+    for (const btn of out.querySelectorAll("button[data-copy]")) {
+      btn.onclick = async () => { await copyText(btn.getAttribute("data-copy") || ""); };
+    }
+    for (const btn of out.querySelectorAll("button[data-issue]")) {
+      btn.onclick = async () => {
+        const uid = btn.getAttribute("data-issue");
+        if (!uid) return;
+        try {
+          const r = await apiJson("/api/v2/users/" + encodeURIComponent(uid) + "/tokens", { method: "POST" });
+          createMsg.textContent = "Issued token:\\n" + fmtJson(r);
+          await refresh();
+        } catch (e) {
+          createMsg.textContent = "Error issuing token: " + (e.status || "") + "\\n" + (e.body || String(e));
+        }
+      };
+    }
+  } catch (e) {
+    out.textContent = "Error: " + (e.status || "") + "\\n" + (e.body || String(e));
+  }
+}
+
+document.getElementById("refresh").onclick = refresh;
+document.getElementById("create-user").onclick = async () => {
+  createMsg.textContent = "Creating...";
+  const user_id = (userIdEl.value || "").trim();
+  const display_name = (displayNameEl.value || "").trim();
+  if (!user_id) { createMsg.textContent = "user_id is required"; return; }
+  const payload = { user_id: user_id };
+  if (display_name) payload.display_name = display_name;
+  try {
+    const r = await apiJson("/api/v2/users", { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify(payload) });
+    createMsg.textContent = "Created:\\n" + fmtJson(r);
+    userIdEl.value = "";
+    displayNameEl.value = "";
+    await refresh();
+  } catch (e) {
+    createMsg.textContent = "Error: " + (e.status || "") + "\\n" + (e.body || String(e));
+  }
+};
+
+refresh();
+""".strip()
+        return HTMLResponse(content=_page("Admin", "admin", body, script))
--- a/src/mvp/py/tests/test_app.py
+++ b/src/mvp/py/tests/test_app.py
@@ -16,6 +16,11 @@ def _write_config(tmp_path: Path) -> Path:
            "entrypoint_resources": {"worker_node": 1},
            "runtime_env": {"env_vars": {}},
        },
+        "data": {
+            # Avoid touching real /private in tests. Keep ray.shared_root as /private
+            # so existing path validation tests remain unchanged.
+            "user_root": str(tmp_path / "users"),
+        },
        "service": {
            "api": {"host": "127.0.0.1", "port": 0},
            "auth": {"token_env": "MVP_INTERNAL_TOKEN"},
@@ -95,6 +100,17 @@ def test_task_submit_get_cancel_logs_queue(tmp_path: Path, monkeypatch):
        assert r3.status_code == 200
        assert "pending" in r3.json()

+        r3b = c.get("/api/v2/tasks?limit=10", headers=headers)
+        assert r3b.status_code == 200
+        assert any(t.get("task_id") == "tid1" for t in r3b.json().get("tasks", []))
+
+        r3c = c.get("/api/v2/tasks?limit=10&offset=0&states=QUEUED", headers=headers)
+        assert r3c.status_code == 200
+        assert all(t.get("state") == "QUEUED" for t in r3c.json().get("tasks", []))
+
+        r3d = c.get("/api/v2/tasks?states=NOPE", headers=headers)
+        assert r3d.status_code == 400
+
        r4 = c.post("/api/v2/tasks/tid1:cancel", headers=headers)
        assert r4.status_code == 200
        assert r4.json()["state"] == "CANCELED"
@@ -118,6 +134,14 @@ def test_task_submit_get_cancel_logs_queue(tmp_path: Path, monkeypatch):
        db.create_attempt(task_id="tid2", attempt_no=1, ray_submission_id="sid2")
        db.set_task_state(task_id="tid2", state="RUNNING", latest_attempt_no=1)

+        r6 = c.get("/api/v2/tasks?limit=1&offset=0&states=RUNNING", headers=headers)
+        assert r6.status_code == 200
+        assert any(t.get("task_id") == "tid2" for t in r6.json().get("tasks", []))
+
+        r7 = c.get("/api/v2/tasks?limit=1&offset=1&states=RUNNING", headers=headers)
+        assert r7.status_code == 200
+        assert "has_more" in r7.json()
+
        r5 = c.get("/api/v2/tasks/tid2/logs?tail=1", headers=headers)
        assert r5.status_code == 200
        assert r5.text.strip() == "c"
@@ -163,3 +187,102 @@ def test_submit_rejects_invalid_jobspec(tmp_path: Path, monkeypatch):
    with TestClient(app) as c:
        r = c.post("/api/v2/tasks", headers={"authorization": "Bearer token1"}, data="workload: nope\n")
        assert r.status_code == 400
+
+
+def test_me_sftp_reset_password_disabled_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))
+
+    # 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.init()
+    db.create_user(user_id="u1", display_name=None)
+    token = db.issue_token(user_id="u1")
+
+    with TestClient(app) as c:
+        r = c.post("/api/v2/me/sftp:reset_password", headers={"authorization": f"Bearer {token}"})
+        assert r.status_code == 400
+
+
+def test_me_sftp_reset_password_enabled_returns_password(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+
+    cfg = yaml.safe_load(_write_config(tmp_path).read_text(encoding="utf-8"))
+    cfg["data"]["sftpgo"] = {
+        "enabled": True,
+        "host": "127.0.0.1",
+        "sftp_port": 2022,
+        "admin_api_base": "http://127.0.0.1:8081",
+        "admin_user": "admin",
+        "admin_password_env": "SFTPGO_ADMIN_PASSWORD",
+    }
+    cfg_path = tmp_path / "cfg_sftp.yaml"
+    cfg_path.write_text(yaml.safe_dump(cfg), encoding="utf-8")
+
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "token1")
+    monkeypatch.setenv("SFTPGO_ADMIN_PASSWORD", "pw1")
+
+    class _FakeSFTPGo:
+        def __init__(self, **kwargs):
+            self.reset = []
+            self.enabled = []
+
+        def reset_password(self, username: str, new_password: str, home_dir: str):
+            assert username
+            assert new_password
+            assert home_dir
+            self.reset.append((username, home_dir))
+
+        def enable_user(self, username: str, home_dir: str):
+            self.enabled.append((username, home_dir))
+
+    fake_client = _FakeSFTPGo()
+
+    class _FakeSFTPGoFactory:
+        def __call__(self, **kwargs):
+            return fake_client
+
+    class _Scheduler:
+        def __init__(self, **kwargs):
+            self.tool = object()
+
+        def run_forever(self, stop_flag):
+            return None
+
+    monkeypatch.setattr(app_mod, "Scheduler", _Scheduler)
+    monkeypatch.setattr(app_mod, "SFTPGoAdminClient", _FakeSFTPGoFactory())
+    app = app_mod.create_app(str(cfg_path))
+
+    # seed user in DB
+    from argus.service.db import Db
+    from argus.service.config import V2Config
+
+    v2_cfg = V2Config.from_root_dict(cfg)
+    db = Db(v2_cfg.sqlite.db_path)
+    db.init()
+    db.create_user(user_id="u1", display_name=None)
+    token = db.issue_token(user_id="u1")
+
+    with TestClient(app) as c:
+        r = c.post("/api/v2/me/sftp:reset_password", headers={"authorization": f"Bearer {token}"})
+        assert r.status_code == 200
+        j = r.json()
+        assert j["user_id"] == "u1"
+        assert isinstance(j["password"], str) and len(j["password"]) >= 8
--- a/src/mvp/py/tests/test_janitor.py
+++ b/src/mvp/py/tests/test_janitor.py
@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from argus.service.db import Db
+from argus.service.janitor import JobsJanitor
+
+
+def _iso_z(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _mk_job_dir(user_root: Path, user_id: str, sid: str) -> Path:
+    p = user_root / user_id / "jobs" / sid
+    p.mkdir(parents=True, exist_ok=True)
+    (p / "marker.txt").write_text("x", encoding="utf-8")
+    return p
+
+
+def _mk_trash_dir(user_root: Path, user_id: str, sid: str) -> Path:
+    p = user_root / user_id / "trash" / "jobs" / sid
+    p.mkdir(parents=True, exist_ok=True)
+    (p / "marker.txt").write_text("x", encoding="utf-8")
+    return p
+
+
+def test_janitor_moves_jobs_to_trash_after_3_days(tmp_path: Path) -> None:
+    db_path = tmp_path / "mvp.sqlite3"
+    user_root = tmp_path / "users"
+    db = Db(str(db_path))
+    db.init()
+
+    task_id = "t1"
+    user_id = "alice"
+    sid = "sid-a01"
+    db.create_task_v25(task_id=task_id, user_id=user_id, workload="sft", jobspec_yaml="{}", nnodes=1, n_gpus_per_node=1)
+    db.create_attempt(task_id=task_id, attempt_no=1, ray_submission_id=sid)
+
+    now = datetime(2025, 1, 10, tzinfo=timezone.utc)
+    ended = now - timedelta(days=4)
+    db.update_attempt(task_id=task_id, attempt_no=1, end_time=_iso_z(ended), ray_status="SUCCEEDED")
+
+    src = _mk_job_dir(user_root, user_id, sid)
+    jan = JobsJanitor(db=db, user_root=str(user_root), trash_after_days=3, purge_after_days=7, interval_s=1)
+    jan.tick_once(now=now)
+
+    assert not src.exists()
+    dst = user_root / user_id / "trash" / "jobs" / sid
+    assert dst.exists()
+    assert (dst / "marker.txt").read_text(encoding="utf-8") == "x"
+
+
+def test_janitor_purges_from_trash_after_7_days(tmp_path: Path) -> None:
+    db_path = tmp_path / "mvp.sqlite3"
+    user_root = tmp_path / "users"
+    db = Db(str(db_path))
+    db.init()
+
+    task_id = "t2"
+    user_id = "alice"
+    sid = "sid-a01"
+    db.create_task_v25(task_id=task_id, user_id=user_id, workload="ppo", jobspec_yaml="{}", nnodes=1, n_gpus_per_node=1)
+    db.create_attempt(task_id=task_id, attempt_no=1, ray_submission_id=sid)
+
+    now = datetime(2025, 1, 10, tzinfo=timezone.utc)
+    ended = now - timedelta(days=8)
+    db.update_attempt(task_id=task_id, attempt_no=1, end_time=_iso_z(ended), ray_status="FAILED")
+
+    _mk_trash_dir(user_root, user_id, sid)
+    jan = JobsJanitor(db=db, user_root=str(user_root), trash_after_days=3, purge_after_days=7, interval_s=1)
+    jan.tick_once(now=now)
+
+    dst = user_root / user_id / "trash" / "jobs" / sid
+    assert not dst.exists()
+
+
+def test_janitor_does_not_touch_recent_jobs(tmp_path: Path) -> None:
+    db_path = tmp_path / "mvp.sqlite3"
+    user_root = tmp_path / "users"
+    db = Db(str(db_path))
+    db.init()
+
+    task_id = "t3"
+    user_id = "alice"
+    sid = "sid-a01"
+    db.create_task_v25(task_id=task_id, user_id=user_id, workload="grpo", jobspec_yaml="{}", nnodes=1, n_gpus_per_node=1)
+    db.create_attempt(task_id=task_id, attempt_no=1, ray_submission_id=sid)
+
+    now = datetime(2025, 1, 10, tzinfo=timezone.utc)
+    ended = now - timedelta(days=1)
+    db.update_attempt(task_id=task_id, attempt_no=1, end_time=_iso_z(ended), ray_status="FAILED")
+
+    src = _mk_job_dir(user_root, user_id, sid)
+    jan = JobsJanitor(db=db, user_root=str(user_root), trash_after_days=3, purge_after_days=7, interval_s=1)
+    jan.tick_once(now=now)
+
+    assert src.exists()
+    assert not (user_root / user_id / "trash" / "jobs" / sid).exists()
+
+
+def test_janitor_skips_tasks_without_user_id(tmp_path: Path) -> None:
+    db_path = tmp_path / "mvp.sqlite3"
+    user_root = tmp_path / "users"
+    db = Db(str(db_path))
+    db.init()
+
+    task_id = "legacy"
+    sid = "sid-legacy"
+    db.create_task(task_id=task_id, workload="sft", jobspec_yaml="{}", nnodes=1, n_gpus_per_node=1)
+    db.create_attempt(task_id=task_id, attempt_no=1, ray_submission_id=sid)
+
+    now = datetime(2025, 1, 10, tzinfo=timezone.utc)
+    ended = now - timedelta(days=10)
+    db.update_attempt(task_id=task_id, attempt_no=1, end_time=_iso_z(ended), ray_status="SUCCEEDED")
+
+    # Even if someone created a matching directory under user_root, janitor shouldn't touch it because user_id is NULL.
+    src = _mk_job_dir(user_root, "alice", sid)
+    jan = JobsJanitor(db=db, user_root=str(user_root), trash_after_days=3, purge_after_days=7, interval_s=1)
+    jan.tick_once(now=now)
+    assert src.exists()
+
+
+def test_janitor_validates_retention_days(tmp_path: Path) -> None:
+    db = Db(str(tmp_path / "mvp.sqlite3"))
+    db.init()
+    try:
+        JobsJanitor(db=db, user_root="/tmp", trash_after_days=-1, purge_after_days=7, interval_s=1)
+        raise AssertionError("expected ValueError")
+    except ValueError:
+        pass
+    try:
+        JobsJanitor(db=db, user_root="/tmp", trash_after_days=3, purge_after_days=1, interval_s=1)
+        raise AssertionError("expected ValueError")
+    except ValueError:
+        pass
+
+
+def test_janitor_noop_when_disabled(tmp_path: Path) -> None:
+    db = Db(str(tmp_path / "mvp.sqlite3"))
+    db.init()
+    user_root = tmp_path / "users"
+    jan = JobsJanitor(db=db, user_root=str(user_root), trash_after_days=0, purge_after_days=0, interval_s=1)
+    jan.tick_once(now=datetime(2025, 1, 1, tzinfo=timezone.utc))
+
+
+def test_janitor_handles_invalid_end_time_and_missing_fields(tmp_path: Path) -> None:
+    db_path = tmp_path / "mvp.sqlite3"
+    user_root = tmp_path / "users"
+    db = Db(str(db_path))
+    db.init()
+
+    now = datetime(2025, 1, 10, tzinfo=timezone.utc)
+    cutoff_ended = now - timedelta(days=10)
+
+    # Missing end_time (empty string) => should be skipped.
+    db.create_task_v25(task_id="t4", user_id="alice", workload="sft", jobspec_yaml="{}", nnodes=1, n_gpus_per_node=1)
+    db.create_attempt(task_id="t4", attempt_no=1, ray_submission_id="sid-empty")
+    db.update_attempt(task_id="t4", attempt_no=1, end_time="", ray_status="SUCCEEDED")
+
+    # Invalid ISO but still lexicographically <= cutoff => should be skipped.
+    db.create_task_v25(task_id="t5", user_id="alice", workload="sft", jobspec_yaml="{}", nnodes=1, n_gpus_per_node=1)
+    db.create_attempt(task_id="t5", attempt_no=1, ray_submission_id="sid-bad")
+    db.update_attempt(task_id="t5", attempt_no=1, end_time="2025-01-01T00:00:00ZZ", ray_status="FAILED")
+
+    _mk_job_dir(user_root, "alice", "sid-empty")
+    _mk_job_dir(user_root, "alice", "sid-bad")
+    jan = JobsJanitor(db=db, user_root=str(user_root), trash_after_days=3, purge_after_days=7, interval_s=1)
+    jan.tick_once(now=cutoff_ended + timedelta(days=10))
+
+    assert (user_root / "alice" / "jobs" / "sid-empty").exists()
+    assert (user_root / "alice" / "jobs" / "sid-bad").exists()
+
+
+def test_janitor_purge_moves_from_jobs_then_deletes(tmp_path: Path, monkeypatch) -> None:
+    db_path = tmp_path / "mvp.sqlite3"
+    user_root = tmp_path / "users"
+    db = Db(str(db_path))
+    db.init()
+
+    task_id = "t6"
+    user_id = "alice"
+    sid = "sid-a01"
+    db.create_task_v25(task_id=task_id, user_id=user_id, workload="ppo", jobspec_yaml="{}", nnodes=1, n_gpus_per_node=1)
+    db.create_attempt(task_id=task_id, attempt_no=1, ray_submission_id=sid)
+
+    now = datetime(2025, 1, 10, tzinfo=timezone.utc)
+    ended = now - timedelta(days=9)
+    db.update_attempt(task_id=task_id, attempt_no=1, end_time=_iso_z(ended), ray_status="SUCCEEDED")
+
+    src = _mk_job_dir(user_root, user_id, sid)
+    jan = JobsJanitor(db=db, user_root=str(user_root), trash_after_days=3, purge_after_days=7, interval_s=1)
+
+    jan.tick_once(now=now)
+    assert not src.exists()
+    assert not (user_root / user_id / "trash" / "jobs" / sid).exists()
+
+
+def test_janitor_run_forever_requires_event_like(tmp_path: Path) -> None:
+    db = Db(str(tmp_path / "mvp.sqlite3"))
+    db.init()
+    jan = JobsJanitor(db=db, user_root=str(tmp_path / "users"), trash_after_days=3, purge_after_days=7, interval_s=1)
+    try:
+        jan.run_forever(object())
+        raise AssertionError("expected ValueError")
+    except ValueError:
+        pass
+
+
+def test_janitor_run_forever_survives_tick_exceptions(tmp_path: Path, monkeypatch) -> None:
+    db = Db(str(tmp_path / "mvp.sqlite3"))
+    db.init()
+    jan = JobsJanitor(db=db, user_root=str(tmp_path / "users"), trash_after_days=3, purge_after_days=7, interval_s=1)
+
+    class Flag:
+        def __init__(self) -> None:
+            self.n = 0
+
+        def is_set(self) -> bool:
+            self.n += 1
+            return self.n >= 2
+
+    monkeypatch.setattr(jan, "tick_once", lambda **_: (_ for _ in ()).throw(RuntimeError("boom")))
+    monkeypatch.setattr("argus.service.janitor.time.sleep", lambda *_: None)
+    jan.run_forever(Flag())
--- a/src/mvp/py/tests/test_service_config.py
+++ b/src/mvp/py/tests/test_service_config.py
@@ -38,3 +38,18 @@ def test_v2_config_requires_mappings():
        V2Config.from_root_dict({"service": ["nope"]})
    with pytest.raises(ValueError, match="config\\.service\\.\\{api,auth,sqlite,scheduler\\} must be mappings"):
        V2Config.from_root_dict({"service": {"api": [1], "auth": {}, "sqlite": {}, "scheduler": {}}})
+
+
+def test_v2_config_requires_data_mappings():
+    from argus.service.config import V2Config
+
+    base = {
+        "ray": {"shared_root": "/private"},
+        "service": {"api": {}, "auth": {}, "sqlite": {}, "scheduler": {}},
+    }
+
+    with pytest.raises(ValueError, match="config\\.data must be a mapping"):
+        V2Config.from_root_dict({**base, "data": ["nope"]})
+
+    with pytest.raises(ValueError, match="config\\.data\\.\\{sftpgo,retention\\} must be mappings"):
+        V2Config.from_root_dict({**base, "data": {"sftpgo": ["x"], "retention": {}}})
--- a/src/mvp/py/tests/test_sftpgo.py
+++ b/src/mvp/py/tests/test_sftpgo.py
@@ -0,0 +1,322 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+from fastapi.testclient import TestClient
+
+
+def _write_config(tmp_path: Path, *, enabled: bool) -> Path:
+    cfg = {
+        "ray": {
+            "address": "http://127.0.0.1:8265",
+            "shared_root": "/private",
+            "entrypoint_resources": {"worker_node": 1},
+            "runtime_env": {"env_vars": {}},
+        },
+        "data": {
+            "user_root": str(tmp_path / "users"),
+            "sftpgo": {
+                "enabled": bool(enabled),
+                "admin_api_base": "http://127.0.0.1:8081/api/v2",
+                "admin_user": "admin",
+                "admin_password_env": "SFTPGO_ADMIN_PASSWORD",
+                "host": "h1.internal",
+                "sftp_port": 2022,
+            },
+        },
+        "service": {
+            "api": {"host": "127.0.0.1", "port": 0},
+            "auth": {"token_env": "MVP_INTERNAL_TOKEN"},
+            "sqlite": {"db_path": str(tmp_path / "mvp.sqlite3")},
+            "scheduler": {"tick_s": 1, "retry_interval_s": 1, "max_running_tasks": 1},
+        },
+    }
+    p = tmp_path / "cfg.yaml"
+    p.write_text(yaml.safe_dump(cfg), encoding="utf-8")
+    return p
+
+
+def test_create_user_calls_sftpgo_when_enabled(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+
+    cfg_path = _write_config(tmp_path, enabled=True)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "adm1")
+    monkeypatch.setenv("SFTPGO_ADMIN_PASSWORD", "pw1")
+
+    calls = {"create": [], "disable": [], "reset": []}
+
+    class _Client:
+        def __init__(self, **kwargs):
+            pass
+
+        def create_user(self, *, username: str, password: str, home_dir: str) -> None:
+            calls["create"].append((username, password, home_dir))
+
+        def enable_user(self, *, username: str, home_dir: str) -> None:
+            # Not used in this test, but required by app for idempotent upsert.
+            return None
+
+        def disable_user(self, *, username: str, home_dir: str) -> None:
+            calls["disable"].append(username)
+
+        def reset_password(self, *, username: str, new_password: str, home_dir: str) -> None:
+            calls["reset"].append((username, new_password))
+
+    monkeypatch.setattr(app_mod, "SFTPGoAdminClient", _Client)
+
+    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))
+
+    admin_headers = {"authorization": "Bearer adm1"}
+    with TestClient(app) as c:
+        r = c.post("/api/v2/users", headers=admin_headers, json={"user_id": "alice"})
+        assert r.status_code == 200
+        assert calls["create"]
+        username, password, home_dir = calls["create"][-1]
+        assert username == "alice"
+        assert password
+        assert home_dir.endswith("/users/alice")
+
+        r2 = c.post("/api/v2/users/alice:disable", headers=admin_headers)
+        assert r2.status_code == 200
+        assert calls["disable"] == ["alice"]
+
+        r3 = c.post("/api/v2/users/alice/sftp:reset_password", headers=admin_headers)
+        assert r3.status_code == 200
+        body = r3.json()
+        assert body["user_id"] == "alice"
+        assert body["password"]
+        assert calls["reset"] and calls["reset"][-1][0] == "alice"
+
+
+def test_create_user_upserts_when_sftpgo_user_already_exists(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+    from argus.service.sftpgo import SFTPGoError
+
+    cfg_path = _write_config(tmp_path, enabled=True)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "adm1")
+    monkeypatch.setenv("SFTPGO_ADMIN_PASSWORD", "pw1")
+
+    calls = {"create": 0, "reset": [], "enable": []}
+
+    class _Client:
+        def __init__(self, **kwargs):
+            pass
+
+        def create_user(self, *, username: str, password: str, home_dir: str) -> None:
+            calls["create"] += 1
+            raise SFTPGoError("sftpgo http error: 409 Conflict")
+
+        def reset_password(self, *, username: str, new_password: str, home_dir: str) -> None:
+            calls["reset"].append((username, new_password))
+
+        def enable_user(self, *, username: str, home_dir: str) -> None:
+            calls["enable"].append(username)
+
+        def disable_user(self, *, username: str, home_dir: str) -> None:
+            return None
+
+    monkeypatch.setattr(app_mod, "SFTPGoAdminClient", _Client)
+
+    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))
+
+    admin_headers = {"authorization": "Bearer adm1"}
+    with TestClient(app) as c:
+        r = c.post("/api/v2/users", headers=admin_headers, json={"user_id": "bob"})
+        assert r.status_code == 200
+        body = r.json()
+        assert body["user_id"] == "bob"
+        assert body.get("sftp", {}).get("password")
+        assert calls["create"] == 1
+        assert calls["reset"] and calls["reset"][-1][0] == "bob"
+        assert calls["enable"] == ["bob"]
+
+
+def test_sftpgo_enabled_requires_admin_password_env(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+
+    cfg_path = _write_config(tmp_path, enabled=True)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "adm1")
+    monkeypatch.delenv("SFTPGO_ADMIN_PASSWORD", raising=False)
+
+    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))
+
+    admin_headers = {"authorization": "Bearer adm1"}
+    with TestClient(app) as c:
+        r = c.post("/api/v2/users", headers=admin_headers, json={"user_id": "alice"})
+        assert r.status_code == 500
+        assert "SFTPGO_ADMIN_PASSWORD" in r.text
+
+
+def test_sftpgo_admin_client_builds_requests(monkeypatch):
+    import json
+    from urllib.request import Request
+
+    from argus.service.sftpgo import SFTPGoAdminClient
+    import argus.service.sftpgo as mod
+
+    seen: list[Request] = []
+
+    class _Resp:
+        def __init__(self, body: bytes = b"ok"):
+            self._body = body
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def read(self):
+            return self._body
+
+    def fake_urlopen(req, timeout=0):
+        seen.append(req)
+        if req.full_url.endswith("/token"):
+            return _Resp(body=b'{"access_token":"t1","expires_at":0}')
+        # allow folder creates to be idempotent in tests
+        if req.full_url.endswith("/folders") and req.get_method() == "POST":
+            return _Resp(body=b'{"message":"created"}')
+        if req.full_url.endswith("/users/alice") and req.get_method() == "GET":
+            return _Resp(
+                body=b'{"username":"alice","status":1,"home_dir":"/private/users/alice","uid":0,"gid":0,"permissions":{"/":["*"]},"virtual_folders":[]}'
+            )
+        return _Resp()
+
+    monkeypatch.setattr(mod, "urlopen", fake_urlopen)
+
+    c = SFTPGoAdminClient(admin_api_base="http://sftpgo.local/api/v2", admin_user="admin", admin_password="pw", common_root="/private/common")
+    c.create_user(username="alice", password="p1", home_dir="/private/users/alice")
+    c.disable_user(username="alice", home_dir="/private/users/alice")
+    c.enable_user(username="alice", home_dir="/private/users/alice")
+    c.reset_password(username="alice", new_password="p2", home_dir="/private/users/alice")
+
+    # Each operation fetches a token, then issues a request.
+    # For create_user: token + ensure folders (2 POSTs) + create user
+    # For disable/enable/reset: token + ensure folders (2 POSTs) + GET user + PUT user
+    assert len(seen) == 1 + 2 + 1 + 3 * (1 + 2 + 1 + 1)
+    assert seen[0].full_url.endswith("/token")
+    assert seen[0].headers.get("Authorization", "").startswith("Basic ")
+    # create_user
+    assert seen[1].full_url.endswith("/folders")
+    assert seen[2].full_url.endswith("/folders")
+    assert seen[3].full_url.endswith("/users")
+    assert seen[3].headers.get("Authorization", "") == "Bearer t1"
+    created = json.loads(seen[3].data.decode("utf-8"))
+    assert created["username"] == "alice"
+    assert "/common" in created.get("permissions", {})
+    assert "/common/datasets" in created.get("permissions", {})
+    assert "/common/hf" in created.get("permissions", {})
+
+    # disable_user
+    assert seen[4].full_url.endswith("/token")
+    assert seen[5].full_url.endswith("/folders")
+    assert seen[6].full_url.endswith("/folders")
+    assert seen[7].full_url.endswith("/users/alice")
+    assert seen[7].get_method() == "GET"
+    assert seen[8].full_url.endswith("/users/alice")
+    assert seen[8].get_method() == "PUT"
+
+    # enable_user
+    assert seen[9].full_url.endswith("/token")
+    assert seen[10].full_url.endswith("/folders")
+    assert seen[11].full_url.endswith("/folders")
+    assert seen[12].full_url.endswith("/users/alice")
+    assert seen[12].get_method() == "GET"
+    assert seen[13].full_url.endswith("/users/alice")
+    assert seen[13].get_method() == "PUT"
+
+    # reset_password
+    assert seen[14].full_url.endswith("/token")
+    assert seen[15].full_url.endswith("/folders")
+    assert seen[16].full_url.endswith("/folders")
+    assert seen[17].full_url.endswith("/users/alice")
+    assert seen[17].get_method() == "GET"
+    assert seen[18].full_url.endswith("/users/alice")
+    assert seen[18].get_method() == "PUT"
+
+
+def test_sftpgo_admin_client_http_error_raises(monkeypatch):
+    from urllib.error import HTTPError
+
+    from argus.service.sftpgo import SFTPGoAdminClient, SFTPGoError
+    import argus.service.sftpgo as mod
+
+    def fake_urlopen(req, timeout=0):
+        if req.full_url.endswith("/token"):
+            class _Resp:
+                def __enter__(self):
+                    return self
+
+                def __exit__(self, exc_type, exc, tb):
+                    return False
+
+                def read(self):
+                    return b'{"access_token":"t1","expires_at":0}'
+
+            return _Resp()
+        raise HTTPError(req.full_url, 500, "boom", hdrs=None, fp=None)
+
+    monkeypatch.setattr(mod, "urlopen", fake_urlopen)
+
+    c = SFTPGoAdminClient(admin_api_base="http://sftpgo.local/api/v2", admin_user="admin", admin_password="pw", common_root="/private/common")
+    try:
+        c.create_user(username="alice", password="p1", home_dir="/private/users/alice")
+        assert False, "expected SFTPGoError"
+    except SFTPGoError as e:
+        assert "http error" in str(e)
+
+
+def test_sftpgo_admin_client_url_error_raises(monkeypatch):
+    from urllib.error import URLError
+
+    from argus.service.sftpgo import SFTPGoAdminClient, SFTPGoError
+    import argus.service.sftpgo as mod
+
+    def fake_urlopen(req, timeout=0):
+        if req.full_url.endswith("/token"):
+            class _Resp:
+                def __enter__(self):
+                    return self
+
+                def __exit__(self, exc_type, exc, tb):
+                    return False
+
+                def read(self):
+                    return b'{"access_token":"t1","expires_at":0}'
+
+            return _Resp()
+        raise URLError("no route")
+
+    monkeypatch.setattr(mod, "urlopen", fake_urlopen)
+
+    c = SFTPGoAdminClient(admin_api_base="http://sftpgo.local/api/v2", admin_user="admin", admin_password="pw")
+    try:
+        c.disable_user(username="alice", home_dir="/private/users/alice")
+        assert False, "expected SFTPGoError"
+    except SFTPGoError as e:
+        assert "connection error" in str(e)
--- a/src/mvp/py/tests/test_ui.py
+++ b/src/mvp/py/tests/test_ui.py
@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from argus.service.app import create_app
+
+
+def _write_config(tmp_path: Path) -> Path:
+    p = tmp_path / "cfg.yaml"
+    p.write_text(
+        """
+ray:
+  address: "http://127.0.0.1:8265"
+  shared_root: "/private"
+  entrypoint_num_cpus: 1
+  entrypoint_resources: { worker_node: 1 }
+  runtime_env: { env_vars: { PYTHONUNBUFFERED: "1" } }
+service:
+  api: { host: "127.0.0.1", port: 8080 }
+  auth: { token_env: "MVP_INTERNAL_TOKEN" }
+  sqlite: { db_path: "%(db)s" }
+data:
+  user_root: "%(users)s"
+  sftpgo: { enabled: false }
+  retention: { jobs_trash_after_days: 3, jobs_purge_after_days: 7, janitor_interval_s: 3600 }
+"""
+        % {"db": str(tmp_path / "mvp.sqlite3"), "users": str(tmp_path / "users")}
+    )
+    return p
+
+
+def test_ui_routes_render_200(tmp_path, monkeypatch):
+    cfg = _write_config(tmp_path)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "admin-token")
+    app = create_app(str(cfg))
+    c = TestClient(app)
+
+    for path in (
+        "/ui",
+        "/ui/login",
+        "/ui/tasks",
+        "/ui/tasks/new",
+        "/ui/data",
+        "/ui/admin",
+        "/ui/tasks/any-task-id",
+        "/ui/tasks/any-task-id/logs",
+    ):
+        r = c.get(path, allow_redirects=True)
+        assert r.status_code == 200
+        assert "<html" in r.text.lower()
+
+
+def test_ui_contains_sidebar_links(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")
+    assert r.status_code == 200
+    for link in ("/ui/tasks", "/ui/tasks/new", "/ui/data", "/ui/login", "/ui/admin"):
+        assert link in r.text
+    assert "Ray Dashboard" in r.text
+
+
+def test_ui_task_detail_shows_ids(tmp_path, monkeypatch):
+    cfg = _write_config(tmp_path)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "admin-token")
+    app = create_app(str(cfg))
+    c = TestClient(app)
+
+    task_id = "mvp3-ppo-20250101-000000-aaaa"
+    r = c.get(f"/ui/tasks/{task_id}")
+    assert r.status_code == 200
+    assert task_id in r.text
+    assert f"/ui/tasks/{task_id}/logs" in r.text
--- a/src/mvp/py/tests/test_users.py
+++ b/src/mvp/py/tests/test_users.py
@@ -14,6 +14,11 @@ def _write_config(tmp_path: Path) -> Path:
            "entrypoint_resources": {"worker_node": 1},
            "runtime_env": {"env_vars": {}},
        },
+        "data": {
+            # Avoid touching real /private in tests. Keep ray.shared_root as /private
+            # so existing path validation tests remain unchanged.
+            "user_root": str(tmp_path / "users"),
+        },
        "service": {
            "api": {"host": "127.0.0.1", "port": 0},
            "auth": {"token_env": "MVP_INTERNAL_TOKEN"},
@@ -59,6 +64,9 @@ def test_admin_create_user_issue_token_and_disabled_rejected(tmp_path: Path, mon

    admin_headers = {"authorization": "Bearer adm1"}
    with TestClient(app) as c:
+        # list users requires admin
+        assert c.get("/api/v2/users", headers={"authorization": "Bearer nope"}).status_code in (401, 403)
+
        r1 = c.post("/api/v2/users", headers=admin_headers, json={"user_id": "alice", "display_name": "Alice"})
        assert r1.status_code == 200
        assert r1.json()["user_id"] == "alice"
@@ -68,6 +76,11 @@ def test_admin_create_user_issue_token_and_disabled_rejected(tmp_path: Path, mon
        token = r2.json()["token"]
        assert token

+        r2b = c.get("/api/v2/users", headers=admin_headers)
+        assert r2b.status_code == 200
+        users = r2b.json()["users"]
+        assert any(u.get("user_id") == "alice" for u in users)
+
        # non-admin token can access regular endpoints
        r3 = c.get("/api/v2/queue", headers={"authorization": f"Bearer {token}"})
        assert r3.status_code == 200
@@ -177,3 +190,165 @@ def test_submit_rejects_non_common_inputs(tmp_path: Path, monkeypatch):
        )
        assert r.status_code == 400
        assert "code_path must start with /private/common/" in r.text
+
+
+def test_submit_accepts_user_dataset_paths_and_local_model_paths(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+
+    cfg_path = _write_config(tmp_path)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "adm1")
+
+    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))
+
+    admin_headers = {"authorization": "Bearer adm1"}
+    with TestClient(app) as c:
+        assert c.post("/api/v2/users", headers=admin_headers, json={"user_id": "alice"}).status_code == 200
+        alice_tok = c.post("/api/v2/users/alice/tokens", headers=admin_headers).json()["token"]
+        alice_headers = {"authorization": f"Bearer {alice_tok}"}
+
+        # User dataset paths are allowed.
+        r1 = c.post(
+            "/api/v2/tasks",
+            headers=alice_headers,
+            data=(
+                "workload: ppo\n"
+                "code_path: /private/common/code/verl\n"
+                "model_id: Qwen/Qwen2.5-0.5B-Instruct\n"
+                "train_file: /private/users/alice/datasets/t\n"
+            ),
+        )
+        assert r1.status_code == 200
+
+        # Local model paths under user models/ are allowed (no TaskSpec schema change).
+        r2 = c.post(
+            "/api/v2/tasks",
+            headers=alice_headers,
+            data=(
+                "workload: ppo\n"
+                "code_path: /private/common/code/verl\n"
+                "model_id: /private/users/alice/models/m1\n"
+                "train_file: /private/common/datasets/t\n"
+            ),
+        )
+        assert r2.status_code == 200
+
+
+def test_submit_rejects_cross_user_paths_and_bad_local_model_dirs(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+
+    cfg_path = _write_config(tmp_path)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "adm1")
+
+    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))
+
+    admin_headers = {"authorization": "Bearer adm1"}
+    with TestClient(app) as c:
+        assert c.post("/api/v2/users", headers=admin_headers, json={"user_id": "alice"}).status_code == 200
+        assert c.post("/api/v2/users", headers=admin_headers, json={"user_id": "bob"}).status_code == 200
+        alice_tok = c.post("/api/v2/users/alice/tokens", headers=admin_headers).json()["token"]
+        bob_tok = c.post("/api/v2/users/bob/tokens", headers=admin_headers).json()["token"]
+        bob_headers = {"authorization": f"Bearer {bob_tok}"}
+
+        # Cross-user dataset path should be rejected.
+        r1 = c.post(
+            "/api/v2/tasks",
+            headers=bob_headers,
+            data=(
+                "workload: ppo\n"
+                "code_path: /private/common/code/verl\n"
+                "model_id: Qwen/Qwen2.5-0.5B-Instruct\n"
+                "train_file: /private/users/alice/datasets/t\n"
+            ),
+        )
+        assert r1.status_code == 400
+        assert "/private/users/bob/datasets/" in r1.text
+
+        # Local model path must be under models/.
+        r2 = c.post(
+            "/api/v2/tasks",
+            headers=bob_headers,
+            data=(
+                "workload: ppo\n"
+                "code_path: /private/common/code/verl\n"
+                "model_id: /private/users/bob/jobs/j1/checkpoints\n"
+                "train_file: /private/common/datasets/t\n"
+            ),
+        )
+        assert r2.status_code == 400
+        assert "model_id local path must start with" in r2.text
+
+
+def test_me_returns_paths_and_retention(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+
+    cfg_path = _write_config(tmp_path)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "adm1")
+
+    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))
+
+    admin_headers = {"authorization": "Bearer adm1"}
+    with TestClient(app) as c:
+        assert c.post("/api/v2/users", headers=admin_headers, json={"user_id": "alice"}).status_code == 200
+        alice_tok = c.post("/api/v2/users/alice/tokens", headers=admin_headers).json()["token"]
+
+        r = c.get("/api/v2/me", headers={"authorization": f"Bearer {alice_tok}"})
+        assert r.status_code == 200
+        obj = r.json()
+        assert obj["user_id"] == "alice"
+        assert obj["paths"]["home"].endswith("/users/alice")
+        assert obj["paths"]["jobs"].endswith("/users/alice/jobs")
+        assert obj["paths"]["trash_jobs"].endswith("/users/alice/trash/jobs")
+        assert obj["retention"]["jobs_trash_after_days"] == 3
+        assert obj["retention"]["jobs_purge_after_days"] == 7
+
+
+def test_create_user_creates_user_dirs(tmp_path: Path, monkeypatch):
+    from argus.service import app as app_mod
+
+    cfg_path = _write_config(tmp_path)
+    monkeypatch.setenv("MVP_INTERNAL_TOKEN", "adm1")
+
+    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))
+
+    admin_headers = {"authorization": "Bearer adm1"}
+    with TestClient(app) as c:
+        assert c.post("/api/v2/users", headers=admin_headers, json={"user_id": "alice"}).status_code == 200
+
+    base = tmp_path / "users" / "alice"
+    assert (base / "datasets").is_dir()
+    assert (base / "models").is_dir()
+    assert (base / "code").is_dir()
+    assert (base / "jobs").is_dir()
+    assert (base / "trash" / "jobs").is_dir()
--- a/src/mvp/scripts/12_install_api_deps.sh
+++ b/src/mvp/scripts/12_install_api_deps.sh
@@ -9,4 +9,4 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 source "${SCRIPT_DIR}/lib.sh"

 dexec "${HEAD_CONTAINER}" bash -lc "python3 -m pip install -U pip >/dev/null 2>&1 || true"
-dexec "${HEAD_CONTAINER}" bash -lc "python3 -m pip install -r /workspace/mvp/py/requirements.txt"
+dexec "${HEAD_CONTAINER}" bash -lc "python3 -c 'import fastapi,uvicorn,yaml' >/dev/null 2>&1 && echo 'api_deps_ok: skip' || (python3 -m pip install -r /workspace/mvp/py/requirements.txt || echo 'WARN: api deps install failed; continuing with preinstalled deps')"
--- a/src/mvp/scripts/60_start_api.sh
+++ b/src/mvp/scripts/60_start_api.sh
@@ -22,7 +22,12 @@ if [[ -z "${MVP_INTERNAL_TOKEN:-}" ]]; then
  exit 1
 fi

-docker exec -d -e MVP_INTERNAL_TOKEN="${MVP_INTERNAL_TOKEN}" "${HEAD_CONTAINER}" bash -lc "nohup python3 /workspace/mvp/py/server.py --config '${CONFIG_IN_CONTAINER}' >>'${LOG_PATH}' 2>&1 & echo \$! >'${PID_PATH}'"
+env_args=(-e "MVP_INTERNAL_TOKEN=${MVP_INTERNAL_TOKEN}")
+if [[ -n "${SFTPGO_ADMIN_PASSWORD:-}" ]]; then
+  env_args+=(-e "SFTPGO_ADMIN_PASSWORD=${SFTPGO_ADMIN_PASSWORD}")
+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}'"

 echo "[host] started; pid stored in ${PID_PATH} (container path)"
 echo "[host] logs: ${LOG_PATH} (container path)"
--- a/src/mvp/scripts/run_all_v30_api.sh
+++ b/src/mvp/scripts/run_all_v30_api.sh
@@ -0,0 +1,242 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck source=lib.sh
+source "${SCRIPT_DIR}/lib.sh"
+
+# E2E v3.0:
+# - Start Ray (head + stateless workers) + SFTPGo (compose)
+# - Start API server with v3.0 config
+# - Create user (API triggers SFTPGo user create) and return one-time SFTP password
+# - (Optional) verify SFTP login
+# - Submit PPO/GRPO/SFT referencing user dataset paths and wait
+
+API_ADDR="${API_ADDR:-http://127.0.0.1:8080}"
+ADMIN_TOKEN="${MVP_INTERNAL_TOKEN:-}"
+USER_ID="${USER_ID:-alice}"
+RESET_DB="${RESET_DB:-1}"
+RESET_SFTPGO="${RESET_SFTPGO:-0}"
+EXPECTED_RAY_NODES="${EXPECTED_RAY_NODES:-3}" # head + 2 workers
+CLUSTER_NAME="${CLUSTER_NAME:-argus-ray}"
+
+CONFIG_IN_CONTAINER="${CONFIG_IN_CONTAINER:-/workspace/mvp/configs/dev_v30.yaml}"
+SFTPGO_ADMIN_PASSWORD="${SFTPGO_ADMIN_PASSWORD:-my-dev-sftpgo-admin}"
+export SFTPGO_ADMIN_PASSWORD
+
+if [[ -z "${ADMIN_TOKEN}" ]]; then
+  echo "ERROR: MVP_INTERNAL_TOKEN must be set in host env (admin token)" >&2
+  exit 1
+fi
+
+api_curl_admin() {
+  curl -sS -H "Authorization: Bearer ${ADMIN_TOKEN}" "$@"
+}
+
+api_wait_ready() {
+  local tries="${1:-60}"
+  for i in $(seq 1 "${tries}"); do
+    if curl -sS -m 2 "${API_ADDR}/docs" >/dev/null 2>&1; then
+      echo "[host] api_ready: ${API_ADDR}"
+      return 0
+    fi
+    echo "[host] waiting api... (${i}/${tries})"
+    sleep 2
+  done
+  echo "ERROR: api not ready: ${API_ADDR}" >&2
+  return 1
+}
+
+sftpgo_wait_ready() {
+  local tries="${1:-60}"
+  local url="${2:-http://127.0.0.1:8081/api/v2/token}"
+  for i in $(seq 1 "${tries}"); do
+    # SFTPGo admin endpoints require auth; readiness means "HTTP reachable and can issue token".
+    if curl -sS -m 2 -u "admin:${SFTPGO_ADMIN_PASSWORD}" "${url}" >/dev/null 2>&1; then
+      echo "[host] sftpgo_ready: ${url} (token ok)"
+      return 0
+    fi
+    echo "[host] waiting sftpgo... (${i}/${tries})"
+    sleep 2
+  done
+  echo "ERROR: sftpgo not ready: ${url}" >&2
+  return 1
+}
+
+ray_wait_ready() {
+  local tries="${1:-60}"
+  for i in $(seq 1 "${tries}"); do
+    if curl -sS -m 2 "${RAY_DASHBOARD_ADDR}/api/version" >/dev/null 2>&1; then
+      echo "[host] ray_dashboard_ready: ${RAY_DASHBOARD_ADDR}"
+      return 0
+    fi
+    echo "[host] waiting ray dashboard... (${i}/${tries})"
+    sleep 2
+  done
+  echo "ERROR: ray dashboard not ready: ${RAY_DASHBOARD_ADDR}" >&2
+  return 1
+}
+
+ray_wait_nodes() {
+  local want="${1:-3}"
+  local tries="${2:-60}"
+  for i in $(seq 1 "${tries}"); do
+    local out n
+    out="$(docker exec -i "${HEAD_CONTAINER}" python3 -c "import ray; ray.init(address='auto', ignore_reinit_error=True, log_to_driver=False, logging_level='ERROR'); print(sum(1 for n in ray.nodes() if n.get('Alive')))" 2>/dev/null || true)"
+    n="$(printf '%s\n' "${out}" | tail -n 1 | tr -cd '0-9' || true)"
+    if [[ "${n}" =~ ^[0-9]+$ ]]; then
+      echo "[host] ray_nodes_alive=${n} (want>=${want})"
+      if [[ "${n}" -ge "${want}" ]]; then
+        return 0
+      fi
+    else
+      echo "[host] waiting ray nodes... (${i}/${tries})"
+    fi
+    sleep 2
+  done
+  echo "ERROR: ray nodes not ready (want>=${want})" >&2
+  docker exec -i "${HEAD_CONTAINER}" bash -lc "ray status || true" >&2 || true
+  return 1
+}
+
+submit_taskspec_inline() {
+  local token="$1"
+  local yaml_body="$2"
+  local resp
+  resp="$(curl -sS -H "Authorization: Bearer ${token}" -H "Content-Type: application/yaml" --data-binary "${yaml_body}" "${API_ADDR}/api/v2/tasks")"
+  echo "[host] submit_resp: ${resp}" >&2
+  printf '%s' "${resp}" | python3 -c 'import sys,json; print(json.load(sys.stdin)["task_id"])'
+}
+
+wait_task() {
+  local token="$1"
+  local task_id="$2"
+  while true; do
+    local body state
+    body="$(curl -sS -H "Authorization: Bearer ${token}" "${API_ADDR}/api/v2/tasks/${task_id}")"
+    state="$(printf '%s' "${body}" | python3 -c 'import sys,json; print(json.load(sys.stdin)["state"])')"
+    echo "[host] task ${task_id}: ${state}"
+
+    if [[ "${state}" == "SUCCEEDED" ]]; then
+      return 0
+    fi
+    if [[ "${state}" == "FAILED" || "${state}" == "CANCELED" ]]; then
+      echo "[host] terminal=${state}; tail logs (best-effort):" >&2
+      curl -sS -H "Authorization: Bearer ${token}" "${API_ADDR}/api/v2/tasks/${task_id}/logs?tail=200" >&2 || true
+      return 1
+    fi
+    sleep 10
+  done
+}
+
+echo "[host] ===== run_all_v30_api.sh begin ====="
+
+"${SCRIPT_DIR}/00_prereq_check.sh"
+"${SCRIPT_DIR}/03_cleanup_v1_legacy.sh"
+"${SCRIPT_DIR}/04_cleanup_v2_legacy.sh"
+
+echo "[host] bring down existing containers (best-effort)"
+"${SCRIPT_DIR}/02_down.sh" || true
+
+if [[ "${RESET_SFTPGO}" == "1" ]]; then
+  echo "[host] reset sftpgo metadata dir (best-effort, via helper container)"
+  SFTPGO_META_DIR="${ROOT_DIR}/../../shared/common/sftpgo"
+  mkdir -p "${SFTPGO_META_DIR}"
+  docker run --rm --entrypoint sh -u 0:0 -v "${SFTPGO_META_DIR}:/mnt" drakkan/sftpgo:latest -lc "rm -rf /mnt/* || true"
+fi
+
+echo "[host] (re)create containers (Ray + SFTPGo)"
+"${SCRIPT_DIR}/01_up.sh"
+
+echo "[host] wait ray head ready"
+ray_wait_ready 60
+
+echo "[host] wait sftpgo ready"
+sftpgo_wait_ready 60 "http://127.0.0.1:8081/api/v2/token"
+
+echo "[host] render v3.0 config with SFTPGo container IP (work around docker DNS issues)"
+SFTPGO_IP="$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' argus-sftpgo)"
+RENDERED_CFG_HOST_PATH="/tmp/dev_v30_rendered.yaml"
+sed -E "s#^(\\s*admin_api_base:) .*#\\1 \"http://${SFTPGO_IP}:8080/api/v2\"#g" "${ROOT_DIR}/configs/dev_v30.yaml" >"${RENDERED_CFG_HOST_PATH}"
+docker cp "${RENDERED_CFG_HOST_PATH}" "${HEAD_CONTAINER}:/tmp/dev_v30_rendered.yaml"
+CONFIG_IN_CONTAINER="/tmp/dev_v30_rendered.yaml"
+
+echo "[host] verify head discovery record (supervised in-container)"
+HEAD_IP_FILE="${SHARED_ROOT}/ray/discovery/${CLUSTER_NAME}/head.json"
+dexec "${HEAD_CONTAINER}" bash -lc "test -f '${HEAD_IP_FILE}' && python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))[\"job_server_url\"])' '${HEAD_IP_FILE}' || true"
+
+echo "[host] wait workers join"
+ray_wait_nodes "${EXPECTED_RAY_NODES}" 120
+
+echo "[host] prepare data/model (idempotent; reuse cache)"
+"${SCRIPT_DIR}/30_prepare_data_and_model.sh"
+
+echo "[host] install api deps in head container"
+"${SCRIPT_DIR}/12_install_api_deps.sh"
+
+echo "[host] stop api (best-effort)"
+"${SCRIPT_DIR}/61_stop_api.sh" || true
+
+if [[ "${RESET_DB}" == "1" ]]; then
+  echo "[host] reset api sqlite db in container (best-effort)"
+  docker exec -i "${HEAD_CONTAINER}" bash -lc "rm -f /private/common/db/mvp.sqlite3 /private/common/db/mvp.sqlite3-wal /private/common/db/mvp.sqlite3-shm || true"
+else
+  echo "[host] keep existing api sqlite db (RESET_DB=${RESET_DB})"
+fi
+
+echo "[host] start api (admin token + sftpgo admin password via env)"
+MVP_INTERNAL_TOKEN="${ADMIN_TOKEN}" CONFIG_IN_CONTAINER="${CONFIG_IN_CONTAINER}" SFTPGO_ADMIN_PASSWORD="${SFTPGO_ADMIN_PASSWORD}" "${SCRIPT_DIR}/60_start_api.sh"
+api_wait_ready 60
+
+echo "[host] create user (expect SFTP one-time password in response)"
+create_resp="$(api_curl_admin -H "Content-Type: application/json" -d "{\"user_id\":\"${USER_ID}\"}" "${API_ADDR}/api/v2/users")"
+echo "[host] create_user_resp: ${create_resp}"
+USER_SFTP_PASSWORD="$(printf '%s' "${create_resp}" | python3 -c 'import sys,json; o=json.load(sys.stdin); print((o.get("sftp") or {}).get("password") or "")')"
+if [[ -z "${USER_SFTP_PASSWORD}" ]]; then
+  echo "ERROR: expected sftp.password in create user response (is data.sftpgo.enabled=true?)" >&2
+  exit 1
+fi
+
+echo "[host] issue user token"
+USER_TOKEN="$(api_curl_admin -X POST "${API_ADDR}/api/v2/users/${USER_ID}/tokens" | python3 -c 'import sys,json; print(json.load(sys.stdin)["token"])')"
+echo "[host] user_token_issued: user=${USER_ID}"
+
+echo "[host] (optional) verify SFTP login (best-effort)"
+if command -v sshpass >/dev/null 2>&1 && command -v sftp >/dev/null 2>&1; then
+  tmp_batch="$(mktemp)"
+  cat >"${tmp_batch}" <<EOF
+pwd
+ls
+EOF
+  # NOTE: this just proves auth works; dataset upload can be done via SFTP manually later.
+  sshpass -p "${USER_SFTP_PASSWORD}" sftp -o StrictHostKeyChecking=no -P 2022 "${USER_ID}@127.0.0.1" -b "${tmp_batch}" >/dev/null 2>&1 || true
+  rm -f "${tmp_batch}" || true
+else
+  echo "[host] skip: sshpass/sftp not found; you can test manually with: sftp -P 2022 ${USER_ID}@<host>"
+fi
+
+echo "[host] ensure user dataset paths exist (copy from common if needed; best-effort)"
+echo "[host] copy dataset into user datasets path inside head container (avoid host permission issues)"
+dexec "${HEAD_CONTAINER}" bash -lc "set -euo pipefail; \
+  mkdir -p '/private/users/${USER_ID}/datasets/gsm8k' '/private/users/${USER_ID}/datasets/gsm8k_sft'; \
+  (cp -f /private/common/datasets/gsm8k/train.parquet '/private/users/${USER_ID}/datasets/gsm8k/train.parquet' 2>/dev/null || cp -f /private/datasets/gsm8k/train.parquet '/private/users/${USER_ID}/datasets/gsm8k/train.parquet' 2>/dev/null || true); \
+  (cp -f /private/common/datasets/gsm8k/test.parquet  '/private/users/${USER_ID}/datasets/gsm8k/test.parquet'  2>/dev/null || cp -f /private/datasets/gsm8k/test.parquet  '/private/users/${USER_ID}/datasets/gsm8k/test.parquet'  2>/dev/null || true); \
+  (cp -f /private/common/datasets/gsm8k_sft/train.parquet '/private/users/${USER_ID}/datasets/gsm8k_sft/train.parquet' 2>/dev/null || cp -f /private/datasets/gsm8k_sft/train.parquet '/private/users/${USER_ID}/datasets/gsm8k_sft/train.parquet' 2>/dev/null || true); \
+  (cp -f /private/common/datasets/gsm8k_sft/test.parquet  '/private/users/${USER_ID}/datasets/gsm8k_sft/test.parquet'  2>/dev/null || cp -f /private/datasets/gsm8k_sft/test.parquet  '/private/users/${USER_ID}/datasets/gsm8k_sft/test.parquet'  2>/dev/null || true)"
+
+echo "[host] submit PPO/GRPO/SFT via API using user dataset paths"
+PPO_TASK_ID="$(submit_taskspec_inline "${USER_TOKEN}" $'workload: ppo\nnnodes: 2\nn_gpus_per_node: 4\ncode_path: /private/common/code/verl/verl_repo\ntrain_file: /private/users/'"${USER_ID}"$'/datasets/gsm8k/train.parquet\nval_file: /private/users/'"${USER_ID}"$'/datasets/gsm8k/test.parquet\nmodel_id: Qwen/Qwen2.5-0.5B-Instruct\ntotal_epochs: 1\ntotal_training_steps: 10\nsave_freq: 10\n')"
+GRPO_TASK_ID="$(submit_taskspec_inline "${USER_TOKEN}" $'workload: grpo\nnnodes: 2\nn_gpus_per_node: 4\ncode_path: /private/common/code/verl/verl_repo\ntrain_file: /private/users/'"${USER_ID}"$'/datasets/gsm8k/train.parquet\nval_file: /private/users/'"${USER_ID}"$'/datasets/gsm8k/test.parquet\nmodel_id: Qwen/Qwen2.5-0.5B-Instruct\ntotal_epochs: 1\ntotal_training_steps: 10\nsave_freq: 10\n')"
+SFT_TASK_ID="$(submit_taskspec_inline "${USER_TOKEN}" $'workload: sft\nnnodes: 1\nn_gpus_per_node: 1\ncode_path: /private/common/code/verl/verl_repo\ntrain_file: /private/users/'"${USER_ID}"$'/datasets/gsm8k_sft/train.parquet\nval_file: /private/users/'"${USER_ID}"$'/datasets/gsm8k_sft/test.parquet\nmodel_id: Qwen/Qwen2.5-0.5B-Instruct\ntotal_epochs: 1\ntotal_training_steps: 10\nsave_freq: 10\n')"
+
+echo "[host] submitted task ids:"
+echo "  ppo=${PPO_TASK_ID}"
+echo "  grpo=${GRPO_TASK_ID}"
+echo "  sft=${SFT_TASK_ID}"
+
+echo "[host] wait for tasks (in submission order)"
+wait_task "${USER_TOKEN}" "${PPO_TASK_ID}"
+wait_task "${USER_TOKEN}" "${GRPO_TASK_ID}"
+wait_task "${USER_TOKEN}" "${SFT_TASK_ID}"
+
+echo "[host] ===== run_all_v30_api.sh done ====="
--- a/src/mvp/scripts/run_e2e_v30_cases.sh
+++ b/src/mvp/scripts/run_e2e_v30_cases.sh
@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+# Minimal E2E runner for v3.0:
+# - Happy path: Ray + SFTPGo + API + 3 workloads
+# - Leaves retention validation as a manual follow-up (adjust thresholds).
+
+ADMIN_TOKEN="${MVP_INTERNAL_TOKEN:-my-dev-token}"
+USER_ID="${USER_ID:-alice}"
+
+echo "[case] HP-1: run_all_v30_api.sh (Ray + SFTPGo + API + 3 workloads)"
+MVP_INTERNAL_TOKEN="${ADMIN_TOKEN}" USER_ID="${USER_ID}" RESET_DB=1 RESET_SFTPGO=0 "${SCRIPT_DIR}/run_all_v30_api.sh"
+
+echo "[case] NOTE: retention validation (C2) is manual:"
+echo "  - set data.retention.jobs_trash_after_days / jobs_purge_after_days to small values in configs/dev_v30.yaml"
+echo "  - restart API server and wait for janitor"
+