diff --git a/tests/test_head_mode.py b/tests/test_head_mode.py index 4cae7d12..4d1a2921 100644 --- a/tests/test_head_mode.py +++ b/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 diff --git a/tests/test_launch_preflight_guards.py b/tests/test_launch_preflight_guards.py index ab9be398..c8d9ee6f 100644 --- a/tests/test_launch_preflight_guards.py +++ b/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(): diff --git a/tests/test_rl_spot_sampler_replace.py b/tests/test_rl_spot_sampler_replace.py index cbe2e7fa..43dcf870 100644 --- a/tests/test_rl_spot_sampler_replace.py +++ b/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(): diff --git a/tests/test_rl_verl_adapter.py b/tests/test_rl_verl_adapter.py index 96c58845..57a5677e 100644 --- a/tests/test_rl_verl_adapter.py +++ b/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): diff --git a/tests/test_rl_verl_buffer_trainer.py b/tests/test_rl_verl_buffer_trainer.py index 58dea824..b15ba1e5 100644 --- a/tests/test_rl_verl_buffer_trainer.py +++ b/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"] diff --git a/tests/test_rl_verl_cut.py b/tests/test_rl_verl_cut.py index 390d4351..e15ad42b 100644 --- a/tests/test_rl_verl_cut.py +++ b/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 diff --git a/tests/test_rl_verl_round_cuts.py b/tests/test_rl_verl_round_cuts.py index 6653eaa6..7451e50a 100644 --- a/tests/test_rl_verl_round_cuts.py +++ b/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)) diff --git a/tests/test_s25_g1c_sampler_resume.py b/tests/test_s25_g1c_sampler_resume.py index 88ad0fa0..e29adaaa 100644 --- a/tests/test_s25_g1c_sampler_resume.py +++ b/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)) diff --git a/tests/test_s26_e9_data_slots.py b/tests/test_s26_e9_data_slots.py new file mode 100644 index 00000000..d9f09bc5 --- /dev/null +++ b/tests/test_s26_e9_data_slots.py @@ -0,0 +1,297 @@ +"""S26 E9.3: disjoint sampler data slots, a normal trainer end on an exhausted +buffer, and the launcher following a rescheduled Modal container's log.""" + +from __future__ import annotations + +import io +import json +import random +import sys +from types import SimpleNamespace + +import pytest +import torch + +from yeto.rl.engine import sample_roles +from yeto.rl.engine.buffer_payload import SampleBufferExhausted, SampleBufferRolloutError, is_buffer_exhausted +from yeto.rl.engine.engine_sampler_entry import JsonlPrompts, SamplerSpecError +from yeto.rl.engine.sample_roles import DataSlot + + +# ------------------------------------------------------------------ DataSlot +def test_data_slot_parse_and_rows(): + assert DataSlot.parse(None) == DataSlot() and DataSlot().whole + slot = DataSlot.parse("1/3") + assert (slot.index, slot.count, slot.env()) == (1, 3, "1/3") + assert slot.rows(list(range(10))) == [1, 4, 7] + for bad in ("3/3", "-1/2", "1/0", "2"): + with pytest.raises(ValueError): + DataSlot.parse(bad) + with pytest.raises(ValueError, match="empty"): + DataSlot(2, 3).rows([0, 1]) + + +def test_slots_of_all_samplers_cover_the_data_once(): + rows = list(range(17)) + drawn = [r for k in range(3) for r in DataSlot(k, 3).rows(rows)] + assert sorted(drawn) == rows + + +def test_pair_island_args_give_each_sampler_its_slot_and_flag(): + pair = SimpleNamespace(rl_island_sample_role=["0:sampler", "1:sampler", "2:trainer"], + rl_sampler_max_backlog_rounds=None, rl_sample_wait_s=None, total_steps=8, + rl_sampler_spot=False) + a0, a1, t = (sample_roles.island_args(pair, m, 3) for m in range(3)) + assert (a0.rl_sampler_data_slot, a1.rl_sampler_data_slot) == ("0/2", "1/2") + assert getattr(t, "rl_sampler_data_slot", None) is None + assert " --rl-sampler-data-slot 1/2" in sample_roles.island_flags(a1) + one = SimpleNamespace(rl_island_sample_role=["0:sampler", "1:trainer"], rl_sampler_max_backlog_rounds=None, + rl_sample_wait_s=None, total_steps=8, rl_sampler_spot=False) + assert sample_roles.island_args(one, 0, 2).rl_sampler_data_slot is None # one sampler: the whole data + + +def test_data_slot_flag_on_both_layers(): + import argparse + + from yeto.rl.engine.run_flags import CLI, ISLAND, flags_for + + for layer in (CLI, ISLAND): + parser = argparse.ArgumentParser() + for flag in flags_for(layer): + if flag.options[0] == "--rl-sampler-data-slot": + parser.add_argument(*flag.options, **dict(flag.kwargs)) + assert parser.parse_args(["--rl-sampler-data-slot", "0/2"]).rl_sampler_data_slot == "0/2" + + +# ------------------------------------------------------------ engine sampler +def _prompts(tmp_path, n=10): + path = tmp_path / "p.jsonl" + path.write_text("".join(json.dumps({"q": f"q{i}", "a": str(i)}) + "\n" for i in range(n))) + return path + + +def test_engine_sampler_prompts_of_two_slots_are_disjoint(tmp_path): + path = _prompts(tmp_path) + ids = [] + for slot in ("0/2", "1/2"): + prompts = JsonlPrompts(path, input_key="q", label_key="a", groups_per_round=2, data_slot=slot) + ids.append([item.prompt_id for r in range(2) for item in prompts(r)]) + assert ids == [["row0", "row2", "row4", "row6"], ["row1", "row3", "row5", "row7"]] + whole = JsonlPrompts(path, input_key="q", label_key="a", groups_per_round=2) + assert [i.prompt_id for i in whole(1)] == ["row2", "row3"] # unchanged without a slot + with pytest.raises(SamplerSpecError, match="data_slot"): + JsonlPrompts(path, input_key="q", label_key="a", groups_per_round=2, data_slot="2/2") + + +def test_stage1_sampler_spec_carries_the_slot_only_with_several_samplers(): + from yeto.rl import stage1 + + src = open(stage1.__file__).read() + assert "slot=DataSlot(samplers.index(node)" in src and '"data_slot": slot.env()' in src + + +# --------------------------------------------------------------- Miles source +class _FakeDataset: + """The fields and shuffle of Miles' ``miles.utils.data.Dataset``.""" + + def __init__(self, rows, seed=3): + self.origin_samples = list(rows) + self.samples = list(rows) + self.seed, self.epoch_id = seed, -1 + + def shuffle(self, new_epoch_id): + if self.epoch_id == new_epoch_id: + return + random.seed(self.seed + new_epoch_id) + permutation = list(range(len(self.samples))) + random.shuffle(permutation) + self.samples = [self.origin_samples[i] for i in permutation] + self.epoch_id = new_epoch_id + + def __len__(self): + return len(self.samples) + + +def _source(rows, shuffle): + source = SimpleNamespace(dataset=_FakeDataset(rows), epoch_id=0, sample_offset=0, sample_group_index=0, + sample_index=0) + if shuffle: + source.dataset.shuffle(0) + return source + + +@pytest.mark.parametrize("shuffle", [False, True]) +def test_miles_data_source_slots_are_disjoint(shuffle): + from yeto.rl.adapters.miles.buffer_rollout import apply_data_slot + + rows = list(range(20)) + drawn = [] + for k in range(2): + source = _source(rows, shuffle) + assert apply_data_slot(source, DataSlot(k, 2), shuffle=shuffle) == 10 + assert sorted(source.dataset.samples) == rows[k::2] + drawn.append(set(source.dataset.samples)) + assert not drawn[0] & drawn[1] and drawn[0] | drawn[1] == set(rows) + + +@pytest.mark.parametrize("shuffle", [False, True]) +def test_miles_slot_rows_record_their_unsplit_row(shuffle): + """S26 E9 G4: the buffer payload shows which data row a sampler drew.""" + from yeto.rl.adapters.miles.buffer_rollout import DATA_ROW_KEY, apply_data_slot + + rows = [SimpleNamespace(prompt=f"q{r}", metadata={"label": r}) for r in range(9)] + source = _source(rows, shuffle) + assert apply_data_slot(source, DataSlot(1, 3), shuffle=shuffle) == 3 + kept = sorted(s.metadata[DATA_ROW_KEY] for s in source.dataset.samples) + assert kept == [1, 4, 7] and all(s.metadata["label"] == s.metadata[DATA_ROW_KEY] for s in rows) + + +def test_miles_data_slot_refuses_after_the_first_draw(): + from yeto.rl.adapters.miles.buffer_rollout import apply_data_slot + + source = _source(list(range(4)), False) + source.sample_offset = 2 + with pytest.raises(SampleBufferRolloutError, match="before the first draw"): + apply_data_slot(source, DataSlot(0, 2), shuffle=False) + whole = _source(list(range(4)), False) + assert apply_data_slot(whole, DataSlot(), shuffle=False) == 4 + + +def test_island_env_exports_the_slot_for_a_sampler_only(): + from yeto.rl.adapters.miles import buffer_rollout as br + + env = br.island_env("sampler", store_root="/s", sampler_id="island0", shards=["island0", "island1"], + data_slot="0/2") + assert env[br.DATA_SLOT_ENV] == "0/2" + with pytest.raises(ValueError, match="only a sampler"): + br.island_env("trainer", store_root="/s", max_policy_age=2, data_slot="0/2") + + +def test_every_island_env_reaches_the_ray_workers(): + """S26 E9 G4 (e9-g4-a): the slot env was set in the island process but not + forwarded to the Ray rollout worker, so both samplers drew the same prompts.""" + from yeto.rl.adapters.miles import buffer_rollout as br + + envs = [br.island_env("sampler", store_root="/s", sampler_id="island0", shards=["island0", "island1"], + data_slot="0/2"), + br.island_env("trainer", store_root="/s", max_policy_age=2, wait_s=9.0, lane=br.BufferLane(1, 2), + shards=["island0", "island1"])] + assert {k for env in envs for k in env} <= set(br.ISLAND_ENVS) + + +# ------------------------------------------------------- exhausted buffer end +def test_is_buffer_exhausted_follows_wrappers(): + inner = SampleBufferExhausted("ended") + assert is_buffer_exhausted(inner) + try: + try: + raise inner + except SampleBufferExhausted as e: + raise RuntimeError("ray task failed") from e + except RuntimeError as outer: + assert is_buffer_exhausted(outer) + ray_like = RuntimeError("RayTaskError") + ray_like.cause = inner + assert is_buffer_exhausted(ray_like) + dual = type("SampleBufferExhausted", (RuntimeError,), {})("dual class of another process") + assert is_buffer_exhausted(dual) + assert not is_buffer_exhausted(SampleBufferRolloutError("other")) + assert not is_buffer_exhausted(None) + + +def test_batch_reader_raises_exhausted_on_an_ended_empty_buffer(): + from yeto.rl.adapters.miles.buffer_rollout import BatchReader + + buffer = SimpleNamespace(read_cursor=5, data_ended=True, select=lambda consumer, n: []) + reader = BatchReader(buffer, max_policy_age=2, wait_s=0, poll_s=0) + with pytest.raises(SampleBufferExhausted, match="no trainable group"): + reader.take(3, 4, current_version=3) + + +def _fake_driver(tmp_path, sync, exhaust_at): + from yeto.rl.engine.algorithm import AlgorithmSpec + from yeto.rl.engine.driver import EventTape, IslandDriver + from yeto.rl.engine.fake import FakeEngine, fake_capabilities + + engine = FakeEngine(tensors={"base_model.model.layer.lora_A.weight": torch.zeros(1, 2)}, step_delta=1.0, capability_profile="miles-like") + generate = engine.rollout.generate + + def gen(rollout_id, **kw): + if rollout_id == exhaust_at: + raise RuntimeError("ray task failed") from SampleBufferExhausted("ended") + return generate(rollout_id, **kw) + + engine.rollout.generate = gen + return IslandDriver(learner_id=0, rollout=engine.rollout, trainer=engine.trainer, + policy_state=engine.policy_state, publisher=engine.publisher, + placement=engine.placement, algorithm=AlgorithmSpec(), sync=sync, + events=EventTape(tmp_path / "e.jsonl", 0), capabilities=fake_capabilities("miles-like")) + + +def test_driver_ends_finished_when_the_buffer_is_exhausted(tmp_path): + from yeto.rl.engine.bridges import LocalOnlySync + + final = _fake_driver(tmp_path, LocalOnlySync(5), exhaust_at=3).run() + assert final.policy_version == 3 + events = [json.loads(line) for line in (tmp_path / "e.jsonl").read_text().splitlines()] + names = [e["event"] for e in events] + assert "rl_buffer_exhausted" in names and "island_leave" not in names + assert next(e for e in events if e["event"] == "rl_buffer_exhausted")["rollout_id"] == 3 + + +def test_driver_still_fails_on_other_errors_and_on_a_strict_roster(tmp_path): + from yeto.rl.engine.bridges import LocalOnlySync, StrictAvgSync + + assert StrictAvgSync.ends_on_data_end is False + sync = LocalOnlySync(5) + sync.ends_on_data_end = False + with pytest.raises(RuntimeError, match="ray task failed"): + _fake_driver(tmp_path, sync, exhaust_at=2).run() + + +# ------------------------------------------------- rescheduled container log +def test_container_log_follower_streams_new_containers_only(): + from yeto.modal_runner import ContainerLogFollower + + lists = iter([{"ta-a"}, {"ta-a"}, {"ta-a", "ta-b"}, {"ta-b"}]) + streamed = [] + + def stream(cid): + streamed.append(cid) + yield "[modal-island 1] rank 0 container ta-b\n" + yield "round 2 done\nround 3 done\n" + + class Collector: + def __init__(self): + self.lines = [] + + def feed(self, line): + self.lines.append(line) + + collector = Collector() + out = io.StringIO() + follower = ContainerLogFollower(lambda: next(lists), stream, expected=1, + route=lambda learner: ("island1", collector) if learner == 1 else None, + out=out) + assert follower.poll_once() == [] and follower.baseline == {"ta-a"} + assert follower.poll_once() == [] + assert follower.poll_once() == ["ta-b"] + follower.threads[0].join(5) + assert follower.poll_once() == [] # followed once + text = out.getvalue() + assert "new Modal container ta-b" in text + assert "[island1] round 2 done" in text and "[island1] round 3 done" in text + assert streamed == ["ta-b"] and collector.lines[-1] == "round 2 done\nround 3 done" + + +def test_container_log_follower_only_when_new_containers_are_expected(): + from yeto.launch.modal_islands import container_log_follower_for + + ops = SimpleNamespace(list_container_ids=lambda: set(), stream_container_logs=lambda cid: iter(())) + cfgs = {"island0": SimpleNamespace(num_nodes=1, learner_id=0)} + assert container_log_follower_for(None, ops, cfgs, None, poller=object(), start=False) is None + assert container_log_follower_for(None, None, cfgs, None, poller=None, start=False) is None + follower = container_log_follower_for(None, ops, cfgs, None, poller=None, start=False) + assert follower is not None and follower.expected_count == 1 + assert container_log_follower_for(None, SimpleNamespace(list_container_ids=lambda: set()), cfgs, None, + poller=None, start=False) is None diff --git a/tests/test_s26_e9_preempt_marker.py b/tests/test_s26_e9_preempt_marker.py new file mode 100644 index 00000000..c376afed --- /dev/null +++ b/tests/test_s26_e9_preempt_marker.py @@ -0,0 +1,201 @@ +"""S26 E9.1: the Modal reclaim signal reaches a driver inside a Ray actor through +the reclaim marker file (runner writes it; a watcher thread writes the notice).""" + +from __future__ import annotations + +import io +import json +import signal +import threading +import time +from pathlib import Path +from types import SimpleNamespace + +from yeto.rl.engine.resume import PREEMPT_WATCH_THREAD, watch_preempt_marker +from yeto.rl.engine.sampler_resume import RECLAIM_MARKER_ENV + + +def _controller(tmp_path): + committed = [] + return SimpleNamespace(checkpoint_store=tmp_path / "store", incarnation={"id": "abc", "index": 2}, + store=SimpleNamespace(commit=lambda: committed.append(1)), committed=committed) + + +def _wait(cond, limit=5.0): + end = time.monotonic() + limit + while time.monotonic() < end: + if cond(): + return True + time.sleep(0.02) + return False + + +def test_no_marker_path_no_watcher(tmp_path): + assert watch_preempt_marker(_controller(tmp_path), environ={}) is None + + +def test_marker_from_the_runner_becomes_the_store_notice_off_the_main_thread(tmp_path): + from yeto.modal_runner import _run_forwarding_to_learner + + marker = tmp_path / "reclaim" + env = {RECLAIM_MARKER_ENV: str(marker)} + controller = _controller(tmp_path) + out = io.StringIO() + holder = {} + # the driver of a verl island runs in a Ray actor thread: start the watcher off the main thread + t = threading.Thread(target=lambda: holder.update(w=watch_preempt_marker( + controller, environ=env, poll_s=0.02, out=out))) + t.start() + t.join() + watcher = holder["w"] + assert watcher is not None and watcher._thread.name == PREEMPT_WATCH_THREAD + assert not watcher.fired.is_set() + + # the Modal runner (container main process) gets the signal and writes the marker + handlers = {} + sig = SimpleNamespace(SIGINT=signal.SIGINT, SIGTERM=signal.SIGTERM, + signal=lambda num, fn: handlers.__setitem__(num, fn)) + + class Proc: + def wait(self): + handlers[signal.SIGTERM](signal.SIGTERM, None) + return 0 + + assert _run_forwarding_to_learner(["true"], env, popen=lambda cmd, env: Proc(), sig=sig) == 0 + assert marker.exists() + assert _wait(watcher.fired.is_set) + notices = tmp_path / "store" / "preempt-notices" + assert _wait(lambda: any(notices.glob("*.json"))) + record = json.loads(next(notices.glob("*.json")).read_text()) + assert record["signal"] == "SIGTERM (reclaim marker)" and record["incarnation"]["id"] == "abc" + assert record["saved"] is False and controller.committed + assert "rl_preempt_notice" in out.getvalue() + + +def test_unreadable_marker_still_writes_the_notice(tmp_path): + marker = tmp_path / "reclaim" + marker.write_text("not json") + controller = _controller(tmp_path) + watcher = watch_preempt_marker(controller, environ={RECLAIM_MARKER_ENV: str(marker)}, poll_s=0.02, + out=io.StringIO()) + notices = Path(tmp_path / "store" / "preempt-notices") + assert _wait(lambda: any(notices.glob("*.json"))) + assert json.loads(next(notices.glob("*.json")).read_text())["signal"] == "unknown (reclaim marker)" + watcher.stop() + + +def test_launcher_sets_the_marker_path_on_resumable_modal_islands_only(): + from yeto.launcher import _resume_marker_env + + resumable = SimpleNamespace(rl_checkpoint_store="modal-volume://v/run", rl_elastic=False) + assert _resume_marker_env(resumable, "modal") == {RECLAIM_MARKER_ENV: "/tmp/yeto-reclaim-requested"} + assert _resume_marker_env(resumable, "nebius") == {} + assert _resume_marker_env(SimpleNamespace(rl_checkpoint_store=None), "modal") == {} + # --rl-elastic keeps its own checkpoint store and LEAVE path + assert _resume_marker_env(SimpleNamespace(rl_checkpoint_store="modal-volume://v/run", rl_elastic=True), + "modal") == {} + + +def test_runner_turns_the_signal_into_the_marker_when_the_launcher_set_it(): + from yeto import modal_runner + + calls = [] + orig = modal_runner._run_forwarding_to_learner + try: + modal_runner._run_forwarding_to_learner = lambda cmd, env, **kw: calls.append(env) or 0 + assert modal_runner.run_island_script(["true"], {RECLAIM_MARKER_ENV: "/tmp/x"}, + popen=lambda *a, **k: None) == 0 + finally: + modal_runner._run_forwarding_to_learner = orig + assert calls + + +def test_verl_island_in_an_actor_thread_watches_the_marker(tmp_path, monkeypatch): + """The verl wiring off the main thread: no signal handler, the marker watch is on.""" + import tests.test_rl_verl_round_cuts as vrc + from yeto.rl.adapters.verl import round_cuts + + marker = tmp_path / "reclaim" + monkeypatch.setenv(RECLAIM_MARKER_ENV, str(marker)) + plan = vrc._plan(store=tmp_path / "store", state_dir=tmp_path / "state") + calls = [] + real = round_cuts.wire_round_cuts + + def wire(driver, resume, **kw): + kw["install_signals"] = True + out = {} + t = threading.Thread(target=lambda: out.update(c=real(driver, resume, **kw))) + t.start() + t.join() + calls.append(resume) + return out["c"] + + monkeypatch.setattr(round_cuts, "wire_round_cuts", wire) + driver, _h, _c, resume, tape = vrc._island(tmp_path, plan, name="island") + wired = vrc._events(tape, "rl_round_cuts_wired")[0] + assert wired["preempt_signals"] == [] and wired["preempt_marker_watch"] is True + marker.write_text(json.dumps({"signal": int(signal.SIGINT)})) + notices = Path(resume.controller.checkpoint_store) / "preempt-notices" + assert _wait(lambda: any(notices.glob("*.json"))) + assert json.loads(next(notices.glob("*.json")).read_text())["signal"] == "SIGINT (reclaim marker)" + + +def test_rank1_reclaim_reaches_the_rank0_driver_through_the_relay(tmp_path): + """Known limit fixed: a reclaim of a rank>0 container of a multi-node island is + relayed by its runner to rank 0, whose marker the driver watcher turns into the notice.""" + from yeto import modal_runner + + marker0, marker1 = tmp_path / "reclaim0", tmp_path / "reclaim1" + controller = _controller(tmp_path) + watcher = watch_preempt_marker(controller, environ={RECLAIM_MARKER_ENV: str(marker0)}, poll_s=0.02, + out=io.StringIO()) + relay = modal_runner.ReclaimRelay( + lambda num, peer: modal_runner._write_reclaim_marker(str(marker0), num, time.time, from_rank=peer), + port=0, host="127.0.0.1").start() + try: + env1 = {RECLAIM_MARKER_ENV: str(marker1), "SKYPILOT_NODE_RANK": "1", + "SKYPILOT_NODE_IPS": "127.0.0.1\n127.0.0.2", + modal_runner.RECLAIM_RELAY_PORT_ENV: str(relay.port)} + handlers = {} + sig = SimpleNamespace(SIGINT=signal.SIGINT, SIGTERM=signal.SIGTERM, + signal=lambda num, fn: handlers.__setitem__(num, fn)) + + class Proc: + def wait(self): + handlers[signal.SIGINT](signal.SIGINT, None) + return 0 + + assert modal_runner._run_forwarding_to_learner(["true"], env1, popen=lambda cmd, env: Proc(), + sig=sig) == 0 + assert marker1.exists() and marker0.exists() + notices = tmp_path / "store" / "preempt-notices" + assert _wait(lambda: any(notices.glob("*.json"))) + record = json.loads(next(notices.glob("*.json")).read_text()) + assert record["signal"] == "SIGINT (reclaim marker, rank 1)" + finally: + relay.stop() + watcher.stop() + + +def test_rank0_of_a_multi_node_island_starts_the_relay(tmp_path): + from yeto import modal_runner + + env = {RECLAIM_MARKER_ENV: str(tmp_path / "m"), "SKYPILOT_NODE_RANK": "0", + "SKYPILOT_NODE_IPS": "127.0.0.1\n127.0.0.2", modal_runner.RECLAIM_RELAY_PORT_ENV: "0"} + seen = {} + sig = SimpleNamespace(SIGINT=signal.SIGINT, SIGTERM=signal.SIGTERM, signal=lambda num, fn: None) + + class Proc: + def wait(self): + seen["relay"] = [t.name for t in threading.enumerate() if t.name == "reclaim-relay"] + return 0 + + assert modal_runner._run_forwarding_to_learner(["true"], env, popen=lambda cmd, env: Proc(), sig=sig) == 0 + assert seen["relay"] == ["reclaim-relay"] + assert not any(t.name == "reclaim-relay" for t in threading.enumerate()) + + +def test_relay_send_fails_softly_when_rank0_is_gone(): + from yeto import modal_runner + + assert modal_runner.send_reclaim_to_rank0("127.0.0.1", 1, 2, 1, timeout_s=0.2, attempts=1) is False diff --git a/tests/test_s26_e9_verl_fully_async_cut.py b/tests/test_s26_e9_verl_fully_async_cut.py new file mode 100644 index 00000000..288d88db --- /dev/null +++ b/tests/test_s26_e9_verl_fully_async_cut.py @@ -0,0 +1,151 @@ +"""S26 E9.2: round cuts on the verl fully_async path (CPU part). + +The data cursor counts prompts the rounds took from the queue; on resume the +rollouter skips them before it starts; the trainer port cuts and restores +through the trainer actor's worker group like the sync path.""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +import torch + +from tests import test_rl_verl_cut as vc +from tests import test_rl_verl_fully_async_64b as fa +from yeto.rl.adapters.verl import fully_async_cut as fac + + +class FakeRollouter: + """The two fields of verl's FullyAsyncRollouter the skip touches.""" + + def __init__(self, prompts, epochs=2): + self.prompts, self.epochs, self.global_steps = list(prompts), epochs, 1 + + def _create_continuous_iterator(self): + for epoch in range(self.epochs): + for p in self.prompts: + yield epoch, p + + +def test_skip_draws_through_the_same_iterator_and_counts_steps(): + r = FakeRollouter("abcd") + assert fac.skip_rollouter_prompts(r, 6) == {"sample_offset": 6} + assert list(r._create_continuous_iterator()) == [(1, "c"), (1, "d")] # across the epoch end + assert r.global_steps == 7 + with pytest.raises(RuntimeError, match="already skipped"): + fac.skip_rollouter_prompts(r, 1) + short = FakeRollouter("ab", epochs=1) + fac.skip_rollouter_prompts(short, 3) + with pytest.raises(RuntimeError, match="data ended after 2 of 3"): + list(short._create_continuous_iterator()) + + +def test_data_cursor_seek_rules(): + seeks = [] + cursor = fac.FullyAsyncDataCursor(lambda n: seeks.append(n) or {"sample_offset": n}) + assert cursor.seek({"sample_offset": 0}) == {"sample_offset": 0} and seeks == [] + assert cursor.seek({"sample_offset": 12}) == {"sample_offset": 12} and seeks == [12] + with pytest.raises(ValueError, match="cannot move back"): + cursor.seek({"sample_offset": 4}) + with pytest.raises(ValueError, match="only"): + cursor.seek({"sample_offset": 12, "epoch": 0}) + cursor.started = True + with pytest.raises(ValueError, match="before the rollouter starts"): + cursor.seek({"sample_offset": 20}) + with pytest.raises(ValueError, match="no seek hook"): + fac.FullyAsyncDataCursor(None).seek({"sample_offset": 1}) + + +def test_optimizer_steps_per_round(): + cfg = {"actor_rollout_ref": {"actor": {"ppo_epochs": 2, "ppo_mini_batch_size": 4}}} + assert fac.optimizer_steps_per_round(cfg, groups_per_round=8) == 4 + with pytest.raises(ValueError, match="whole number"): + fac.optimizer_steps_per_round(cfg, groups_per_round=6) + + +def test_pool_cursor_counts_every_pulled_prompt_kept_or_dropped(tmp_path): + queue = [fa._sample("a", 0, 0, [(0, 4)]), fa._sample("b", 0, 0, [(0, 3)]), + fa._sample("c", 0, 1, [(0, 2), (1, 3)]), fa._sample("d", 1, 1, [(1, 4)]), + fa._sample("e", 0, 0, [(0, 5)]), # round 2: over age, dropped -- still consumed + fa._sample("f", 1, 2, [(1, 1), (2, 2)]), fa._sample("g", 2, 2, [(2, 6)])] + actor = fa.FakeTrainerActor(tmp_path / "rb", queue) + driver, island, _tape, _started = fa._driver(tmp_path, actor) + driver.run() + assert driver.rollout.data_cursor() == {"sample_offset": 7} + with pytest.raises(ValueError, match="before the rollouter starts"): + driver.rollout.seek_data_cursor({"sample_offset": 9}) + + +def test_pool_seek_before_the_start_skips_the_rollouter(tmp_path): + from yeto.rl.adapters.verl.fully_async_ports import FullyAsyncRolloutPool + + rollouter = FakeRollouter("abcdef") + events = [] + island = SimpleNamespace(required=2, rollouter_started=False, + seek_rollouter=lambda n: fac.skip_rollouter_prompts(rollouter, n), + emit=lambda event, **f: events.append((event, f))) + pool = FullyAsyncRolloutPool(island) + assert pool.seek_data_cursor({"sample_offset": 4}) == {"sample_offset": 4} + assert [p for _, p in rollouter._create_continuous_iterator()][:2] == ["e", "f"] + assert events == [("rl_verl_data_seek", {"data_cursor": {"sample_offset": 4}, "fully_async": True})] + + +def test_trainer_group_cuts_and_restores_through_the_worker_group(tmp_path): + """FullyAsyncTrainerGroup's cut verbs reach the ranks through call_workers (the + trainer actor's yeto_call_workers) and keep the sync path's checks.""" + from yeto.rl.adapters.verl.fully_async_ports import FullyAsyncTrainerGroup + + def group(handles): + worker = SimpleNamespace(module=handles.module, optimizer=handles.optimizer, + lr_scheduler=handles.scheduler) + island = SimpleNamespace(call_workers=lambda fn, *a: [fn(worker, *a)], optimizer_steps_per_round=vc.K, + emit=lambda *a, **k: None) + return FullyAsyncTrainerGroup(island) + + trained = vc._handles(vc._model()) + for r in range(vc.CUT): + vc._round(trained, r) + assert group(trained).save_cut(epoch=1, context=vc._context(tmp_path)) == "cut-v" + fresh = vc._handles(vc._model()) + port = group(fresh) + assert port.layout() == vc.vcut.layout_for(1) + port.restore_cut("cut-v", epoch=1, root=str(tmp_path), expect=vc._expect()) + port.confirm_resumed_data_cursor({"sample_offset": vc.PROMPTS_PER_ROUND * vc.CUT}) + vc._round(fresh, vc.CUT) + events = [] + port._check_first_step_after_resume(vc.CUT, lambda event, **f: events.append(event)) + assert events == ["rl_resume_check"] + vc._round(trained, vc.CUT) + vc._assert_same_train_state(trained, fresh) + + +def test_trainer_group_without_cuts_refuses_to_cut(): + from yeto.rl.adapters.verl.fully_async_ports import FullyAsyncTrainerGroup + + port = FullyAsyncTrainerGroup(SimpleNamespace(call_workers=lambda fn, *a: [], optimizer_steps_per_round=None)) + with pytest.raises(RuntimeError, match="without round cuts"): + port._cuts() + + +def test_a_fully_async_round_cut_takes_the_pool_with_a_policy_age(tmp_path): + """S26 E9 G2 (e9-g2-b): with --rl-max-policy-age > 0 the cut asks the rollout + pool for its in-flight work; the fully_async pool had no export, so every cut + was refused (RebuildRefused). It now exports nothing (queued groups are + generated again after a resume) and the cut context is built.""" + from yeto.rl.adapters.verl.fully_async_ports import FullyAsyncRolloutPool + from yeto.rl.engine.algorithm import AlgorithmSpec + from yeto.rl.engine.round_cut import CutSource + + island = SimpleNamespace(required=2, rollouter_started=True, seek_rollouter=None, + emit=lambda event, **f: None) + pool = FullyAsyncRolloutPool(island) + pool._cursor.consume(5) + state = SimpleNamespace(policy_version=3, policy_tensor_hash=lambda: "h" * 64) + driver = SimpleNamespace(published_state=state, published_version=3, local_step=3, + profile=SimpleNamespace(max_policy_age=2), at_safe_point=True, expected_token="yeto:3:h") + source = CutSource(driver=driver, trainer=None, rollout=pool, ledger=None, algorithm=AlgorithmSpec(), + backend_fingerprint="sha256:" + "0" * 64, cut_root="", global_batch_size=1) + context = source.context("r000003-x") + assert context.data == {"sample_offset": 5, "buffer_length": 0} + assert tuple(context.in_flight) == () and context.ledger["max_policy_age"] == 2 diff --git a/tests/test_s26_e9_verl_multigpu.py b/tests/test_s26_e9_verl_multigpu.py new file mode 100644 index 00000000..17d569c1 --- /dev/null +++ b/tests/test_s26_e9_verl_multigpu.py @@ -0,0 +1,92 @@ +"""S26 E9 G3: a verl sync island uses every card of its container; a multi-card +verl island on Modal runs NCCL with the shared-memory transport off.""" +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from yeto.gpu_spec import parse_gpu_spec +from yeto.launch import modal_islands +from yeto.rl.adapters.verl import config as vconf +from yeto.rl.adapters.verl.island_entry import island_gpus + + +def _run(**kw) -> vconf.VerlRunConfig: + return vconf.VerlRunConfig(model_path="/m", train_file="/t", val_file="/v", out_dir="/o", **kw) + + +def test_two_card_sync_island_trains_on_both_ranks(): + overrides = vconf.build_overrides(_run(n_gpus=2)) + assert "trainer.n_gpus_per_node=2" in overrides and "trainer.nnodes=1" in overrides + + +def test_zero_cards_is_refused(): + with pytest.raises(vconf.VerlConfigError, match="n_gpus"): + vconf.validate(_run(n_gpus=0)) + + +@pytest.mark.parametrize("info, want", [ + ({"before": {"CPU": 32.0, "GPU": 2.0}}, 2), + ({"before": {"CPU": 8.0, "GPU": 2.0}, "after": {"CPU": 32.0, "GPU": 2.0}}, 2), + ({"before": {"CPU": 32.0}}, 1), # no GPU resource reported + ({"skipped": "no RAY_ADDRESS"}, 1), # no island-local Ray +]) +def test_island_gpus_reads_the_island_ray(info, want): + assert island_gpus(info) == want + + +@pytest.mark.parametrize("backend, gpu, want", [ + ("verl", "modal:2xa10g", True), + ("verl", "modal:1xa10g", False), + ("miles", "modal:2xa10g", False), + (None, "modal:2xa10g", False), # default backend is Miles +]) +def test_nccl_shm_off_only_for_multi_card_verl(backend, gpu, want): + (spec,) = parse_gpu_spec(gpu) + assert modal_islands.verl_needs_nccl_shm_off(SimpleNamespace(rl_backend=backend), spec) is want + + +@pytest.mark.parametrize("extra, want", [ + ((), "1"), + (("--modal-env", "NCCL_SHM_DISABLE=0"), "0"), # an explicit --modal-env wins +]) +def test_modal_island_config_carries_the_default(monkeypatch, tmp_path, extra, want): + from yeto import cli + + monkeypatch.setenv("HOME", str(tmp_path)) + args = cli.parse_args(["--training-mode", "rl", "--rl-backend", "verl", "--gpu", "modal:2xa10g", + "--cluster-prefix", "e9g3", "--model", "Qwen/Qwen3-0.6B", "--data", "d", *extra]) + args.rl_image = "verl-build:" + "a" * 40 # what resolve_default_rl_image sets for verl + (spec,) = parse_gpu_spec(args.gpu) + task = SimpleNamespace(envs={}, secrets={}, file_mounts={}, run="true", setup="") + cfg = modal_islands.build_modal_island_config(args, spec, 0, task, "none") + assert cfg.envs["NCCL_SHM_DISABLE"] == want + + +def _write_readbacks(root: str, n: int) -> None: + import os + + from yeto.rl.adapters.verl import vllm_readback as rb + + os.environ[rb.READBACK_ENV] = root + for _ in range(n): + rb.after_add_lora(SimpleNamespace(), 1) # a broken manager: the error record is written + + +def test_readback_from_several_vllm_replicas_at_once(tmp_path): + """e9b-g3-a: two vLLM replicas (one per card) wrote readback-v0.json through one + shared tmp name; the second os.replace raised FileNotFoundError inside vLLM.""" + import json + import multiprocessing + + (tmp_path / "expected_version").write_text("0") + ctx = multiprocessing.get_context("fork") + procs = [ctx.Process(target=_write_readbacks, args=(str(tmp_path), 200)) for _ in range(2)] + for p in procs: + p.start() + for p in procs: + p.join(60) + assert [p.exitcode for p in procs] == [0, 0] + assert json.loads((tmp_path / "readback-v0.json").read_text())["version"] == "0" + assert not list(tmp_path.glob("*.tmp")) diff --git a/tests/test_s26_e9_verl_pair_launch.py b/tests/test_s26_e9_verl_pair_launch.py new file mode 100644 index 00000000..533ab9c8 --- /dev/null +++ b/tests/test_s26_e9_verl_pair_launch.py @@ -0,0 +1,287 @@ +"""S26 E9.4: one launch starts a verl trainer island and its sampler islands (dry run). + +``yeto launch --rl-backend verl --rl-island-sample-role ...`` builds every island +with the verl island entry; a sampler island runs the engine sampler on the verl +image's vLLM (:mod:`yeto.rl.adapters.verl.sampler_entry`). No cloud, no GPU.""" + +from __future__ import annotations + +import shlex +import sys +import types + +import pytest + +from tests.test_rl_engine_selection import _cli +from yeto.rl.adapters.verl import island_entry, sampler_entry +from yeto.rl.engine import sample_roles + +STORE = "modal-volume://yeto-e9/run" + + +def _tasks(monkeypatch, extra, count): + from tests.test_rl_launcher import _Resources, _Storage, _StorageMode, _Task + from yeto import launcher + from yeto.gpu_spec import parse_gpu_spec + from yeto.launch.rl_island_task import make_miles_island_task + + monkeypatch.setitem(sys.modules, "sky", types.SimpleNamespace( + Task=_Task, Resources=_Resources, Storage=_Storage, StorageMode=_StorageMode)) + args = _cli(tuple(extra)) + launcher._prepare_rl_args(args) + args.model_revision, args.data_revision = "a" * 40, "b" * 40 + args.source_sha256, args.reward_sha256 = "c" * 64, "d" * 64 + specs = parse_gpu_spec(args.gpu) + out = [] + for m in range(count): + task = launcher._island_task(make_miles_island_task, args, specs[m], m, count, "127.0.0.1:1", {}) + line = next(l for l in task.run.splitlines() if "-m yeto.rl.adapters.verl.island_entry" in l) + argv = shlex.split(line.split("-m yeto.rl.adapters.verl.island_entry", 1)[1] + .replace("$LEARNER_ID", str(m)).replace("$SYNCER_ADDR", "127.0.0.1:1")) + out.append(island_entry.parse_args(argv)[0]) + return args, out + + +def test_synced_verl_trainers_and_a_separate_sampler_launch(monkeypatch): + """Multi-island sync (S25 C3V shape): N synced trainers read lanes of one store and + learner 0 releases; a sampler launch over the same store runs the engine sampler. + A relaunched publisher (learner 0) gets the same flags (the fleet controller re-runs + the same task), so it re-joins as the publisher.""" + from yeto.launch.modal_islands import effective_recover_timeout + from yeto.rl.engine.merged_release import PUBLISHER_LEARNER_ID + + monkeypatch.setenv("YETO_ISLAND_HMAC_KEY", "k" * 64) # test key; elastic islands sign messages + common = ["--rl-backend", "verl", "--rl-checkpoint-store", STORE, "--rl-max-policy-age", "2", + "--reward-function", "gsm8k_reward:score", "--total-steps", "4"] + targs, trainers = _tasks(monkeypatch, common + ["--gpu", "modal:1xa10g,modal:1xa10g", + "--rl-sample-role", "trainer", + "--rl-island-scheduling", "elastic"], 2) + for m, t in enumerate(trainers): + assert t.rl_sample_role == "trainer" and t.rl_sample_lanes == 2 and t.rl_sample_store + assert t.rl_resume_store is None and t.learner_id == m + assert island_entry.sample_buffer_section(t)["lane"] == f"{m}/2" + assert PUBLISHER_LEARNER_ID == 0 + assert effective_recover_timeout(targs) > 0 # relaunch allowed: the publisher can be replaced + _sargs, (sampler,) = _tasks(monkeypatch, common + ["--gpu", "modal:1xa10g", "--rl-sample-role", "sampler", + "--rl-single-island-no-sync"], 1) + assert sampler.rl_sample_role == "sampler" and sampler.rl_resume_store == trainers[0].rl_sample_store + spec = sampler_entry.sampler_spec(sampler, model_path="/m", prompts_path="/p", card_model="A10G", + engine_version="v", event_tape="/t") + assert spec["sampler_id"] == "island0" and spec["store"] == trainers[0].rl_sample_store + + +def _islands(monkeypatch, roles): + from tests.test_rl_launcher import _Resources, _Storage, _StorageMode, _Task + from yeto import launcher + from yeto.gpu_spec import parse_gpu_spec + from yeto.launch.rl_island_task import make_miles_island_task + + monkeypatch.setitem(sys.modules, "sky", types.SimpleNamespace( + Task=_Task, Resources=_Resources, Storage=_Storage, StorageMode=_StorageMode)) + extra = ["--rl-backend", "verl", "--rl-single-island-no-sync", "--rl-checkpoint-store", STORE, + "--rl-max-policy-age", "2", "--gpu", ",".join(["modal:1xa10g"] * len(roles)), + "--reward-function", "gsm8k_reward:score", "--total-steps", "4"] + for m, role in enumerate(roles): + extra += ["--rl-island-sample-role", f"{m}:{role}"] + args = _cli(tuple(extra)) + launcher._prepare_rl_args(args) + args.model_revision, args.data_revision = "a" * 40, "b" * 40 + args.source_sha256, args.reward_sha256 = "c" * 64, "d" * 64 + specs = parse_gpu_spec(args.gpu) + out = [] + for m in range(len(roles)): + task = launcher._island_task(make_miles_island_task, args, specs[m], m, len(roles), "none", {}) + line = next(l for l in task.run.splitlines() if "-m yeto.rl.adapters.verl.island_entry" in l) + argv = shlex.split(line.split("-m yeto.rl.adapters.verl.island_entry", 1)[1].replace("$LEARNER_ID", "0")) + parsed, _ignored = island_entry.parse_args(argv) + out.append((parsed, task.envs)) + return out + + +def test_one_launch_gives_one_sampler_and_one_trainer(monkeypatch): + (sampler, s_env), (trainer, _t_env) = _islands(monkeypatch, ["sampler", "trainer"]) + assert (sampler.rl_sample_role, trainer.rl_sample_role) == ("sampler", "trainer") + assert sampler.rl_sampler_node == "island0" and sampler.rl_resume_store == trainer.rl_resume_store + assert island_entry.sample_buffer_section(sampler) is None # main() runs the engine sampler + section = island_entry.sample_buffer_section(trainer) + assert section["store"] == trainer.rl_resume_store and "shards" not in section + assert s_env.get("YETO_SPOT_RECLAIM_MARKER") # the reclaim signal reaches the sampler + spec = sampler_entry.sampler_spec(sampler, model_path="/m", prompts_path="/p.jsonl", card_model="A10G", + engine_version="0.x", event_tape="/t/events.jsonl", python="python3") + assert spec["sampler_id"] == "island0" and spec["store"] == sampler.rl_resume_store + assert spec["island_id"] == 0 # the launcher reads the sampler tape as island tape + assert spec["engine"]["adapter"] == "vllm-server" and spec["num_rollouts"] == 4 + assert spec["max_backlog_groups"] == 4 * sample_roles.DEFAULT_MAX_BACKLOG_ROUNDS + assert spec["reward"] == "yeto.rl.gsm8k_reward:score" and "buffer_shards" not in spec + from yeto.rl.engine.engine_sampler_entry import island_config + + assert island_config(spec).sampler_id == "island0" # the entry accepts the spec + from yeto.rl.adapters.sampler_engines import from_spec + + assert spec["engine"]["factory"] == from_spec.FACTORY + + +def test_two_samplers_get_shards_and_disjoint_data_slots(monkeypatch): + (s0, _), (s1, _), (trainer, _) = _islands(monkeypatch, ["sampler", "sampler", "trainer"]) + assert (s0.rl_buffer_shards, s1.rl_buffer_shards) == (["island0", "island1"],) * 2 + assert (s0.rl_sampler_data_slot, s1.rl_sampler_data_slot) == ("0/2", "1/2") + assert (s0.global_rounds, s1.global_rounds, trainer.global_rounds) == (2, 2, 4) # the samplers share the run + spec = sampler_entry.sampler_spec(s1, model_path="/m", prompts_path="/p.jsonl", card_model="A10G", + engine_version="0.x", event_tape="/t/e.jsonl") + assert spec["buffer_shards"] == ["island0", "island1"] and spec["prompts"]["data_slot"] == "1/2" + assert island_entry.sample_buffer_section(trainer)["shards"] == ["island0", "island1"] + + +def test_sampler_spec_refuses_missing_flags_and_unmapped_rewards(): + base = dict(rl_sampler_node="island0", rl_resume_store="/s", learner_id=3, groups_per_round=2, rollout_max_response_len=8, + samples_per_group=2, seed=1, global_rounds=2, reward_function="gsm8k_reward:score") + kw = dict(model_path="/m", prompts_path="/p", card_model="c", engine_version="v", event_tape="/t") + with pytest.raises(sampler_entry.SamplerIslandError, match="resume-store"): + sampler_entry.sampler_spec(types.SimpleNamespace(**{**base, "rl_resume_store": None}), **kw) + # a separate sampler launch (no pair): the Miles sampler's id + assert sampler_entry.sampler_spec(types.SimpleNamespace(**{**base, "rl_sampler_node": None}), + **kw)["sampler_id"] == "island3" + with pytest.raises(sampler_entry.SamplerIslandError, match="no engine sampler reward"): + sampler_entry.sampler_spec(types.SimpleNamespace(**{**base, "reward_function": "x:y"}), **kw) + + +def test_prompt_rows_render_messages(): + rows = [{"messages": [{"role": "user", "content": "1+1?", "extra": 0}], "label": 2}] + out = sampler_entry.prompt_rows(rows, lambda m: "|".join(x["content"] for x in m)) + assert out == [{"text": "1+1?", "label": "2"}] + with pytest.raises(sampler_entry.SamplerIslandError): + sampler_entry.prompt_rows([], str) + + +def test_engine_sampler_writes_its_named_shard(tmp_path): + from yeto.rl.engine.engine_sampler_entry import SamplerSpecError, build_stores + + _releases, buffer = build_stores({"store": str(tmp_path), "sampler_id": "island1", + "buffer_shards": ["island0", "island1"]}) + assert buffer.shard == "island1" + with pytest.raises(SamplerSpecError, match="not one of the buffer shards"): + build_stores({"store": str(tmp_path), "sampler_id": "island9", "buffer_shards": ["island0"]}) + + +def test_trainer_buffer_merges_named_shards(tmp_path): + from yeto.rl.engine.resume import store_for + from yeto.rl.engine.sample_buffer import open_trainer_buffer + + buffer = open_trainer_buffer(store_for(str(tmp_path)), batch_size=1, shards=["island0", "island1"]) + assert buffer._merge and set(buffer._expected_shards) == {"island0", "island1"} + plain = open_trainer_buffer(store_for(str(tmp_path / "one")), batch_size=1) + assert not plain._merge and plain._expected_shards is None + + +def _group(monkeypatch, gpus, samplers, total_steps=4, scheduling="elastic"): + """One launch of synced trainers plus sampler islands (S26 E9.4 synced group).""" + from tests.test_rl_launcher import _Resources, _Storage, _StorageMode, _Task + from yeto import launcher + from yeto.gpu_spec import parse_gpu_spec + from yeto.launch.rl_island_task import make_miles_island_task + + monkeypatch.setenv("YETO_ISLAND_HMAC_KEY", "k" * 64) # test key; elastic islands sign messages + monkeypatch.setitem(sys.modules, "sky", types.SimpleNamespace( + Task=_Task, Resources=_Resources, Storage=_Storage, StorageMode=_StorageMode)) + extra = ["--rl-backend", "verl", "--rl-checkpoint-store", STORE, "--rl-max-policy-age", "2", + "--reward-function", "gsm8k_reward:score", "--total-steps", str(total_steps), + "--gpu", ",".join(["modal:1xa10g"] * gpus), "--rl-sample-role", "trainer", + "--rl-island-scheduling", scheduling] + for m in samplers: + extra += ["--rl-island-sample-role", f"{m}:sampler"] + args = _cli(tuple(extra)) + launcher._prepare_rl_args(args) + args.model_revision, args.data_revision = "a" * 40, "b" * 40 + args.source_sha256, args.reward_sha256 = "c" * 64, "d" * 64 + specs = parse_gpu_spec(args.gpu) + learners = sample_roles.syncer_learner_count(args, len(specs)) + out = [] + for m in range(len(specs)): + task = launcher._island_task(make_miles_island_task, args, specs[m], m, learners, "127.0.0.1:1", {}) + line = next(l for l in task.run.splitlines() if "-m yeto.rl.adapters.verl.island_entry" in l) + argv = shlex.split(line.split("-m yeto.rl.adapters.verl.island_entry", 1)[1] + .replace("$LEARNER_ID", task.envs["LEARNER_ID"]).replace("$SYNCER_ADDR", "127.0.0.1:1")) + out.append((island_entry.parse_args(argv)[0], task, line)) + return args, learners, out + + +def test_one_launch_starts_synced_trainers_and_their_sampler(monkeypatch): + """Known limit fixed: synced trainers and a sampler start from one command.""" + args, learners, ((t0, _, l0), (t1, _, l1), (s, s_task, s_line)) = _group(monkeypatch, 3, [2]) + assert sample_roles.is_synced_group(args) and not sample_roles.is_pair(args) + assert learners == 2 # the syncer waits for the two trainers only + for k, t in enumerate((t0, t1)): + assert t.rl_sample_role == "trainer" and t.learner_id == k and t.rl_sample_lanes == 2 + assert t.rl_sample_store and t.rl_resume_store is None + assert island_entry.sample_buffer_section(t)["lane"] == f"{k}/2" + assert "shards" not in island_entry.sample_buffer_section(t) # one sampler writes the index + assert "--rl-island-scheduling elastic" in l0 and "--syncer" in l1 + assert s.rl_sample_role == "sampler" and s.rl_sampler_node == "island2" + assert s.rl_resume_store == t0.rl_sample_store # the same store + assert "--rl-single-island-no-sync" in s_line and "--rl-island-scheduling" not in s_line + assert s_task.envs["LEARNER_ID"] == "0" and s_task.envs.get("YETO_SPOT_RECLAIM_MARKER") + spec = sampler_entry.sampler_spec(s, model_path="/m", prompts_path="/p", card_model="A10G", + engine_version="v", event_tape="/t") + assert spec["num_rollouts"] == 8 # the rounds of both trainers (2 x 4) + + +@pytest.mark.parametrize("scheduling", ["elastic", "legacy"]) +def test_synced_group_quorum_counts_trainers_only(monkeypatch, scheduling): + """e9b-sync-b: quorum 3 (every island) over --learners 2 made the syncer exit.""" + args, learners, _ = _group(monkeypatch, 3, [2], scheduling=scheduling) + assert learners == 2 and args.quorum == 2 + + +def test_synced_group_with_two_samplers_shards_and_slots(monkeypatch): + _args, learners, islands = _group(monkeypatch, 4, [0, 3]) + (s0, _, _), (t0, _, _), (t1, _, _), (s1, _, _) = islands + assert learners == 2 and (t0.learner_id, t1.learner_id) == (0, 1) + assert island_entry.sample_buffer_section(t0)["shards"] == ["island0", "island3"] + assert (s0.rl_sampler_data_slot, s1.rl_sampler_data_slot) == ("0/2", "1/2") + for s in (s0, s1): + spec = sampler_entry.sampler_spec(s, model_path="/m", prompts_path="/p", card_model="A10G", + engine_version="v", event_tape="/t") + assert spec["num_rollouts"] == 4 # 2 trainers x 4 rounds over 2 samplers + + +@pytest.mark.parametrize("gpus,samplers,steps,match", [ + (2, [1], 4, "at least two trainer"), + (3, [2], 3, None), # 2 x 3 rounds over one sampler: fine + (4, [2, 3], 3, None), # 2 x 3 = 6 over 2 samplers: fine +]) +def test_synced_group_launch_checks(monkeypatch, gpus, samplers, steps, match): + if match is None: + _group(monkeypatch, gpus, samplers, total_steps=steps) + return + with pytest.raises((ValueError, SystemExit), match=match): + _group(monkeypatch, gpus, samplers, total_steps=steps) + + +def test_synced_group_lists_samplers_only(): + args = types.SimpleNamespace(rl_island_sample_role=["0:trainer", "2:sampler"], rl_sample_role="trainer", + rl_single_island_no_sync=False, tuning="lora", total_steps=4, + rl_node_class=None) + with pytest.raises(ValueError, match="sampler islands only"): + sample_roles.check_synced_group(args, 3) + + +def test_verl_trainer_buffer_merges_shards_without_a_sampler_index(tmp_path): + """G1 (e9-g1-d) root cause: the verl buffer trainer opens its buffer with + create_index=False; with named shards no sampler writes index.jsonl, so the + merging trainer must create it (it is the index's only writer).""" + from tests.test_rl_spot_sampler_replace import _cursor, _round + from yeto.rl.adapters.miles import buffer_rollout as br + from yeto.rl.engine.resume import CheckpointStore + from yeto.rl.engine.sample_buffer import FileSampleBuffer, open_trainer_buffer + + writer = br.SamplerRoundWriter(FileSampleBuffer(CheckpointStore(tmp_path), batch_size=1, + index="shards/island0.jsonl"), "island0", incarnation="a") + writer.write_round(0, _round(0, 3), _cursor(3)) + assert not (tmp_path / "sample-buffer/index.jsonl").exists() + trainer = open_trainer_buffer(CheckpointStore(tmp_path), batch_size=1, create_index=False, + shards=["island0", "island1"]) + assert len(trainer.select("t", 3)) == 3 + assert (tmp_path / "sample-buffer/index.jsonl").exists() + # a plain (single-sampler) reader still never creates the index + reader = open_trainer_buffer(CheckpointStore(tmp_path / "plain"), batch_size=1, create_index=False) + assert not reader._merge and not (tmp_path / "plain/sample-buffer/index.jsonl").exists() diff --git a/tools/eval/miles.py b/tools/eval/miles.py index 3f9538e6..401e3fae 100644 --- a/tools/eval/miles.py +++ b/tools/eval/miles.py @@ -30,9 +30,11 @@ def _gpu(cfg: dict, n: int) -> str: def _common(cfg: dict, run_id: str, prefix: str) -> str: - # --preflight-threads off: Modal islands put no SkyPilot threads on this host; the scripts - # check the host thread count (< 3000) before the launch instead - return (f"launch --training-mode rl --rl-engine ports --preflight-threads off " + # the launcher thread preflight stays on (default wait); a cell may loosen it only with + # explicit flags in cfg["preflight_args"] (e.g. "--preflight-thread-start 2950"), which the + # launcher prints and records in the run manifest + preflight = cfg.get("preflight_args", "") + return (f"launch --training-mode rl --rl-engine ports {preflight + ' ' if preflight else ''}" f"--modal-gpu-exact --modal-retries 0 --modal-cpu 8 " f"--modal-memory 64 --modal-tape-volume {TAPE_VOLUME} --controller local --cluster-prefix {prefix} " f"--tuning lora --lora-r {cfg['lora_r']} --rl-observe-timeline {MODEL} {DATA} " diff --git a/tools/eval/miles_head.sh b/tools/eval/miles_head.sh index dac65373..e4ac5b03 100755 --- a/tools/eval/miles_head.sh +++ b/tools/eval/miles_head.sh @@ -8,6 +8,9 @@ set -u : "${P:?}" "${R:?}" "${REPO:?}" "${PY:?}" "${SKY:?}" HARD=${HARD:-2700}; WD=$(( HARD + 300 )); STEPS=${STEPS:-4}; SROUNDS=$(( 2 * STEPS + 6 )); GPU=${GPU:-a10g} +# launcher thread preflight stays on; loosen only with explicit flags, e.g. +# PREFLIGHT="--preflight-thread-start 2950" (the launcher logs and records the loosened values) +PREFLIGHT=${PREFLIGHT:-} SHA=${SHA:-HEAD}; MODAL="$PY -m modal" APP=yeto-$P; SP=$P-s; SAPP=yeto-$SP; TAPEVOL=yeto-event-tapes; STOREVOL=${STOREVOL:-yeto-eval-store}; HEAD_REGION=nebius/eu-north1 export CARGO_HOME=/home/michael/work/gpu-default-modal/cargo RUSTUP_HOME=/home/michael/work/gpu-default-modal/rustup @@ -30,8 +33,8 @@ LORA="--tuning lora --lora-r 16" MODALX="--modal-gpu-exact --modal-retries 0 --modal-cpu 8 --modal-memory 64 --modal-tape-volume $TAPEVOL" STORE="--rl-checkpoint-store modal-volume://$STOREVOL/$P" COMMON="--training-mode rl --rl-engine ports $MODALX $MODEL $DATA $LORA $STORE --rl-algorithm-spec examples/rl_algorithms/decoupled-ppo-c2.json --rl-max-policy-age 2 --rl-observe-timeline --reward-function gsm8k_reward:score --fragments 1 --pipeline 1 --rollout-batch-size 32 --n-samples-per-prompt 4 --rollout-max-response-len 1024 --seq-len 2048 --inner-lr 1e-5 --seed 17 --apply-chat-template-kwargs '{\"enable_thinking\": false}' --trust-remote-code" -TARGS="launch --controller head --syncer-region $HEAD_REGION --preflight-threads off --rl-stall-timeout 2400 --on-demand --no-island-relaunch --modal-timeout-s $HARD --cluster-prefix $P --gpu modal:1x$GPU,modal:1x$GPU --rl-island-scheduling elastic --rl-sample-role trainer --rl-sample-wait-s 900 $COMMON --total-steps $STEPS" -SARGS="launch --preflight-threads off --controller local --cluster-prefix $SP --gpu modal:1x$GPU --rl-single-island-no-sync --rl-sample-role sampler --rl-sampler-max-backlog-rounds 4 $COMMON --total-steps $SROUNDS" +TARGS="launch --controller head --syncer-region $HEAD_REGION $PREFLIGHT --rl-stall-timeout 2400 --on-demand --no-island-relaunch --modal-timeout-s $HARD --cluster-prefix $P --gpu modal:1x$GPU,modal:1x$GPU --rl-island-scheduling elastic --rl-sample-role trainer --rl-sample-wait-s 900 $COMMON --total-steps $STEPS" +SARGS="launch $PREFLIGHT --controller local --cluster-prefix $SP --gpu modal:1x$GPU --rl-single-island-no-sync --rl-sample-role sampler --rl-sampler-max-backlog-rounds 4 $COMMON --total-steps $SROUNDS" git -C $REPO rev-parse $SHA > $R/yeto_sha.txt; echo "$TARGS" > $R/args.txt; echo "$SARGS" > $R/sargs.txt echo "$APP $SAPP $HEAD_CL" > $R/resources.txt eval "$(/usr/bin/python3 - <<'PYC' diff --git a/yeto/cli.py b/yeto/cli.py index 7c68497f..62f8c050 100644 --- a/yeto/cli.py +++ b/yeto/cli.py @@ -1538,9 +1538,13 @@ def cmd_head(payload: str) -> int: from .rl.sao_streaming_manifest import load_sao_streaming_manifest args.rl_sao_manifest = load_sao_streaming_manifest(args.rl_sao_contracts) - # Same count launcher.run uses: --gpu islands plus --external-learners + # Same count launcher.run uses: the synced islands (a synced group's sampler + # islands do not join the syncer, S26 e9b-sync-a) plus --external-learners # seats; without the seats the syncer rejects every manual joiner. - num_learners = len(parse_gpu_spec(args.gpu)) + (getattr(args, "external_learners", 0) or 0) + from .rl.engine import sample_roles + + num_learners = (sample_roles.syncer_learner_count(args, len(parse_gpu_spec(args.gpu))) + + (getattr(args, "external_learners", 0) or 0)) syncer = launcher.LocalSyncer(args, num_learners) syncer.start() syncer.start_log_forwarder() diff --git a/yeto/launch/modal_islands.py b/yeto/launch/modal_islands.py index a7f12af6..89736be2 100644 --- a/yeto/launch/modal_islands.py +++ b/yeto/launch/modal_islands.py @@ -25,6 +25,19 @@ from .fleet import TEARDOWN_UNVERIFIED_EXIT +def verl_needs_nccl_shm_off(args, spec: ClusterSpec) -> bool: + """A verl island with several cards in one Modal container runs NCCL with + its shared-memory transport off. + + S26 E9 e9-g2-a: on a 2-card Modal container verl's NCCL checkpoint engine + failed attaching ``/dev/shm/nccl-*`` ("No such file or directory", NCCL + system error); the same run with ``NCCL_SHM_DISABLE=1`` passed (e9-g2-d/e). + NCCL then uses P2P/sockets between the two local ranks. Miles islands are + not changed (no failure seen there).""" + backend = getattr(args, "rl_backend", None) or "miles" + return backend == "verl" and int(spec.gpus_per_node or 1) > 1 + + def build_modal_island_config(args, spec: ClusterSpec, learner_id: int, task, syncer_addr: str): """Turn the sky task the factory built for this island into the Modal island config: same run script, same envs (syncer address swapped for @@ -50,6 +63,8 @@ def build_modal_island_config(args, spec: ClusterSpec, learner_id: int, task, sy # The critic syncer shares the actor syncer's host: swap it too, or a # Modal island behind --syncer-public-addr dials the private address. envs["CRITIC_SYNCER_ADDR"] = critic_syncer_address(syncer_addr) + if rl and verl_needs_nccl_shm_off(args, spec): + envs.setdefault("NCCL_SHM_DISABLE", "1") # --modal-env below still overrides it for item in getattr(args, "modal_env", None) or []: key, sep, value = str(item).partition("=") if not sep or not key: @@ -301,6 +316,23 @@ def container_set_poller_for(args, modal_ops, modal_cfgs, on_change, *, start: b return poller.start() if start else poller +def container_log_follower_for(args, modal_ops, modal_cfgs, route, *, poller, start: bool = True): + """The follower of rescheduled containers' logs (modal_runner.ContainerLogFollower), + only when a new container is expected (no failing container guard: ``poller`` + is None) and the ops can stream one container. None otherwise.""" + if poller is not None or modal_ops is None or not modal_cfgs: + return None + if not (callable(getattr(modal_ops, "list_container_ids", None)) + and callable(getattr(modal_ops, "stream_container_logs", None))): + return None + from ..modal_runner import ContainerLogFollower + + expected = sum(max(1, int(getattr(cfg, "num_nodes", 1) or 1)) for cfg in modal_cfgs.values()) + follower = ContainerLogFollower(modal_ops.list_container_ids, modal_ops.stream_container_logs, + expected, route=route) + return follower.start() if start else follower + + def effective_recover_timeout(args) -> float: """The fleet controller's learner relaunch budget. diff --git a/yeto/launch/rl_args.py b/yeto/launch/rl_args.py index 0c21669b..0906c6d0 100644 --- a/yeto/launch/rl_args.py +++ b/yeto/launch/rl_args.py @@ -148,6 +148,14 @@ def sao_contract_mounts(args, learner_id: int) -> dict[str, str]: for shipped in (island.runtime, island.layout_attestation)} +def _synced_learners(args, specs) -> int: + """Islands that join the syncer: a synced group's sampler islands do not + (S26 e9b-sync-b: quorum 3 over --learners 2 made the syncer exit).""" + from ..rl.engine.sample_roles import syncer_learner_count + + return syncer_learner_count(args, len(specs)) + + def _prepare_dense_full_sweep(args, *, num_islands: int) -> None: """full-parameter preset (P0-D8): one actor role swept as whole-model fragments. @@ -732,7 +740,7 @@ def _prepare_rl_args( raise ValueError("decoupled RL pipeline must be between 1 and fragments") if args.fragment_pattern != "binpack": raise ValueError("decoupled RL requires --fragment-pattern binpack") - args.quorum = len(specs) + args.quorum = _synced_learners(args, specs) args.grace_ms = 0 args.sync_interval_steps = float(args.local_rl_rounds_per_sync) args.delta_correction = "none" @@ -747,7 +755,7 @@ def _prepare_rl_args( _prepare_sao_streaming(args, num_islands=len(specs)) elif not args.experimental_rl_sync: args.fragments = 1 - args.quorum = len(specs) + args.quorum = _synced_learners(args, specs) args.grace_ms = 0 args.pipeline = 1 args.sync_interval_steps = 0.0 diff --git a/yeto/launch/rl_ports.py b/yeto/launch/rl_ports.py index 63b41f24..c2d7401f 100644 --- a/yeto/launch/rl_ports.py +++ b/yeto/launch/rl_ports.py @@ -566,7 +566,7 @@ def _ports_infra_flags(args) -> tuple[str, str]: if resume_store is not None and sample_role_is_synced_trainer(args): # S25 C3: synced trainers mount the store for the buffer and the releases only flags += (f" --rl-sample-store {shlex.quote(resume_store[0])}" - f" --rl-sample-lanes {rl_island_count(args)}") + f" --rl-sample-lanes {getattr(args, 'rl_synced_trainers', None) or rl_island_count(args)}") resume_store = None if resume_store is not None: # rl-resume-from-checkpoint: non-elastic resume store flags += f" --rl-resume-store {shlex.quote(resume_store[0])}" diff --git a/yeto/launch_preflight.py b/yeto/launch_preflight.py index 539b07a0..c173c4da 100644 --- a/yeto/launch_preflight.py +++ b/yeto/launch_preflight.py @@ -185,6 +185,24 @@ def thread_message(count: ThreadCount, cfg: ThreadConfig, headline: str) -> str: ]) +def thread_relaxations(cfg: ThreadConfig) -> dict: + """The thread settings that are looser than the defaults, as ``{name: [value, default]}``. + + The check is on by default; a run may only loosen it with an explicit flag, and the + launcher prints and records every loosened value (empty dict: defaults or stricter).""" + + relaxed = {} + if cfg.mode == "off": + relaxed["mode"] = ["off", DEFAULT_THREAD_MODE] + if cfg.start > DEFAULT_THREAD_START: + relaxed["start"] = [cfg.start, DEFAULT_THREAD_START] + if cfg.hard > DEFAULT_THREAD_HARD: + relaxed["hard"] = [cfg.hard, DEFAULT_THREAD_HARD] + if cfg.mode == "wait" and cfg.wait_s > DEFAULT_THREAD_WAIT_S: + relaxed["wait_s"] = [cfg.wait_s, DEFAULT_THREAD_WAIT_S] + return relaxed + + def thread_preflight(args, *, counter: Callable[[], ThreadCount] | None = None, clock: Callable[[], float] = time.monotonic, sleep: Callable[[float], None] = time.sleep, @@ -197,10 +215,15 @@ def thread_preflight(args, *, counter: Callable[[], ThreadCount] | None = None, out = sys.stderr if out is None else out cfg = ThreadConfig.from_args(args) + relaxed = thread_relaxations(cfg) + if relaxed: + print("[preflight] WARNING: thread preflight RELAXED by explicit flags: " + + ", ".join(f"{k} {v[0]} (default {v[1]})" for k, v in relaxed.items()), + file=out, flush=True) if cfg.mode == "off": print("[preflight] WARNING: thread preflight is OFF (--preflight-threads off); the " "s1run thread guard may tear down this run", file=out, flush=True) - return {"mode": "off", "disabled": True} + return {"mode": "off", "disabled": True, "relaxed": relaxed} counter = counter or count_threads started = clock() count = counter() @@ -233,7 +256,7 @@ def thread_preflight(args, *, counter: Callable[[], ThreadCount] | None = None, f"(SkyPilot API server {count.sky_api if count.sky_api_found else 'not found'})", file=out, flush=True) return {"mode": cfg.mode, "start": cfg.start, "hard": cfg.hard, "wait_s": cfg.wait_s, - "waited": waited, "readings": readings, **count.to_dict()} + "waited": waited, "readings": readings, "relaxed": relaxed, **count.to_dict()} # --------------------------------------------------------------------------- diff --git a/yeto/launcher.py b/yeto/launcher.py index 0abd2f4b..cb398506 100644 --- a/yeto/launcher.py +++ b/yeto/launcher.py @@ -60,7 +60,7 @@ ) from .launch.modal_islands import ( _modal_tape_run_dir, _tail_modal, _verify_modal_app_stopped, build_modal_island_config, - container_set_poller_for, effective_recover_timeout, pull_modal_tapes, + container_log_follower_for, container_set_poller_for, effective_recover_timeout, pull_modal_tapes, require_modal_for_gpu_exact, ) from .launch.fleet import ( @@ -184,15 +184,29 @@ def _island_task(task_factory, args, spec, m, num_learners, syncer_addr, overrid iargs = droppable.role_args(io.island_args(rl_island_args(args, m), m, overrides), role) # L5.4 pair: the island's sample role; it runs as learner 0 of 1 (no syncer). iargs = sample_roles.island_args(iargs, m, rl_island_count(args)) - learner_id, island_learners = sample_roles.island_learner(args, m, num_learners) + learner_id, island_learners = sample_roles.island_learner(args, m, num_learners, rl_island_count(args)) task = task_factory(iargs, spec, learner_id, island_learners, syncer_addr) io.apply_to_task(task, {**io.island_env(args, m, overrides), **(droppable.role_env(role, spec.cloud, spec.region) if role else {}), **sample_roles.island_env(iargs), - **_spot_sampler_cloud_env(iargs, getattr(spec, "cloud", None))}) + **_spot_sampler_cloud_env(iargs, getattr(spec, "cloud", None)), + **_resume_marker_env(iargs, getattr(spec, "cloud", None))}) return task +def _resume_marker_env(iargs, cloud: str | None) -> dict[str, str]: + """A resumable island on Modal (``--rl-checkpoint-store`` without + ``--rl-elastic``: the island's ``--rl-resume-store``): the reclaim marker path + (S26 E9.1). Modal signals only the container's main process (the runner), so + the runner writes the marker and the driver's watcher thread -- in a Ray actor + on verl -- writes the store's preemption notice.""" + if cloud != "modal" or not getattr(iargs, "rl_checkpoint_store", None) or getattr(iargs, "rl_elastic", False): + return {} + from .rl.engine.sampler_resume import DEFAULT_RECLAIM_MARKER, RECLAIM_MARKER_ENV + + return {RECLAIM_MARKER_ENV: DEFAULT_RECLAIM_MARKER} + + def _spot_sampler_cloud_env(iargs, cloud: str | None) -> dict[str, str]: """A spot sampler island on Nebius: the VM stop reaches the sampler through the host unit of ``yeto.cloud.nebius_reclaim`` (Modal signals the container itself).""" @@ -304,7 +318,10 @@ def run(args, on_clusters=None, local_syncer=None, on_instance_ids=None) -> int: # counts them in --learners and its port is already public, so they # simply dial in with the printed join command. external = max(0, getattr(args, "external_learners", 0) or 0) - num_learners = len(specs) + external + from .rl.engine import sample_roles + + # S26 E9.4: the sampler islands of a synced group are no learners of the syncer + num_learners = sample_roles.syncer_learner_count(args, len(specs)) + external require_modal_for_gpu_exact(args, specs) warn_if_model_wont_fit(args, specs) prefix = args.cluster_prefix @@ -618,6 +635,17 @@ def container_changed() -> bool: if container_poller is not None: container_guards.append(container_poller) + def route_container(learner_id: int): + # the island of a rescheduled container's lines (unique learner id only: + # the islands of a pair all run as learner 0) + names = [n for n, cfg in modal_cfgs.items() if int(cfg.learner_id) == int(learner_id)] + if len(names) != 1: + return None + return names[0], (event_collectors.get(names[0]) if names[0] in echo_names else None) + + # daemon threads; each container's log ends when the app stops + container_log_follower_for(args, modal_ops, modal_cfgs, route_container, poller=container_poller) + from . import runs from .modal_runner import RoutingOps @@ -990,13 +1018,15 @@ def dry_run_plan(args) -> dict: if rl: role = _island_role(args, specs, learner_id) task = _island_task(make_miles_island_task, args, spec, learner_id, - len(specs) + external, "$SYNCER_ADDR", island_overrides, role) + sample_roles.syncer_learner_count(args, len(specs)) + external, + "$SYNCER_ADDR", island_overrides, role) if role is not None: from .cloud.droppable import billing, rejoin_source entry["island_role"] = {"role": role, "billing": billing(role), "rejoin": rejoin_source(role)} - if sample_roles.is_store_coupled(args): # L5.4/L7.5: per-island role, learner 0 of 1 + if sample_roles.is_store_coupled(args) or sample_roles.is_synced_group(args): + # L5.4/L7.5: per-island role, learner 0 of 1; S26 E9.4: synced group entry["sample_role"] = sample_roles.island_args( args, learner_id, len(specs)).rl_sample_role if learner_id == trainer_loop.held_island(args): @@ -1024,7 +1054,7 @@ def dry_run_plan(args) -> dict: "syncer": None if no_sync else ("head VM" if head else f"{args.cluster_prefix}-syncer"), # P0-D8: one syncer per trainable role (actor first); none without outer sync "role_syncers": [] if not rl or no_sync else _dry_run_role_syncers( - args, len(specs) + external), + args, sample_roles.syncer_learner_count(args, len(specs)) + external), "outer_sync": not no_sync, # rl-trainer-start-stop L5.3b/L5.4: the head's scheduler and the island it owns "trainer_scheduler": (None if not trainer_loop.enabled(args) else { diff --git a/yeto/modal_runner.py b/yeto/modal_runner.py index b25be3c4..870414d3 100644 --- a/yeto/modal_runner.py +++ b/yeto/modal_runner.py @@ -951,22 +951,124 @@ def _run_forwarding_to_learner(cmd: list[str], env: dict, *, popen, sig, clock=t marker file the learner's watcher thread picks up. Nothing is signalled: the learner and the Ray processes keep running until the platform kills the container, so the rollout executor can answer the in-flight export.""" - proc = popen(cmd, env=env) path = env.get("YETO_SPOT_RECLAIM_MARKER") or "/tmp/yeto-reclaim-requested" + rank, node_ips = _node_rank_and_ips(env) + relay = None + if rank == 0 and len(node_ips) > 1: + # S26 E9.1: the driver runs on rank 0; a reclaim of a rank>0 container reaches + # it only through this relay (the peer runner sends its signal over the cluster net) + relay = ReclaimRelay(lambda num, peer: _write_reclaim_marker(path, num, clock, from_rank=peer), + port=reclaim_relay_port(env)).start() + proc = popen(cmd, env=env) def forward(num, _frame): - try: - tmp = f"{path}.tmp" - with open(tmp, "w", encoding="utf-8") as fh: - fh.write(json.dumps({"signal": int(num), "time": clock()}) + "\n") - os.replace(tmp, path) - print(f"[modal-island] signal {num}: reclaim marker {path} written for the learner", flush=True) - except OSError as exc: - print(f"[modal-island] signal {num}: reclaim marker not written ({exc})", flush=True) + _write_reclaim_marker(path, num, clock, from_rank=rank) + if rank > 0 and node_ips: + send_reclaim_to_rank0(node_ips[0], reclaim_relay_port(env), num, rank) for name in ("SIGINT", "SIGTERM"): sig.signal(getattr(sig, name), forward) - return proc.wait() + try: + return proc.wait() + finally: + if relay is not None: + relay.stop() + + +RECLAIM_RELAY_PORT_ENV = "YETO_RECLAIM_RELAY_PORT" +RECLAIM_RELAY_PORT_DEFAULT = 29417 + + +def reclaim_relay_port(env: dict) -> int: + return int(env.get(RECLAIM_RELAY_PORT_ENV) or RECLAIM_RELAY_PORT_DEFAULT) + + +def _node_rank_and_ips(env: dict) -> tuple[int, list[str]]: + """This container's rank and the cluster's node ips (from :func:`skypilot_env`).""" + ips = [ip for ip in (env.get("SKYPILOT_NODE_IPS") or "").splitlines() if ip.strip()] + return int(env.get("SKYPILOT_NODE_RANK") or 0), ips + + +def _write_reclaim_marker(path: str, num: int, clock, *, from_rank: int) -> None: + """Write the reclaim marker the learner's watcher polls; report, never raise + (it runs in a signal handler or the relay thread).""" + try: + tmp = f"{path}.{os.getpid()}.{threading.get_ident()}.tmp" + with open(tmp, "w", encoding="utf-8") as fh: + record = {"signal": int(num), "time": clock()} + if from_rank: # relayed from a rank>0 container (the driver names it) + record["rank"] = int(from_rank) + fh.write(json.dumps(record) + "\n") + os.replace(tmp, path) + print(f"[modal-island] signal {num} (rank {from_rank}): reclaim marker {path} written for the learner", + flush=True) + except OSError as exc: + print(f"[modal-island] signal {num} (rank {from_rank}): reclaim marker not written ({exc})", flush=True) + + +def send_reclaim_to_rank0(host: str, port: int, num: int, rank: int, *, timeout_s: float = 2.0, + attempts: int = 3) -> bool: + """A rank>0 runner tells rank 0's relay that this container is being reclaimed. + Short timeouts: it runs inside Modal's grace window. Returns True once acked.""" + msg = (json.dumps({"signal": int(num), "rank": int(rank)}) + "\n").encode() + for attempt in range(attempts): + try: + with socket.create_connection((host, port), timeout=timeout_s) as conn: + conn.sendall(msg) + conn.settimeout(timeout_s) + if conn.recv(16).startswith(b"ok"): + print(f"[modal-island] signal {num}: relayed to rank 0 at {host}:{port}", flush=True) + return True + except OSError as exc: + print(f"[modal-island] signal {num}: relay to rank 0 failed (attempt {attempt + 1}: {exc})", + flush=True) + time.sleep(0.2) + return False + + +class ReclaimRelay: + """On rank 0 of a multi-node island: accept reclaim messages from peer runners + and hand each to ``on_reclaim(signal, rank)`` (which writes the local marker).""" + + def __init__(self, on_reclaim, *, port: int, host: str = "0.0.0.0") -> None: + self.on_reclaim, self.host, self.port = on_reclaim, host, port + self._sock: socket.socket | None = None + self._thread: threading.Thread | None = None + self._stop = threading.Event() + + def start(self) -> "ReclaimRelay": + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((self.host, self.port)) + sock.listen(8) + sock.settimeout(0.5) + self._sock, self.port = sock, sock.getsockname()[1] + self._thread = threading.Thread(target=self._serve, name="reclaim-relay", daemon=True) + self._thread.start() + print(f"[modal-island] reclaim relay listening on port {self.port}", flush=True) + return self + + def _serve(self) -> None: + while not self._stop.is_set(): + try: + conn, _addr = self._sock.accept() + except (socket.timeout, OSError): + continue + with conn: + try: + conn.settimeout(2.0) + doc = json.loads(conn.makefile("r", encoding="utf-8").readline()) + self.on_reclaim(int(doc["signal"]), int(doc["rank"])) + conn.sendall(b"ok\n") + except (OSError, ValueError, KeyError, TypeError) as exc: + print(f"[modal-island] reclaim relay: bad message ({exc})", flush=True) + + def stop(self) -> None: + self._stop.set() + if self._sock is not None: + self._sock.close() + if self._thread is not None: + self._thread.join(timeout=2.0) class ModalOps: @@ -1194,6 +1296,19 @@ def stream_logs(self, call_id: str): for entry in modal.FunctionCall.from_id(call_id).logs.stream(): yield getattr(entry, "message", str(entry)) + def stream_container_logs(self, container_id: str): + """The log lines of one container (``modal container logs``; follows while it + runs). The function-call stream of :meth:`stream_logs` does not deliver the + lines of a container that Modal started after a reschedule (S25 C1).""" + proc = subprocess.Popen([sys.executable, "-m", "modal", "container", "logs", str(container_id)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + try: + assert proc.stdout is not None + yield from proc.stdout + finally: + if proc.poll() is None: + proc.terminate() + CONTAINER_LINE_RE = re.compile(r"\[modal-island (\d+)\] rank (\d+) container (\S+)") @@ -1291,6 +1406,94 @@ def stop(self) -> None: self._stop.set() +class ContainerLogFollower: + """Follows the logs of the Modal containers the call log streams miss (S26 E9.3). + + After ``modal container stop`` (or a real reclaim) Modal runs the island's + input again in a new container, and the function-call log stream never + delivers that container's lines (S19, S25 C1), so the launcher showed only + the first container. Every ``interval_s`` this lists the app's running + containers; the first ``expected`` ids are the baseline (the call streams + show them); for every later id it prints one launcher warning and streams + that container's own log (``stream_container(id)``) on a daemon thread. + + Each line is printed with a ``[