S26 E1: single-island overhead (background round cuts, publish fast path, profilers)

openspec/changes/rl-resume-from-checkpoint/design.md

@@ -79,6 +79,7 @@ Modal 文档：退出处理函数有 30 秒宽限，被抢占时也调用；GPU
 - **单岛切点**（沿用格式）：adapter、优化器、学习率调度器、随机数、iteration、consumed_samples、pointer（下一轮号、策略版本、策略哈希、启动序号）+ 新增 `run_fingerprint`（算法、模型、LoRA 形状、并行布局、学习率配置、数据集指纹与洗牌种子）与 `lr_at_next_round`（下一轮应使用的学习率）与 `consumed_prompt_ids_digest`（已用题号集合的摘要，详细列表放账本）。
 - **多岛**：每个岛存自己的切点（同上，路径 `<store>/islands/<岛名>/`）；syncer 检查点同步到 `<store>/syncer/`；切点 pointer 多记 `outer_version`。
 - **频率**：`--rl-cut-every N`（默认：0.6B 为 1，FN 为 2）；最后一轮、收到"我方停机"时总是存一次。
+- **Default since 2026-10-21 (E1)**: `--rl-cut-every` defaults to 4; publication checks (policy hash, engine read-back) run on the first publication and every 4th round; `--rl-verify-every-round` restores per-round checks. See the requirement "Round-cut interval and publication checks" in specs/rl-resume-from-checkpoint/spec.md.
 - **保留**：`--rl-cut-keep K`（默认 2）；新切点 MANIFEST 落盘并提交后才删旧的。
 - **目录**：`<store>/<运行名>/cuts/<轮号>/{shards…, manifest.json, MANIFEST.sha256}` + `<store>/<运行名>/LATEST`（最后写，指向最近完整切点）。
 

openspec/changes/rl-resume-from-checkpoint/specs/rl-resume-from-checkpoint/spec.md

@@ -49,3 +49,10 @@ launcher SHALL 接受 `--rl-checkpoint-store modal-volume://<卷名>[/前缀]`
 #### Scenario: 续训后的 tape
 - **WHEN** 第二次启动重跑了第一次未进切点的第 4 轮
 - **THEN** 第一次的第 4 轮记录仍在，`rl_resume.discarded_rounds` 含 4
+
+### Requirement: Round-cut interval and publication checks (E1, 2026-10-21)
+The product default round-cut interval SHALL be 4 rounds (`--rl-cut-every 4`; `--rl-cut-every 1` cuts every round), plus the last round and our own stop. A publication SHALL hash the policy and read the engine weight checksums back on the first publication, on every round whose number is a multiple of the cut interval (which includes every round that keeps a round cut), and on every round with `--rl-verify-every-round`. Every other publication SHALL still check that the trainer holds the published policy (trainer weights version) and SHALL be recorded with `sync/publication_verified: false` (verified publications carry no such field). A publication that was not verified MUST NOT be committed as a round cut.
+
+#### Scenario: an unverified round at a cut
+- **WHEN** a cut is due at a safe point whose published policy was not verified
+- **THEN** the cut is refused ("not verified") and the previous pointer stays the committed one

tests/golden/decoupling/codex_harness.json

@@ -81,6 +81,10 @@
   "bridge",
   "--model-name",
   "SimpleNamespace",
+  "--attention-dropout",
+  "0.0",
+  "--hidden-dropout",
+  "0.0",
   "--num-layers",
   "2",
   "--hidden-size",
@@ -229,7 +233,7 @@
   "<MILES-RESOLVED-TOOL-CALL-PARSER>",
   "--group-query-attention"
  ],
- "miles_argv_sha256": "279d3bae4c7c8381c9dfbea4e0e83ffdbfcd0afbeabb0f8931d7c3312984f771",
+ "miles_argv_sha256": "8a2934f5ac2b11fae39d1374ed834d0ab5ed4eb6c1d74f515a2f64bc2505830c",
  "name": "codex_harness",
  "placement": {
   "allow_cross_node_engine_tp": false,

tests/golden/decoupling/decoupled.json

@@ -63,6 +63,10 @@
   "bridge",
   "--model-name",
   "SimpleNamespace",
+  "--attention-dropout",
+  "0.0",
+  "--hidden-dropout",
+  "0.0",
   "--num-layers",
   "2",
   "--hidden-size",
@@ -195,7 +199,7 @@
   "10",
   "--group-query-attention"
  ],
- "miles_argv_sha256": "1108c08ed4e9ef28ec5ad2e37b3fe6ceb804db496824c9de4743b028a400a663",
+ "miles_argv_sha256": "6ff6da47c19cf9f9eac0ec0b9040e9e199220578244928625c5bec362379ccf6",
  "name": "decoupled",
  "placement": {
   "allow_cross_node_engine_tp": false,

tests/golden/decoupling/drgrpo.json

@@ -110,6 +110,10 @@
   "bridge",
   "--model-name",
   "SimpleNamespace",
+  "--attention-dropout",
+  "0.0",
+  "--hidden-dropout",
+  "0.0",
   "--num-layers",
   "2",
   "--hidden-size",
@@ -245,7 +249,7 @@
   "10",
   "--group-query-attention"
  ],
- "miles_argv_sha256": "7ee222703a50f20c2c18b817596ddbca10140fe94645ca28acdcba05c871cc2b",
+ "miles_argv_sha256": "8545800520cd18d7ee951fbb8a06d012cbb6bbd39d70968e85f35c1b05d2c6e7",
  "name": "drgrpo",
  "placement": {
   "allow_cross_node_engine_tp": false,

tests/golden/decoupling/elastic.json

@@ -66,6 +66,10 @@
   "bridge",
   "--model-name",
   "SimpleNamespace",
+  "--attention-dropout",
+  "0.0",
+  "--hidden-dropout",
+  "0.0",
   "--num-layers",
   "2",
   "--hidden-size",
@@ -201,7 +205,7 @@
   "120",
   "--group-query-attention"
  ],
- "miles_argv_sha256": "c88e8996324d0261dc72f5b2949afc1e7eb81db916d3746ac76addea772020d1",
+ "miles_argv_sha256": "5aeca40e53d9f1b1a614a6a6ec814e5d037b354da3f448f04c1d10417751d135",
  "name": "elastic",
  "placement": {
   "allow_cross_node_engine_tp": false,

tests/golden/decoupling/grpo_default.json

@@ -59,6 +59,10 @@
   "bridge",
   "--model-name",
   "SimpleNamespace",
+  "--attention-dropout",
+  "0.0",
+  "--hidden-dropout",
+  "0.0",
   "--num-layers",
   "2",
   "--hidden-size",
@@ -191,7 +195,7 @@
   "10",
   "--group-query-attention"
  ],
- "miles_argv_sha256": "cd9c51c2946a4e45bb63cd153ae64bc3ec0e3e998cf3d5e64fe6d4ed882c156b",
+ "miles_argv_sha256": "bb315aa0c82a03c2821899a3a376db84f09d30f1d32e2f6640833aec455d8634",
  "name": "grpo_default",
  "placement": {
   "allow_cross_node_engine_tp": false,

tests/golden/decoupling/grpo_tis.json

@@ -106,6 +106,10 @@
   "bridge",
   "--model-name",
   "SimpleNamespace",
+  "--attention-dropout",
+  "0.0",
+  "--hidden-dropout",
+  "0.0",
   "--num-layers",
   "2",
   "--hidden-size",
@@ -243,7 +247,7 @@
   "10",
   "--group-query-attention"
  ],
- "miles_argv_sha256": "06b407103acaff8218e5bab2309192be3af014b6ca635dd7c72f27cca28b1a2b",
+ "miles_argv_sha256": "70d04ad1b86c39f254ac2631b2be4a32558c00f4e0e7cc8c1767e03ea84b03ee",
  "name": "grpo_tis",
  "placement": {
   "allow_cross_node_engine_tp": false,

tests/golden/decoupling/seq_adv_maxrl.json

@@ -116,6 +116,10 @@
   "bridge",
   "--model-name",
   "SimpleNamespace",
+  "--attention-dropout",
+  "0.0",
+  "--hidden-dropout",
+  "0.0",
   "--num-layers",
   "2",
   "--hidden-size",
@@ -250,7 +254,7 @@
   "10",
   "--group-query-attention"
  ],
- "miles_argv_sha256": "8acfb1140db90c4d02fd4534526adcea04ca37a569125cee1952d3a697eb882d",
+ "miles_argv_sha256": "fd095a4b6d682aa8ba80b1953964b57575721859844217c78253847b7349a820",
  "name": "seq_adv_maxrl",
  "placement": {
   "allow_cross_node_engine_tp": false,

tests/rl_cut_fakes.py

@@ -27,7 +27,7 @@ def named_parameters(self, model):
     def is_adapter(self, name):
         return "lora" in name
 
-    def export_optimizer(self, optimizer, named):
+    def export_optimizer(self, optimizer, named, *, on_device=False):  # clones stay on their device
         by_id = {id(p): n for n, p in named}
         entries = {}
         for group in optimizer.param_groups:
@@ -170,7 +170,7 @@ def __init__(self, coord=None, *, side_effect_free=True):
         super().__init__(coord)
         self.side_effect_free = side_effect_free
 
-    def export_optimizer(self, optimizer, named):
+    def export_optimizer(self, optimizer, named, *, on_device=False):  # clones stay on their device
         from collections import defaultdict
 
         from yeto.rl.adapters.miles.cut_plugin import side_effect_free_state

tests/rl_reshard_fakes.py

@@ -66,7 +66,7 @@ def named_parameters(self, model):
     def is_adapter(self, name):
         return "lora" in name
 
-    def export_optimizer(self, optimizer, named):
+    def export_optimizer(self, optimizer, named, *, on_device=False):  # clones stay on their device
         optimizer.init_state()
         hyper = {k: v for k, v in optimizer.param_groups[0].items() if k != "params"}
         entries = {}

tests/test_rl_cut_distopt_pp_ep.py

@@ -31,6 +31,9 @@
 DPS_SOURCES = {
     "m5-c35702e": (os.environ.get("YETO_MILES_CHECKOUT", "/home/michael/work/miles-m3"), "c35702e"),
     "s25-distopt": (os.environ.get("YETO_MILES_DISTOPT_CHECKOUT", "/home/michael/work/miles-next"), "s25-distopt"),
+    # E1: + export_named_optimizer_state(device=None) on the image pin 44d41efd (not pushed)
+    "e1-device-export": (os.environ.get("YETO_MILES_DISTOPT_CHECKOUT", "/home/michael/work/miles-next"),
+                         "yeto/s26-e1-device-export"),
 }
 
 
@@ -355,3 +358,186 @@ def test_local_states_cover_with_fillers():
     assert cp.range_problems(state, state) == []
     other = {"entries": {"a": {**state["entries"]["a"], "start": 5}}}
     assert len(cp.range_problems(state, other)) == 2
+
+
+# --------------------------------------------------------------------------
+# E1: background cuts copy their state device-to-device first
+# --------------------------------------------------------------------------
+
+
+@needs_dps
+@pytest.mark.parametrize("layout", ["pp2dp1", "pp2dp2", "ep2"])
+def test_device_export_is_the_fork_host_export_once_on_the_host(layout):
+    ranks = _world(layout)
+    _train(ranks)
+    for rank in ranks:
+        named = list(rank.model[0].named_parameters())
+        host = DPS.export_named_optimizer_state(rank.optimizer, named)
+        device = cp.to_host(cp._device_export(DPS)(rank.optimizer, named))
+        assert cp.state_digest(device) == cp.state_digest(host)
+        assert device["format"] == host["format"] and list(device["entries"]) == list(host["entries"])
+        for name, e in host["entries"].items():
+            got = device["entries"][name]
+            assert {k: v for k, v in got.items() if k not in ("tensors", "scalars")} == \
+                {k: v for k, v in e.items() if k not in ("tensors", "scalars")}
+            for part in ("tensors", "scalars"):
+                assert list(got[part]) == list(e[part])
+                for key, t in e[part].items():
+                    assert got[part][key].dtype == t.dtype and torch.equal(got[part][key], t), (name, key)
+
+
+def test_device_export_uses_the_fork_device_argument_when_it_has_one():
+    calls = []
+
+    def export(optimizer, named, *, device="cpu"):
+        calls.append(device)
+        return {"entries": {}}
+
+    cp._device_export(SimpleNamespace(export_named_optimizer_state=export))("opt", [])
+    assert calls == [None]
+
+
+@needs_dps
+@pytest.mark.parametrize("staged", [False, True], ids=["resident", "sleeps_after_cut"])
+@pytest.mark.parametrize("layout", ["pp2dp2", "ep2"])
+def test_background_device_capture_writes_the_inline_shard(tmp_path, layout, staged, monkeypatch):
+    """Same shard (file bytes, digests, N7.6 resume state) from the device path as from the
+    inline host path, although the trainer trains on before the writer copies to the host."""
+    ranks = _world(layout)
+    _train(ranks)
+    inline = [cp.save_cut_shard(r, directory=str(tmp_path / "inline"), cut_id="c1") for r in ranks]
+    monkeypatch.setattr(cp, "_device_copy_plan", lambda actor, named: (True, "ok"))
+    if staged:  # a sleeping trainer's copies go through the staging buffer (CPU stand-ins too)
+        real_stage = cp._stage
+        monkeypatch.setattr(cp, "_sleeps_after_cut", lambda actor: True)
+        monkeypatch.setattr(cp, "_stage", lambda snap: real_stage(snap, include_cpu=True))
+    device = []
+    for r, rank in enumerate(ranks):  # one pending write per cut id per process: rank by rank
+        started = cp.save_cut_shard(rank, directory=str(tmp_path / "device"), cut_id="c1", background=True)
+        assert started["capture"] == {"mode": "device", "reason": "ok"}, started
+        for k, leaf in enumerate(_leaves(rank)):  # the next step changes the live state first
+            leaf.train(2000 + 10 * r + k)
+        rank.opt_param_scheduler.num_steps = 16
+        device.append(cp.finish_cut_shard(rank, cut_id="c1"))
+    assert [d.pop("writer") for d in device] == [{"mode": "process"}] * len(ranks)
+    assert cp._STAGING.free.is_set() and (cp._STAGING.buffer is not None or not staged)
+    assert device == inline
+    for name in sorted(p.name for p in (tmp_path / "inline").iterdir()):
+        assert (tmp_path / "device" / name).read_bytes() == (tmp_path / "inline" / name).read_bytes()
+
+
+def test_background_capture_falls_back_to_the_host_and_says_why(tmp_path, monkeypatch):
+    from tests.rl_cut_fakes import make_rank, train_step
+
+    rank = make_rank(0)
+    train_step(rank, torch.randn(3, 6, generator=torch.Generator().manual_seed(7)))
+    inline = cp.save_cut_shard(rank, directory=str(tmp_path / "inline"), cut_id="c1")
+    monkeypatch.setattr(cp, "_device_copy_plan", lambda actor, named: (False, "low_device_memory(free=1,need=2)"))
+    started = cp.save_cut_shard(rank, directory=str(tmp_path / "low"), cut_id="c1", background=True)
+    assert started["capture"] == {"mode": "host", "reason": "low_device_memory(free=1,need=2)"}
+    assert _without_writer(cp.finish_cut_shard(rank, cut_id="c1")) == inline
+
+    real = cp._snapshot
+
+    def oom(actor, backend, named, *, on_device=False):
+        if on_device:
+            raise torch.cuda.OutOfMemoryError("CUDA out of memory")
+        return real(actor, backend, named)
+
+    monkeypatch.setattr(cp, "_device_copy_plan", lambda actor, named: (True, "ok"))
+    monkeypatch.setattr(cp, "_snapshot", oom)
+    monkeypatch.setattr(torch.cuda, "empty_cache", lambda: None)
+    started = cp.save_cut_shard(rank, directory=str(tmp_path / "oom"), cut_id="c1", background=True)
+    assert started["capture"] == {"mode": "host", "reason": "device_oom"}
+    assert _without_writer(cp.finish_cut_shard(rank, cut_id="c1")) == inline
+
+
+def _without_writer(summary):
+    assert summary.pop("writer")["mode"] == "process", summary
+    return summary
+
+
+def _one_rank_cut(tmp_path, name, *, background):
+    import random
+
+    import numpy as np
+
+    from tests.rl_cut_fakes import make_rank, train_step
+
+    random.seed(5), np.random.seed(5), torch.manual_seed(5)  # the shard holds the global RNG state
+    rank = make_rank(0)
+    train_step(rank, torch.randn(3, 6, generator=torch.Generator().manual_seed(7)))
+    if not background:
+        return cp.save_cut_shard(rank, directory=str(tmp_path / name), cut_id="c1")
+    cp.save_cut_shard(rank, directory=str(tmp_path / name), cut_id="c1", background=True)
+    return cp.finish_cut_shard(rank, cut_id="c1")
+
+
+def test_a_dead_writer_process_falls_back_to_the_thread_and_says_why(tmp_path, monkeypatch):
+    """The writer process dies before it writes (its start fails): this rank's thread
+    writes the same shard and the summary records the fallback."""
+    import multiprocessing
+    import os
+    from concurrent.futures import ProcessPoolExecutor
+
+    inline = _one_rank_cut(tmp_path, "inline", background=False)
+    dead = ProcessPoolExecutor(max_workers=1, mp_context=multiprocessing.get_context("spawn"),
+                               initializer=os._exit, initargs=(3,))
+    monkeypatch.setattr(cp, "_WRITER", [dead])
+    got = _one_rank_cut(tmp_path, "dead", background=True)
+    writer = got.pop("writer")
+    assert writer["mode"] == "thread" and "writer process failed" in writer["fallback"], writer
+    assert got == inline
+    assert cp._WRITER == []  # the broken pool is dropped; the next cut starts a new one
+    assert (tmp_path / "dead" / got["path"]).read_bytes() == (tmp_path / "inline" / got["path"]).read_bytes()
+
+
+def test_a_partial_file_of_a_dead_writer_is_removed_before_the_fallback(tmp_path, monkeypatch):
+    inline = _one_rank_cut(tmp_path, "inline", background=False)
+    target = tmp_path / "partial"
+    target.mkdir()
+    (target / (inline["path"] + ".tmp")).write_bytes(b"half")
+    (target / inline["path"]).write_bytes(b"whole but unreported")
+
+    class Broken:
+        def submit(self, *a, **k):
+            raise RuntimeError("writer died")
+
+        def shutdown(self, **k):
+            pass
+
+    monkeypatch.setattr(cp, "_WRITER", [Broken()])
+    got = _one_rank_cut(tmp_path, "partial", background=True)
+    assert got.pop("writer") == {"mode": "thread", "fallback": "writer process failed: RuntimeError('writer died')"}
+    assert got == inline and not (target / (inline["path"] + ".tmp")).exists()
+
+
+def test_the_writer_process_can_be_turned_off(tmp_path, monkeypatch):
+    inline = _one_rank_cut(tmp_path, "inline", background=False)
+    monkeypatch.setenv(cp.WRITER_PROCESS_ENV, "0")
+    got = _one_rank_cut(tmp_path, "thread", background=True)
+    assert got.pop("writer") == {"mode": "thread", "fallback": "disabled"}
+    assert got == inline
+
+
+def test_device_copy_plan_needs_cuda_and_can_be_turned_off(monkeypatch):
+    named = [("lora_a", torch.nn.Parameter(torch.zeros(2)))]
+    assert cp._device_copy_plan(None, named) == (False, "not_cuda")
+    sleeper = SimpleNamespace(args=SimpleNamespace(offload_train=True))  # staged, below
+    assert cp._device_copy_plan(sleeper, named) == (False, "not_cuda")
+    monkeypatch.setenv(cp.DEVICE_COPY_ENV, "0")
+    assert cp._device_copy_plan(None, named) == (False, "disabled")
+
+
+def test_pack_hands_the_state_over_in_one_buffer_and_unpacks_it_exactly():
+    shared = torch.arange(6.0)
+    state = {"a": torch.randn(3, 4), "b": [torch.ones(2, dtype=torch.bfloat16), torch.tensor(7)],
+             "view": torch.arange(10.0)[2:5], "twice": (shared, shared), "empty": torch.zeros(0), "n": 3}
+    skeleton, layout, flat = cp._pack(state)
+    assert len(layout) == 3  # a, b[0], b[1]; the view, the shared storage and the empty one are not packed
+    assert flat.is_shared() and skeleton["n"] == 3
+    back = cp._unpack(skeleton, layout, flat)
+    assert cp.state_digest(back) == cp.state_digest(state)
+    for key in ("a", "view", "empty"):
+        assert back[key].dtype == state[key].dtype and torch.equal(back[key], state[key])
+    assert back["b"][0].untyped_storage().data_ptr() != flat.untyped_storage().data_ptr()  # own copies