S26-E9: robustness fixes (preemption marker, verl cuts, data slots, one-command verl pairs)

tests/test_head_mode.py

@@ -907,3 +907,23 @@ def test_head_public_image_sends_no_registry_login(fake_sky, monkeypatch, tmp_pa
     (_, job_task), = fake_sky["execs"]
     assert job_task.secrets == {"MODAL_TOKEN_ID": "ak-1", "MODAL_TOKEN_SECRET": "as-1"}
     assert not any(k.startswith("SKYPILOT_DOCKER") for k in (job_task.envs or {}))
+
+
+def test_head_syncer_skips_a_synced_groups_sampler_islands(monkeypatch):
+    """S26 e9b-sync-a: the head syncer waited for 3 learners (2 trainers + the
+    sampler island), so the two synced trainers never got their first round."""
+    seen = {}
+
+    class FakeSyncer:
+        def __init__(self, args, n):
+            seen["n"] = n
+            raise SystemExit(0)
+
+    monkeypatch.setattr(launcher, "LocalSyncer", FakeSyncer)
+    ns = cli.parse_args(["--training-mode", "rl", "--rl-backend", "verl", "--gpu",
+                         "modal:1xa10g,modal:1xa10g,modal:1xa10g", "--model", "Qwen/Qwen3-0.6B",
+                         "--data", "org/ds", "--rl-sample-role", "trainer", "--rl-island-sample-role",
+                         "2:sampler", "--rl-max-policy-age", "2", "--cluster-prefix", "hsg"])
+    with pytest.raises(SystemExit):
+        cli.cmd_head(json.dumps(cli._serializable_args(ns)))
+    assert seen["n"] == 2

tests/test_launch_preflight_guards.py

@@ -143,7 +143,32 @@ def test_threads_off_skips_and_warns():
     out = io.StringIO()
     rec = lp.thread_preflight(_targs(preflight_threads="off"),
                               counter=lambda: pytest.fail("off must not read"), out=out)
-    assert rec == {"mode": "off", "disabled": True} and "OFF" in out.getvalue()
+    assert rec == {"mode": "off", "disabled": True, "relaxed": {"mode": ["off", "wait"]}}
+    assert "OFF" in out.getvalue() and "RELAXED" in out.getvalue()
+
+
+def test_threads_default_is_on_and_not_relaxed():
+    out = io.StringIO()
+    rec = lp.thread_preflight(_targs(), counter=_counter([100]), out=out)
+    assert rec["mode"] == "wait" and rec["relaxed"] == {} and "RELAXED" not in out.getvalue()
+
+
+def test_threads_loosened_values_are_logged_and_recorded():
+    out = io.StringIO()
+    rec = lp.thread_preflight(_targs(preflight_thread_start=2950, preflight_thread_hard=3200),
+                              counter=_counter([2900]), out=out)
+    assert rec["relaxed"] == {"start": [2950, 2800], "hard": [3200, 3000]}
+    assert "start 2950 (default 2800)" in out.getvalue() and "hard 3200 (default 3000)" in out.getvalue()
+    # a stricter setting is not a relaxation
+    assert lp.thread_relaxations(lp.ThreadConfig(mode="error", start=100, hard=200)) == {}
+
+
+def test_eval_launch_args_keep_the_thread_preflight_on():
+    from pathlib import Path
+    from tools.eval import miles
+    root = Path(__file__).resolve().parents[1]
+    assert "--preflight-threads off" not in (root / "tools/eval/miles_head.sh").read_text()
+    assert "--preflight-threads off" not in Path(miles.__file__).read_text()
 
 
 def test_thread_message_text():

tests/test_rl_spot_sampler_replace.py

@@ -339,7 +339,10 @@ def test_the_island_resolves_its_sampler_id_and_the_shards():
     assert (env[br.SAMPLER_ID_ENV], env[br.SHARDS_ENV]) == ("island1", "island0,island1")
 
 
-def test_only_sampler_islands_get_the_reclaim_marker():
+def test_sampler_and_resumable_trainer_islands_get_the_reclaim_marker():
+    """S25 C1: the sampler islands; S26 E9.1: also the trainer, a resumable island on
+    Modal (--rl-checkpoint-store), whose driver turns the marker into the preemption
+    notice."""
     from yeto.gpu_spec import parse_gpu_spec
     from yeto.launch.modal_islands import build_modal_island_config
     from yeto.launcher import _island_task, make_miles_island_task
@@ -348,7 +351,7 @@ def test_only_sampler_islands_get_the_reclaim_marker():
     for m, spec in enumerate(parse_gpu_spec(args.gpu)):
         task = _island_task(make_miles_island_task, args, spec, m, 3, "none", {})
         cfg = build_modal_island_config(args, spec, m, task, "none")
-        assert (RECLAIM_MARKER_ENV in cfg.envs) == (m < 2)
+        assert RECLAIM_MARKER_ENV in cfg.envs, m
 
 
 def test_two_samplers_refuse_the_trainer_scheduler():

tests/test_rl_verl_adapter.py

@@ -94,7 +94,7 @@ def test_verl_run_config_rules_refuse_multi_gpu_and_megatron_options():
     (dict(top_p=0.9), "top_p"),
     (dict(top_k=50), "top_p/top_k"),
     (dict(temperature=0.7), "温度"),
-    (dict(n_gpus=2), "1 张卡"),
+    (dict(n_gpus=0), "n_gpus 必须 >= 1"),
 ])
 def test_config_rules_refuse(kw, match):
     with pytest.raises(vconf.VerlConfigError, match=match):

tests/test_rl_verl_buffer_trainer.py

@@ -218,7 +218,7 @@ def yeto_call_workers(self, fn, tensors, reset):
         return [{"applied": len(tensors)}]
 
 
-def _trainer(tmp_path, store_dir, actor, *, rounds, required, base):
+def _trainer(tmp_path, store_dir, actor, *, rounds, required, base, buffer=None):
     from yeto.rl.adapters.verl.entry import verl_capabilities
     from yeto.rl.adapters.verl.fully_async_ports import FullyAsyncIsland, FullyAsyncTrainerGroup
     from yeto.rl.adapters.verl.ports_impl import VerlPolicyState
@@ -234,8 +234,10 @@ def _trainer(tmp_path, store_dir, actor, *, rounds, required, base):
     tape = EventTape(tmp_path / "tape.jsonl", 0)
     island.emit = lambda event, **f: tape.append({"event": event, **f})
     store = CheckpointStore(store_dir)
+    if buffer is None:  # not ``or``: an empty FileSampleBuffer is falsy (len 0)
+        buffer = FileSampleBuffer(store, batch_size=1, create_index=False)
     rollout, publisher, placement = install_buffer_ports(
-        island, buffer=FileSampleBuffer(store, batch_size=1, create_index=False),
+        island, buffer=buffer,
         releases=PolicyReleaseStore(store), shape=bf.RowShape(64, 16, 0), base_dir=base,
         tokenize=lambda text: [7, 8], poll_s=0.01, wait_s=30)
     profile = far.execution_profile_for(TOLERANT, 1, groups_per_round=required, samples_per_group=2, sync="none")
@@ -353,8 +355,10 @@ def test_island_entry_plans_a_buffer_trainer_and_refuses_other_roles():
                                                 "--rl-single-island-no-sync",
                                                 "--rl-max-policy-age", "1", "--rl-sample-wait-s", "60"))
     assert section == {"store": "/s", "wait_s": 60.0, "lane": None}
+    # S26 E9.4: a sampler island runs the engine sampler (no buffer section); other roles stop
+    assert sample_buffer_section(_entry_args("--rl-sample-role", "sampler", "--rl-resume-store", "/s")) is None
     with pytest.raises(SystemExit, match="engine sampler"):
-        sample_buffer_section(_entry_args("--rl-sample-role", "sampler", "--rl-resume-store", "/s"))
+        sample_buffer_section(_entry_args("--rl-sample-role", "convertible", "--rl-resume-store", "/s"))
     with pytest.raises(SystemExit, match="max-policy-age"):
         sample_buffer_section(_entry_args("--rl-sample-role", "trainer", "--rl-resume-store", "/s"))
 
@@ -469,3 +473,83 @@ def test_a_synced_feed_reads_its_lane_translates_merged_versions_and_logs_each_r
     # a restarted island continues after the recorded cursor
     again = _synced_publisher(tmp_path, store, 1, [])[2].feed
     assert again.buffer.read_cursor == 6
+
+
+def test_two_shard_samplers_drain_their_backlog_through_the_verl_release(tmp_path):
+    """S26 E9 G1 (e9-g1-f) root cause: the verl trainer released without the per-shard
+    read positions, so a shard sampler never saw its backlog drain (stuck at the cap).
+    Two shard samplers with a 2-group cap now feed three rounds of 2 groups."""
+    from yeto.rl.engine.engine_sampler_entry import build_stores
+    from yeto.rl.engine.sample_buffer import open_trainer_buffer
+
+    store_dir = tmp_path / "store"
+    store = CheckpointStore(store_dir)
+    base, _weights = _base(tmp_path, dtype=torch.float32)
+    done, threads, events = threading.Event(), [], []
+    for sid in ("island0", "island1"):
+        releases, buffer = build_stores({"store": str(store_dir), "sampler_id": sid,
+                                         "buffer_shards": ["island0", "island1"]})
+        config = EngineSamplerConfig(sampler_id=sid, num_rollouts=40, max_backlog_groups=2,
+                                     sampling=SamplingParams(max_tokens=6, n=2), max_prompt_tokens=32, poll_s=0.01)
+        sampler = build_engine_sampler(
+            config, engine=FakeEngine(max_model_len=64), releases=releases, buffer=buffer,
+            emit=lambda n, _sid=sid, **f: events.append((_sid, n, f)),
+            prompts=lambda r, _sid=sid: [PromptItem(f"{_sid}-p{r}", f"prompt {r}", label="x")],
+            reward=lambda item, c: float(len(c.token_ids) % 2),
+            checks=PostLoadChecks(expected_expert_matrices=None, fixed_prompts=(), probe_pairs=(("Q", " 7"),)),
+            should_stop=done.is_set, sleep=lambda s: time.sleep(0.005))
+        def run(_sampler=sampler, _sid=sid):
+            try:
+                _sampler.run()
+            except BaseException as error:  # noqa: BLE001 - surfaced by the assert below
+                events.append((_sid, "crashed", {"error": repr(error)}))
+                raise
+
+        threads.append(threading.Thread(target=run, daemon=True))
+        threads[-1].start()
+    trainer_buffer = open_trainer_buffer(store, batch_size=1, create_index=False, shards=["island0", "island1"])
+    driver, _island, _publisher = _trainer(tmp_path, store_dir, FakeBufferActor(), rounds=3, required=2,
+                                           base=base, buffer=trainer_buffer)
+    try:
+        driver.run()
+    except Exception:
+        listing = sorted(str(p.relative_to(store_dir)) for p in (store_dir / "sample-buffer").rglob("*.jsonl"))
+        raise AssertionError(f"buffer {len(trainer_buffer)} shards {trainer_buffer.index.shard_names()} "
+                             f"files {listing}; "
+                             f"sampler events: {[(sid, n) for sid, n, _f in events][-20:]}; "
+                             f"crashes: {[f for _sid, n, f in events if n == 'crashed']}")
+    finally:
+        done.set()
+        for t in threads:
+            t.join(timeout=10)
+    assert driver.rounds_completed == 3
+    releases = _lines(tmp_path / "out" / RELEASE_LOG.format(learner=0))
+    shards = [r["buffer_shards"] for r in releases if r["released"]]
+    assert all(set(s) <= {"island0", "island1"} and sum(s.values()) == r["buffer_selected"]
+               for s, r in zip(shards, [r for r in releases if r["released"]]))
+    assert PolicyReleaseStore(store).latest().buffer_shards == shards[-1]
+
+
+def test_verl_buffer_trainer_ends_finished_when_the_samplers_end_early(tmp_path):
+    """S26 E9 G1 (e9-g1-h): the buffer ended with groups passed over as too old, so
+    the trainer could not fill its last round. On verl that was "rollouter ended
+    the queue" (INFRA_FAILED); it is the buffer's end: the run ends FINISHED."""
+    store_dir = tmp_path / "store"
+    store = CheckpointStore(store_dir)
+    base, _weights = _base(tmp_path, dtype=torch.float32)
+    engine = FakeEngine(max_model_len=64)
+    config = EngineSamplerConfig(sampler_id="island0", num_rollouts=4, max_backlog_groups=8,
+                                 sampling=SamplingParams(max_tokens=6, n=2), max_prompt_tokens=32, poll_s=0.01)
+    sampler = build_engine_sampler(
+        config, engine=engine, releases=PolicyReleaseStore(store), buffer=FileSampleBuffer(store, batch_size=1),
+        emit=lambda n, **f: None,
+        prompts=lambda r: [PromptItem(f"p{r}", f"prompt {r}", label="x")], reward=lambda item, c: float(len(c.token_ids) % 2),
+        checks=PostLoadChecks(expected_expert_matrices=None, fixed_prompts=(), probe_pairs=(("Q", " 7"),)),
+        should_stop=lambda: False, sleep=lambda s: time.sleep(0.005))
+    sampler.run()  # 4 groups, then the data end
+    driver, _island, _publisher = _trainer(tmp_path, store_dir, FakeBufferActor(), rounds=3, required=2, base=base)
+    driver.run()
+    assert driver.rounds_completed == 2
+    events = _lines(tmp_path / "tape.jsonl")
+    exhausted = [e for e in events if e["event"] == "rl_buffer_exhausted"]
+    assert [e["rollout_id"] for e in exhausted] == [2] and "SampleBufferExhausted" in exhausted[0]["error"]

tests/test_rl_verl_cut.py

@@ -342,3 +342,37 @@ def handles(model):
             assert torch.equal(p, q), name
             for key in ("exp_avg", "exp_avg_sq", "step"):
                 assert torch.equal(trained.optimizer.state[p][key], fresh.optimizer.state[q][key]), key
+
+
+def test_a_requires_grad_parameter_outside_the_optimizer_is_not_in_the_cut(tmp_path):
+    """S26 E9.2: verl's state dict carries ``lm_head.weight`` (requires_grad, not in the
+    optimizer: 865 MB per rank file on GPU T-C4). Only the optimizer's parameters are
+    saved; the restore needs nothing else and continues bit for bit."""
+
+    def handles(model):
+        # layer 2 keeps requires_grad but no optimizer holds it (the lm_head shape)
+        opt = torch.optim.AdamW([p for n, p in model.named_parameters() if not n.startswith("2.")], lr=1e-2)
+        return TrainHandles(model, opt, torch.optim.lr_scheduler.LambdaLR(opt, lambda s: 1.0 / (1 + s)))
+
+    trained = handles(_model())
+    assert trained.module[2].weight.requires_grad
+    for r in range(CUT):
+        _round(trained, r)
+    worker = SimpleNamespace(module=trained.module, optimizer=trained.optimizer, lr_scheduler=trained.scheduler)
+    vcut.VerlCutTrainer(lambda fn, *a: [fn(worker, *a)], optimizer_steps_per_round=K).save_cut(
+        epoch=1, context=_context(tmp_path))
+    saved = torch.load(tmp_path / "cut-v" / vcut.rank_file(0), weights_only=True)["model"]
+    assert sorted(saved) == ["0.bias", "0.weight"]
+    fresh = handles(_model())
+    fresh_worker = SimpleNamespace(module=fresh.module, optimizer=fresh.optimizer, lr_scheduler=fresh.scheduler)
+    vcut.VerlCutTrainer(lambda fn, *a: [fn(fresh_worker, *a)], optimizer_steps_per_round=K).restore_cut(
+        "cut-v", epoch=1, root=str(tmp_path), expect=_expect())
+    for r in range(CUT, TOTAL):
+        _round(trained, r)
+        _round(fresh, r)
+    for (name, p), (_, q) in zip(trained.module.named_parameters(), fresh.module.named_parameters()):
+        assert torch.equal(p, q), name
+        if p in trained.optimizer.state:
+            for key in ("exp_avg", "exp_avg_sq", "step"):
+                assert torch.equal(trained.optimizer.state[p][key], fresh.optimizer.state[q][key]), key
+    assert trained.scheduler.last_epoch == fresh.scheduler.last_epoch

tests/test_rl_verl_round_cuts.py

@@ -223,8 +223,8 @@ def test_resume_config_from_flags():
         round_cuts.resume_config(_args(rl_cut_every=2))
     with pytest.raises(ValueError, match="--rl-single-island-no-sync"):
         round_cuts.resume_config(_args(rl_resume_store="/s", rl_single_island_no_sync=False))
-    with pytest.raises(ValueError, match="fully_async"):
-        round_cuts.resume_config(_args(rl_resume_store="/s", rl_max_policy_age=2))
+    # S26 E9.2: the fully_async path (--rl-max-policy-age > 0) cuts too
+    assert round_cuts.resume_config(_args(rl_resume_store="/s", rl_max_policy_age=2))["store"] == "/s"
     with pytest.raises(ValueError, match="must be >= 1"):
         round_cuts.resume_config(_args(rl_resume_store="/s", rl_cut_keep=0))
 

tests/test_s25_g1c_sampler_resume.py

@@ -150,6 +150,8 @@ def test_reclaim_marker_stops_writes_and_leaves_preempted(tmp_path):
     [tape] = (tmp_path / "tape").glob("events.*.jsonl")
     leave = [e for e in _tape(tape) if e["event"] == "rl_sampler_leave"]
     assert len(leave) == 1 and leave[0]["reason"] == "PREEMPTED"
+    # the next incarnation continues: this tape is not complete (S26 E9)
+    assert not any(e["event"] == "rl_learner_finalized" for e in _tape(tape))
 
 
 def test_caller_stop_ends_the_island_finished_and_on_event_sees_the_tape(tmp_path):
@@ -166,6 +168,9 @@ def test_caller_stop_ends_the_island_finished_and_on_event_sees_the_tape(tmp_pat
     [tape] = (tmp_path / "tape").glob("events.*.jsonl")
     assert [e["event"] for e in _tape(tape)] == [e for e, _ in seen]
     assert [f["reason"] for e, f in seen if e == "rl_sampler_leave"] == ["FINISHED"]
+    # S26 E9 G1 (e9-g1-i, launcher exit 3): a finished island's tape ends with the
+    # record the launcher requires of every island tape
+    assert seen[-1] == ("rl_learner_finalized", {"rounds": 1})
 
 
 def test_hold_after_reclaim_only_when_the_marker_is_there(tmp_path):
@@ -176,3 +181,18 @@ def test_hold_after_reclaim_only_when_the_marker_is_there(tmp_path):
     assert entry.hold_after_reclaim(env, hold_s=5.0, sleep=slept.append) is False and slept == []
     (tmp_path / "marker").write_text("{}")
     assert entry.hold_after_reclaim(env, hold_s=5.0, sleep=slept.append) is True and slept == [5.0]
+
+
+def test_a_launcher_island_tape_is_recovered_complete(tmp_path):
+    """S26 E9 G1 (e9-g1-j, exit 3): with the spec's island_id every record carries
+    it, so the launcher's tape recovery accepts the file and sees the finalized
+    record; without it the records are discarded as not island tape lines."""
+    from yeto.rl.event_echo import TapeCollector
+
+    spec = {**_spec(tmp_path, rounds=1), "island_id": 0}
+    assert entry.run(spec, resource_sampler=lambda emit: None, incarnation="aaaa0001") == 1
+    [tape] = (tmp_path / "tape").glob("events.*.jsonl")
+    assert all(r["island_id"] == 0 for r in _tape(tape))
+    collector = TapeCollector(tmp_path / "events" / "l0.jsonl", fresh=False)
+    assert collector.recover_from_file(tape) is True
+    assert collector.discarded == 0 and collector.count == len(_tape(tape))