S26-E0.1: efficiency evaluation (bare vs yeto, verl and Miles)

tests/test_rl_miles_adapter_config.py

@@ -795,3 +795,21 @@ def test_dynamic_batching_refused_for_bshd_gated_delta_net_models():
     recompute_only = dataclasses.replace(
         cfg, model_recipe=recipe, train_memory=rc.TrainMemoryConfig(activation_recompute=True))
     assert "--recompute-granularity" in mc.translate_run_config(recompute_only, AlgorithmSpec()).argv
+
+
+def test_gradient_accumulation_fusion_is_opt_in():
+    # S26 E0.1: the Miles recipes train with the fusion on; yeto keeps it off by default
+    off = " ".join(mc.translate_run_config(make_config(), AlgorithmSpec()).argv)
+    assert "--no-gradient-accumulation-fusion" in off
+    cfg = dataclasses.replace(make_config(), train_memory=rc.TrainMemoryConfig(gradient_accumulation_fusion=True))
+    assert "--no-gradient-accumulation-fusion" not in mc.translate_run_config(cfg, AlgorithmSpec()).argv
+    assert rc._train_memory(SimpleNamespace(seq_len=8, rl_gradient_accumulation_fusion=True)) \
+        .gradient_accumulation_fusion
+
+
+def test_generic_recipe_trains_without_dropout():
+    cfg = make_config()
+    if cfg.model_recipe.name != rc.RECIPE_GENERIC:
+        cfg = sub(cfg, "model_recipe", name=rc.RECIPE_GENERIC)
+    joined = " ".join(mc.translate_run_config(cfg, AlgorithmSpec()).argv)
+    assert "--attention-dropout 0.0 --hidden-dropout 0.0" in joined

tests/test_rl_verl_adapter.py

@@ -447,3 +447,27 @@ def test_verl_data_cursor_refuses_what_it_cannot_honour():
         cur.seek({"sample_offset": 2})
     with pytest.raises(RuntimeError, match="twice"):
         VerlDataCursor(trainer)
+
+
+def test_fully_async_runtime_env_keeps_nccl_host_buffers_on_shm(monkeypatch):
+    # S26 E0.1: a cuMem host-buffer sender and a /dev/shm receiver cannot attach on Modal
+    import sys
+    import types
+
+    pkg = types.ModuleType("verl.trainer.constants_ppo")
+    pkg.get_ppo_ray_runtime_env = lambda _x: {"env_vars": {}}
+    for name in ("verl", "verl.trainer"):
+        monkeypatch.setitem(sys.modules, name, sys.modules.get(name) or types.ModuleType(name))
+    monkeypatch.setitem(sys.modules, "verl.trainer.constants_ppo", pkg)
+    ray_stub = types.ModuleType("ray")
+    ray_stub.remote = lambda *a, **k: (lambda cls: cls)
+    monkeypatch.setitem(sys.modules, "ray", sys.modules.get("ray") or ray_stub)
+    monkeypatch.setitem(sys.modules, "yeto.rl.adapters.verl.verl_main", None)
+    monkeypatch.delitem(sys.modules, "yeto.rl.adapters.verl.verl_main")
+    from yeto.rl.adapters.verl import verl_main
+
+    monkeypatch.delenv(verl_main.CUMEM_HOST_ENV, raising=False)
+    assert verl_main._runtime_env(fully_async=True)["env_vars"][verl_main.CUMEM_HOST_ENV] == "0"
+    assert verl_main.CUMEM_HOST_ENV not in verl_main._runtime_env()["env_vars"]
+    monkeypatch.setenv(verl_main.CUMEM_HOST_ENV, "1")
+    assert verl_main._runtime_env(fully_async=True)["env_vars"][verl_main.CUMEM_HOST_ENV] == "1"

tools/eval/__main__.py

@@ -5,6 +5,13 @@
     python -m tools.eval judge DIR/eval-e3-verl-a
     python -m tools.eval report DIR/* --out report.md [--html report.html]
 
+Efficiency evaluation (S26 E0.1, package :mod:`.efficiency`):
+
+    python -m tools.eval eff-plan qwen3-0.6b [--arm ARM]
+    python -m tools.eval eff-run qwen3-0.6b --arm bare-verl-sync --run-id eff-... [--out DIR] [--dry-run]
+    python -m tools.eval eff-parse DIR/eff-...
+    python -m tools.eval eff-report DIR/eff-* --out report.md
+
 ``run`` refuses a run id that is not pre-registered in the GPU ledger (the id
 must appear in ``gpu-spend.md``), checks the local thread count and Ray, writes
 ``cell.json``, runs the cell's harness (raw data into ``raw/``), then judges.
@@ -123,6 +130,71 @@ def cmd_report(args) -> int:
     return 0
 
 
+EFF_OUT = Path("/home/michael/work/infra-drafts/SESSION26/logs/E0")
+
+
+def cmd_eff_plan(args) -> int:
+    from .efficiency import spec
+
+    sp = spec.get_spec(args.preset)
+    arms = [args.arm] if args.arm else list(spec.ARMS)
+    doc = {"spec": sp.to_dict(), "fingerprint": spec.control_fingerprint(sp),
+           "arms": {a: spec.plan(sp, a, "<run id>") for a in arms},
+           "verl_knob_diff": {m: spec.verl_knob_diff(sp, m) for m in spec.MODES}}
+    print(json.dumps(doc, indent=1, default=str))
+    return 0
+
+
+def cmd_eff_run(args) -> int:
+    from .efficiency import modal_eff, spec
+
+    sp = spec.get_spec(args.preset)
+    plan = spec.plan(sp, args.arm, args.run_id)
+    run_dir = Path(args.out) / args.run_id
+    if not args.dry_run:
+        if (run_dir / "arm.json").exists():
+            raise SystemExit(f"{run_dir} already used")
+        _preflight(args.run_id)
+    plan["code_sha"] = subprocess.run(["git", "-C", str(Path(__file__).resolve().parents[2]), "rev-parse", "HEAD"],
+                                      capture_output=True, text=True).stdout.strip()
+    plan["images"] = spec.images()
+    spec.write_plan(run_dir / "arm.json", plan)
+    if plan["harness"] == "miles_local":
+        summary = modal_eff.run_miles_launcher(plan, run_dir, dry_run=args.dry_run)
+        if args.dry_run:
+            print((run_dir / "dry-run.log").read_text())
+            return summary["rc"]
+    elif args.dry_run:
+        print(plan.get("script") or json.dumps(plan.get("argv")))
+        return 0
+    else:
+        summary = modal_eff.run(plan, run_dir)
+    (run_dir / "run.json").write_text(json.dumps(summary, indent=1, default=str))
+    return cmd_eff_parse(argparse.Namespace(run_dir=str(run_dir)))
+
+
+def cmd_eff_parse(args) -> int:
+    from .efficiency import metrics, parse
+
+    run_dir = Path(args.run_dir)
+    parsed = parse.parse_run(run_dir)
+    (run_dir / "parsed.json").write_text(json.dumps(parsed, indent=1, default=str))
+    summary = metrics.summarize(parsed)
+    (run_dir / "summary.json").write_text(json.dumps(summary, indent=1, default=str))
+    print(json.dumps({"run": parsed["run_id"], "arm": parsed["arm"], "rounds": summary["rounds"],
+                      "missing": summary["missing"]}))
+    return 0 if summary["rounds"] == summary["rounds_expected"] else 1
+
+
+def cmd_eff_report(args) -> int:
+    from .efficiency import report as eff_report
+
+    runs = eff_report.load([Path(d) for d in args.run_dirs])
+    Path(args.out).write_text(eff_report.markdown(runs, title=args.title))
+    print(f"{len(runs)} parsed run(s) -> {args.out}")
+    return 0
+
+
 def main(argv: list[str] | None = None) -> int:
     p = argparse.ArgumentParser(prog="python -m tools.eval")
     sub = p.add_subparsers(dest="cmd", required=True)
@@ -143,6 +215,27 @@ def main(argv: list[str] | None = None) -> int:
     rep.add_argument("--html", default=None)
     rep.add_argument("--title", default="yeto RL capability evaluation")
     rep.set_defaults(fn=cmd_report)
+    from .efficiency.spec import ARMS, PRESETS
+
+    ep = sub.add_parser("eff-plan")
+    ep.add_argument("preset", choices=sorted(PRESETS))
+    ep.add_argument("--arm", choices=ARMS, default=None)
+    ep.set_defaults(fn=cmd_eff_plan)
+    er = sub.add_parser("eff-run")
+    er.add_argument("preset", choices=sorted(PRESETS))
+    er.add_argument("--arm", choices=ARMS, required=True)
+    er.add_argument("--run-id", required=True)
+    er.add_argument("--out", default=str(EFF_OUT))
+    er.add_argument("--dry-run", action="store_true")
+    er.set_defaults(fn=cmd_eff_run)
+    eparse = sub.add_parser("eff-parse")
+    eparse.add_argument("run_dir")
+    eparse.set_defaults(fn=cmd_eff_parse)
+    erep = sub.add_parser("eff-report")
+    erep.add_argument("run_dirs", nargs="+")
+    erep.add_argument("--out", required=True)
+    erep.add_argument("--title", default="yeto RL efficiency evaluation")
+    erep.set_defaults(fn=cmd_eff_report)
     args = p.parse_args(argv)
     return args.fn(args)
 

tools/eval/efficiency/__init__.py

@@ -0,0 +1,24 @@
+"""yeto RL efficiency evaluation (S26 E0.1 / E0.2).
+
+Eight arms per model preset: {bare, yeto} x {verl, miles} x {sync, async}. The
+bare arms run the framework's own entry point from the same image (no yeto
+code on their path); the yeto arms run the same framework through yeto. Every
+arm is generated from one :class:`.spec.EffSpec`, so model, data, order seed,
+reward, hyperparameters, max tokens, card type and count and image are equal
+by construction (:func:`.spec.control_fingerprint`).
+
+    python -m tools.eval eff-plan qwen3-0.6b [--arm ARM]
+    python -m tools.eval eff-run qwen3-0.6b --arm bare-verl-sync --run-id ID [--dry-run]
+    python -m tools.eval eff-parse RUN_DIR
+    python -m tools.eval eff-report RUN_DIR ... --out report.md
+
+* :mod:`.spec` -- presets, arms, the per-arm command plans;
+* :mod:`.parse` -- framework logs and yeto tapes -> one per-round record format;
+* :mod:`.metrics` -- per-run summary (throughput, segments, cost, off-policy);
+* :mod:`.report` -- Markdown report with yeto-vs-bare overhead and range verdicts;
+* :mod:`.modal_eff` -- the Modal harness (bare arms and yeto+verl);
+  yeto+Miles runs through the yeto launcher (``tools/eval/miles_local.sh``).
+
+Raw data is written to ``<run dir>/raw/`` before anything is computed;
+``parsed.json`` and the report are derived from it and can be rebuilt.
+"""

tools/eval/efficiency/algo-eff-miles.json

@@ -0,0 +1,53 @@
+{
+  "advantage": {
+    "estimator": "grpo",
+    "reward_binary": false,
+    "reward_postprocess": null,
+    "rewards_normalization": true,
+    "std_normalization": true,
+    "whiten": false
+  },
+  "correction": {
+    "batch_normalize": false,
+    "function": null,
+    "level": "token",
+    "lower": 0.0,
+    "method": "tis",
+    "mismatch_metrics": false,
+    "opsm_delta": null,
+    "ratio_mask": "none",
+    "upper": 2.0,
+    "use_rollout_logprobs": false,
+    "veto": null
+  },
+  "entropy_coef": 0.0,
+  "execution": {
+    "group_required": true,
+    "max_policy_staleness": 2,
+    "needs_critic": false,
+    "needs_rollout_logprobs": false
+  },
+  "kl": {
+    "coef": null,
+    "estimator": null,
+    "placement": "none",
+    "unbiased": false
+  },
+  "loss": {
+    "aggregation": "default",
+    "custom_loss": null,
+    "eps_clip": 0.2,
+    "eps_clip_c": null,
+    "eps_clip_high": 0.28,
+    "reducer": null,
+    "variant": "policy_loss"
+  },
+  "plugins": [],
+  "sampling": {
+    "filter": null,
+    "max_replacements": null,
+    "over_sampling_batch_size": null,
+    "overlong_filter": false
+  },
+  "schema": "yeto-rl-algorithm-spec-v2"
+}

tools/eval/efficiency/eff_prep.py

@@ -0,0 +1,65 @@
+"""Model and data preparation for the bare-framework arms (no yeto import).
+
+Run inside the container as ``python eff_prep.py <spec json>``. It downloads
+the pinned model snapshot and the pinned dataset revision, then writes the
+training prompts in the dataset's own row order:
+
+* ``/root/eff-data/train.parquet`` and ``test.parquet`` -- verl rows, the same
+  shape as ``yeto.rl.adapters.verl.island_entry.prepare_data`` (the yeto+verl
+  arm's data), so bare verl and yeto+verl read identical rows;
+* ``/root/eff-data/train.jsonl`` -- Miles rows (``prompt`` = chat messages,
+  ``label``), for ``--input-key prompt --label-key label --apply-chat-template``;
+* ``/root/eff-data/order.json`` -- sha256 of the label sequence (row order) and
+  the row count, so a report can show that two arms read the same rows.
+
+Shuffling is left to each framework (seeded by the spec's ``seed``), as in the
+yeto arm of the same framework.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+from pathlib import Path
+
+DATA_ROOT = Path("/root/eff-data")
+MODEL_ROOT = Path("/root/eff-models")
+VAL_ROWS = 64
+
+
+def main(spec_path: str) -> int:
+    import pandas as pd
+    from huggingface_hub import hf_hub_download, snapshot_download
+
+    spec = json.loads(Path(spec_path).read_text())
+    model_dir = MODEL_ROOT / spec["model"].split("/")[-1]
+    snapshot_download(spec["model"], revision=spec["model_revision"], local_dir=str(model_dir))
+    DATA_ROOT.mkdir(parents=True, exist_ok=True)
+    order = {}
+    for split in ("train", "test"):
+        src = hf_hub_download(spec["data"], f"{split}.parquet", repo_type="dataset", revision=spec["data_revision"])
+        frame = pd.read_parquet(src)
+        rows = [{
+            "data_source": "yeto/" + spec["data"],
+            "prompt": [{"role": m["role"], "content": m["content"]} for m in row["messages"]],
+            "ability": "math",
+            "reward_model": {"style": "rule", "ground_truth": str(row["label"])},
+            "extra_info": {"split": split, "index": int(i)},
+        } for i, row in frame.iterrows()]
+        if split == "test":
+            rows = rows[:VAL_ROWS]
+        pd.DataFrame(rows).to_parquet(DATA_ROOT / f"{split}.parquet")
+        labels = [r["reward_model"]["ground_truth"] for r in rows]
+        order[split] = {"rows": len(rows), "label_sha256": hashlib.sha256("\n".join(labels).encode()).hexdigest()}
+        if split == "train":
+            with (DATA_ROOT / "train.jsonl").open("w") as fh:
+                for r in rows:
+                    fh.write(json.dumps({"prompt": r["prompt"], "label": r["reward_model"]["ground_truth"]}) + "\n")
+    (DATA_ROOT / "order.json").write_text(json.dumps({**order, "model_dir": str(model_dir)}, indent=1))
+    print(json.dumps({"model_dir": str(model_dir), **order}))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1]))

tools/eval/efficiency/eff_reward.py

@@ -0,0 +1,39 @@
+"""GSM8K exact-number reward for the bare-framework arms (no yeto import).
+
+The grading rule is a verbatim copy of ``yeto.rl.rewards.builtin.gsm8k_grade``
+(the reward of the yeto arms, ``--reward-function gsm8k_reward:score``); a CPU
+test asserts both give the same value on the same cases. The file is copied
+into the bare containers as ``/root/eff_reward.py``:
+
+* verl: ``reward.custom_reward_function.path=/root/eff_reward.py`` and
+  ``reward.custom_reward_function.name=compute_score``;
+* Miles: ``--custom-rm-path eff_reward.score`` (``/root`` on ``PYTHONPATH``).
+"""
+
+from __future__ import annotations
+
+import re
+
+
+def _num(s):
+    s = str(s).replace(",", "")
+    m = re.findall(r"-?\d+(?:\.\d+)?", s)
+    return float(m[-1]) if m else None
+
+
+def grade(response: str, label) -> float:
+    gold = _num(str(label).split("####")[-1]) if label is not None else None
+    text = response or ""
+    boxed = re.findall(r"\\boxed\{([^}]*)\}", text)
+    pred = _num(boxed[-1]) if boxed else _num(text[-200:])
+    return 1.0 if (gold is not None and pred is not None and abs(pred - gold) < 1e-6) else 0.0
+
+
+def compute_score(data_source, solution_str, ground_truth, extra_info=None, **kwargs) -> float:
+    """verl custom reward signature."""
+    return grade(solution_str, ground_truth)
+
+
+async def score(args, sample, **kwargs) -> float:
+    """Miles custom reward signature (``--custom-rm-path``)."""
+    return grade(sample.response or "", sample.label)