rlf-codex-rl 5.8：--rl-training-attention-backend 与 --session-server-workers 两个训练侧 flag

tests/test_rl_training_attention_backend.py

@@ -0,0 +1,166 @@
+"""--rl-training-attention-backend override (rlf-codex-rl 5.8).
+
+The recipe picks the Megatron --attention-backend (qwen3_5 -> flash,
+generic -> unfused). The 5.8 gate run hit a flash-attn cute JIT compile
+failure on L40S, so the CLI gained an explicit override; flash-only
+recipes refuse a non-flash override.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+import types
+
+import pytest
+
+from yeto.rl.adapters.miles.island_entry import build_miles_argv
+
+
+def _args(**extra):
+    base = dict(
+        seq_len=128,
+        groups_per_round=4,
+        samples_per_group=2,
+        optimizer_steps=1,
+        lora_targets="auto",
+        lora_r=4,
+        seed=1,
+        learner_id=0,
+        global_rounds=3,
+        inner_lr=1e-4,
+        reward_function="pkg.reward:score",
+        over_sampling_batch_size=6,
+        rollout_max_response_len=64,
+        custom_generate_function_path="pkg.agent.generate",
+        custom_agent_function_path="pkg.agent.run",
+        agent_max_seq_len=96,
+        use_session_server=True,
+        session_server_ip="127.0.0.1",
+        session_server_port=[31000, 31002],
+        tito_model="qwen3",
+        tito_allowed_append_roles=["tool", "user"],
+        actor_num_nodes=1,
+        actor_num_gpus_per_node=8,
+        expert_parallel=None,
+        apply_chat_template_kwargs={"enable_thinking": False},
+    )
+    base.update(extra)
+    return argparse.Namespace(**base)
+
+
+def _provider(**extra):
+    base = dict(
+        hidden_size=16,
+        num_attention_heads=4,
+        num_layers=2,
+        ffn_hidden_size=32,
+        num_query_groups=2,
+        kv_channels=4,
+        seq_length=256,
+        vocab_size=64,
+        normalization="RMSNorm",
+        layernorm_epsilon=1e-6,
+        position_embedding_type="yarn",
+        rotary_base=10000,
+        gated_linear_unit=True,
+        share_embeddings_and_output_weights=False,
+        add_bias_linear=False,
+        add_qkv_bias=True,
+        qk_layernorm=True,
+        num_moe_experts=8,
+        moe_ffn_hidden_size=24,
+        moe_router_topk=2,
+        moe_layer_freq=[0, 1],
+        moe_shared_expert_intermediate_size=12,
+        multi_latent_attention=True,
+        q_lora_rank=8,
+        kv_lora_rank=4,
+        qk_head_dim=4,
+        qk_pos_emb_head_dim=2,
+        v_head_dim=4,
+    )
+    base.update(extra)
+    return argparse.Namespace(**base)
+
+
+@pytest.fixture(autouse=True)
+def _chat_template_stub(monkeypatch):
+    monkeypatch.setitem(
+        sys.modules,
+        "miles.utils.chat_template_utils",
+        types.SimpleNamespace(
+            resolve_reasoning_and_tool_call_parser=lambda model: ("qwen3", "qwen25")
+        ),
+    )
+
+
+def _backend_of(argv):
+    return argv[argv.index("--attention-backend") + 1]
+
+
+def test_gdn_recipe_default_stays_flash():
+    argv = build_miles_argv(
+        _args(),
+        model_path="/model",
+        prompt_path="/prompts.jsonl",
+        provider=_provider(experimental_attention_variant="gated_delta_net"),
+        target_modules=["qkv_proj", "out_proj"],
+    )
+    assert _backend_of(argv) == "flash"
+
+
+def test_override_unfused_on_gdn_recipe():
+    argv = build_miles_argv(
+        _args(rl_training_attention_backend="unfused"),
+        model_path="/model",
+        prompt_path="/prompts.jsonl",
+        provider=_provider(experimental_attention_variant="gated_delta_net"),
+        target_modules=["qkv_proj", "out_proj"],
+    )
+    assert _backend_of(argv) == "unfused"
+
+
+def test_override_fused_on_generic_recipe():
+    argv = build_miles_argv(
+        _args(rl_training_attention_backend="fused"),
+        model_path="/model",
+        prompt_path="/prompts.jsonl",
+        provider=_provider(),
+        target_modules=["qkv_proj", "out_proj"],
+    )
+    assert _backend_of(argv) == "fused"
+
+
+def test_auto_default_when_attr_missing():
+    # Old launch-side namespaces (and most tests) never set the attribute.
+    argv = build_miles_argv(
+        _args(),
+        model_path="/model",
+        prompt_path="/prompts.jsonl",
+        provider=_provider(),
+        target_modules=["qkv_proj", "out_proj"],
+    )
+    assert _backend_of(argv) == "unfused"  # generic recipe default
+
+
+def test_session_server_workers_forwarded_when_set():
+    argv = build_miles_argv(
+        _args(session_server_workers=8),
+        model_path="/model",
+        prompt_path="/prompts.jsonl",
+        provider=_provider(),
+        target_modules=["qkv_proj", "out_proj"],
+    )
+    assert argv[argv.index("--session-server-workers") + 1] == "8"
+
+
+def test_session_server_workers_absent_by_default():
+    argv = build_miles_argv(
+        _args(),
+        model_path="/model",
+        prompt_path="/prompts.jsonl",
+        provider=_provider(),
+        target_modules=["qkv_proj", "out_proj"],
+    )
+    assert "--session-server-workers" not in argv

tests/test_rlf_rl_gate.py

@@ -19,6 +19,7 @@
     GateEvidenceError,
     build_gate_rows,
     collect_policy_update,
+    collect_policy_update_from_tape,
     collect_rollout_samples,
     prepare_prompt_data,
     run_gate,
@@ -164,6 +165,61 @@ def test_real_collector_passing_run_announces_verdict(tmp_path):
         evidence["policy_update_evidence"]["after_digest"]
 
 
+# -- policy-update proof from the island event tape (代拍板 v2) -----------------
+
+def _tape(tmp_path, events):
+    path = tmp_path / "rl-island-0.jsonl"
+    path.write_text("".join(json.dumps(e) + "\n" for e in events), encoding="utf-8")
+    return path
+
+
+def _publication(version, digest):
+    return {"event": "rl_publication", "policy_version": version,
+            "sync/publication_payload_hash": digest}
+
+
+def test_tape_policy_update_detects_changed_payload(tmp_path):
+    tape = _tape(tmp_path, [
+        {"event": "rl_driver_start"},
+        _publication(0, "aaa"),
+        "not json at all",
+        _publication(1, "bbb"),
+        _publication(2, "ccc"),
+    ])
+    update = collect_policy_update_from_tape(tape, step=2)
+    assert update.updated is True
+    assert update.before_digest == "aaa" and update.after_digest == "ccc"
+    assert update.changed == ("policy_payload",)
+
+
+def test_tape_policy_update_unchanged_payload_marks_not_updated(tmp_path):
+    tape = _tape(tmp_path, [_publication(0, "aaa"), _publication(1, "aaa")])
+    update = collect_policy_update_from_tape(tape, step=1)
+    assert update.updated is False and update.changed == ()
+
+
+def test_tape_policy_update_fails_closed_on_single_version(tmp_path):
+    tape = _tape(tmp_path, [_publication(0, "aaa")])
+    with pytest.raises(GateEvidenceError, match="published policy version"):
+        collect_policy_update_from_tape(tape, step=1)
+
+
+def test_tape_policy_update_fails_closed_on_missing_hash(tmp_path):
+    tape = _tape(tmp_path, [_publication(0, "aaa"), {"event": "rl_publication",
+                                                    "policy_version": 1}])
+    with pytest.raises(GateEvidenceError, match="payload hash"):
+        collect_policy_update_from_tape(tape, step=1)
+
+
+def test_tape_policy_update_passes_full_gate(tmp_path):
+    samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups()))
+    tape = _tape(tmp_path, [_publication(0, "aaa"), _publication(1, "bbb")])
+    update = collect_policy_update_from_tape(tape, step=1)
+    rows = build_gate_rows(samples, update)
+    evidence = run_gate(rows, collector="miles_island", update=update)
+    assert evidence["verdict"] is True
+
+
 # -- evidence on disk: content and flag hygiene --------------------------------
 
 def test_write_evidence_scrubs_flags_and_keeps_detail(tmp_path):

yeto/cli.py

@@ -138,6 +138,14 @@ def _add_launch_args(p: argparse.ArgumentParser) -> None:
         default="generic",
         help="model-specific Miles memory/kernel contract",
     )
+    rl.add_argument(
+        "--rl-training-attention-backend",
+        choices=["auto", "flash", "fused", "unfused"],
+        default="auto",
+        help="override the recipe's Megatron --attention-backend (auto = recipe "
+        "decides); use unfused when the flash-attn cute JIT cannot compile on "
+        "the target GPU. Recipes with flash-only kernels refuse non-flash.",
+    )
     rl.add_argument(
         "--expert-full-count",
         type=int,
@@ -698,6 +706,14 @@ def _add_launch_args(p: argparse.ArgumentParser) -> None:
     rl.add_argument("--use-session-server", action="store_true")
     rl.add_argument("--session-server-ip", default=None)
     rl.add_argument("--session-server-port", type=int, nargs="+", default=None)
+    rl.add_argument(
+        "--session-server-workers",
+        type=int,
+        default=None,
+        help="Miles session server instance count (default: Miles' 32); lower it "
+        "on few-core single-GPU islands to avoid the parallel-import storm "
+        "(32 sglang imports can miss the 300 s ready window)",
+    )
     rl.add_argument("--tito-model", default=None)
     rl.add_argument(
         "--codex-backend-profile",

yeto/launcher.py

@@ -2627,13 +2627,14 @@ def _prepare_rl_args(
     if not args.use_session_server and (
         args.session_server_ip is not None
         or args.session_server_port is not None
+        or getattr(args, "session_server_workers", None) is not None
         or args.tito_model is not None
         or getattr(args, "codex_backend_profile", None) is not None
         or getattr(args, "tito_allowed_append_roles", None) is not None
     ):
         raise ValueError(
-            "--session-server-ip/--session-server-port/--tito-model/"
-            "--codex-backend-profile/"
+            "--session-server-ip/--session-server-port/--session-server-workers/"
+            "--tito-model/--codex-backend-profile/"
             "--tito-allowed-append-roles requires --use-session-server"
         )
     roles = getattr(args, "tito_allowed_append_roles", None)
@@ -2650,6 +2651,8 @@ def _prepare_rl_args(
         raise ValueError(
             "--session-server-port requires one positive port or an increasing range"
         )
+    if getattr(args, "session_server_workers", None) is not None and int(args.session_server_workers) < 1:
+        raise ValueError("--session-server-workers must be at least 1")
     _default_session_server_port(args)
 
     specs = [rl_island_spec(args, spec) for spec in parse_gpu_spec(args.gpu)]
@@ -3968,7 +3971,12 @@ def make_miles_island_task(
     flags = (
         f" --model {shlex.quote(args.model)}"
         f" --rl-model-recipe {shlex.quote(args.rl_model_recipe)}"
-        f" --data {shlex.quote(learner_data_arg(args.data))}"
+        + (
+            f" --rl-training-attention-backend {args.rl_training_attention_backend}"
+            if getattr(args, "rl_training_attention_backend", "auto") not in (None, "auto")
+            else ""
+        )
+        + f" --data {shlex.quote(learner_data_arg(args.data))}"
         + (
             " --rl-single-island-no-sync"
             if getattr(args, "rl_single_island_no_sync", False)
@@ -4101,6 +4109,8 @@ def make_miles_island_task(
         if args.session_server_port:
             ports = " ".join(str(port) for port in args.session_server_port)
             flags += f" --session-server-port {ports}"
+        if getattr(args, "session_server_workers", None) is not None:
+            flags += f" --session-server-workers {int(args.session_server_workers)}"
         if args.tito_model:
             flags += f" --tito-model {shlex.quote(args.tito_model)}"
         if getattr(args, "codex_backend_profile", None):

yeto/rl/adapters/miles/config.py

@@ -449,6 +449,7 @@ def _check_sglang_tp(value, config):
     "agent.use_session_server": _ok,
     "agent.session_server_ip": _ok,
     "agent.session_server_port": _ok,
+    "agent.session_server_workers": _ok,
     "agent.tito_model": _ok,
     "agent.tito_allowed_append_roles": _stock_codex_append_roles_or_none(
         "upstream Miles has no --tito-allowed-append-roles; the append roles are "
@@ -972,6 +973,8 @@ def translate_run_config(
         if agent.session_server_port:
             values.append("--session-server-port")
             values.extend(str(port) for port in agent.session_server_port)
+        if agent.session_server_workers is not None:
+            values.extend(("--session-server-workers", str(agent.session_server_workers)))
         if agent.tito_model:
             resolver = tito_parser_resolver or _resolve_tito_parsers
             values.extend(("--tito-model", agent.tito_model))

yeto/rl/adapters/miles/island_entry.py

@@ -35,6 +35,11 @@ def parse_args(argv=None):
         choices=["generic", "deepseek-v4-flash"],
         default="generic",
     )
+    parser.add_argument(
+        "--rl-training-attention-backend",
+        choices=["auto", "flash", "fused", "unfused"],
+        default="auto",
+    )
     parser.add_argument("--expert-full-count", type=int, default=0)
     parser.add_argument("--expert-full-lr", type=float, default=1e-6)
     parser.add_argument("--expert-selection-sha256", default=None)
@@ -126,6 +131,7 @@ def parse_args(argv=None):
     parser.add_argument("--use-session-server", action="store_true")
     parser.add_argument("--session-server-ip", default=None)
     parser.add_argument("--session-server-port", type=int, nargs="+", default=None)
+    parser.add_argument("--session-server-workers", type=int, default=None)
     parser.add_argument("--tito-model", default=None)
     parser.add_argument("--codex-backend-profile", default=None)
     parser.add_argument("--codex-exclude-reasoning-from-loss", action="store_true")
@@ -1735,6 +1741,8 @@ def _legacy_miles_argv(config) -> list[str]:
         if agent.session_server_port:
             values.append("--session-server-port")
             values.extend(str(port) for port in agent.session_server_port)
+        if agent.session_server_workers is not None:
+            values.extend(("--session-server-workers", str(agent.session_server_workers)))
         if agent.tito_model:
             from miles.utils.chat_template_utils import (
                 resolve_reasoning_and_tool_call_parser,