diff --git a/evidence/rlf/decision-log.md b/evidence/rlf/decision-log.md index af0593ae2..13e2814d2 100644 --- a/evidence/rlf/decision-log.md +++ b/evidence/rlf/decision-log.md @@ -49,3 +49,24 @@ - 教训：岛停止前 tape/checkpoint 未拉回（"no pulled Modal tape file"），本跑 rollout 的 reward 证据随岛销毁。下次开卡：跑挂时先拉回 ~/yeto-output/ 与 ~/yeto-rl/ 再停。 - failure class：trainer_kernel_compile（与 RLF 无关；与模型/LoRA shape 相关）。 - 途中修复（已提交 3ff8e93e）：launcher 透传 YETO_RLF_* 前缀、rlf_modal_provider 入 MODAL_SANDBOX_PROVIDERS、岛侧任务目录 /root/sky_workdir 路径、sandbox token 走 Modal Secret。 + +## 2026-10-11 gate 第三跑（rlf-gate-20261011-5，失败，双根因均已定位） + +代码 d15f6e1b（session-server-workers=8 生效：48s 就绪，远小于 300s 窗），unfused 后端，约 16 min ≈$0.55。 +进展链完整走到训练步：v0 rl_publication 落磁带（policy update 证据已可收集）、rollout 0 收齐 16 样本进训练转换。两个失败： + +1. **训练步 CUDA OOM**（unfused 后端，seq 8192：注意力分数矩阵全程驻留，与 LoRA 无关）。下一跑改 `--rl-training-attention-backend fused`（cuDNN，显存与 flash 同量级且完全绕开 flash-attn cute JIT 编译失败）。 +2. **RLF CTF 沙箱被 Modal 300s 空闲回收**：`RlfModalVmConfig` 写死 `idle_timeout=min(ttl_s,300)`，模型长生成轮（exec 空窗 >300s）后 VM 消失，下一 exec 抛 NotFoundError → episode 无 trusted finish → preflight fail-closed 丢弃该组（3 组被丢、由补位凑齐 16 样本；reward 全 0 与 0.8B 能力一致，非管线错误）。修复 de0a7b7a：idle_timeout 默认跟随 ttl_s，可显式覆盖。判据：fail-closed 行为本身正确，属环境 TTL 配置缺陷，非代码竞态。 + +待用户重启后第四跑：fused + 沙箱修复，PREFIX 递增，预算 ≤$3。 + +## 2026-10-11 gate 第四跑（rlf-gate-20261011-6，失败）与第五跑决策（代拍板） + +第四跑（fused/L40S/0.8B，约 12 min ≈$0.4）：沙箱修复生效（NotFoundError 归零）、fused 旗标生效（Megatron 打印 AttnBackend.fused）、rollout 0 收齐 16 样本。训练步失败：TE 报 "No dot product attention backend is available"——cuDNN fused 对 L40S/sm89 + 该输入组合不可用。三条后端全灭：flash（cute JIT 编译错）、unfused（seq8192 OOM：mbs 已为 1、recompute 未启用且无透传旗标、单样本分数块 7.34 GiB、总缺口 ~5.4 GiB）、fused（不支持）。trusted-fields 拒收 2 组为 ret2win 超时中止的 by-design 行为。 + +第五跑决策（代拍板，留痕）： +1. 卡型换 H200×1（141 GiB）+ unfused：4B 在 seq 8192 mbs1 无 recompute 估算峰值 ~87 GiB（据 0.8B 实测 42.5+7.3 外推），141 GiB 有充足余量；不加 recompute 透传（无旗标、加管道有风险）。 +2. 模型换 Qwen3.5-4B（rev 851bf6e8，M1 生产钉版，profile qwen35）：0.8B 连续两跑 16+ 样本 reward 全 0，gate 五判据中 std>0/success_count/advantage 三项都依赖至少一次 ezshell 成功，0.8B 不可行。4B 是 M1 agentic 跑验证过的形状（H200 + session server + TorchMemorySaver 共置）。 +3. 其余不变（LoRA r16 attention、seq 8192、GRPO 2×4×4、worker 8）。预算预登记 ≤$6。 + +第五跑（rlf-gate-20261011-7）head 形态（代拍板）：本地 preflight 线程守卫把内核线程计入 root（~1000），真实用户线程仅 ~1841 仍卡 2800 线；按用户指示改用 `--controller head`（Nebius eu-north1 无卡 head）+ `--preflight-threads off`——理由：guard 的目的是保护跑 driver 的机器，head 模式下 driver 在 Nebius VM，本地只剩瘦客户端，本地计数是不适用的机器。Modal 令牌按 s19-critic-g3.sh 模式走环境变量进 head（不落盘）。跑完先从 head 拉回产物再 sky down。 diff --git a/tests/test_rlf_rl_gate.py b/tests/test_rlf_rl_gate.py index a63c4929f..faa063dc9 100644 --- a/tests/test_rlf_rl_gate.py +++ b/tests/test_rlf_rl_gate.py @@ -268,3 +268,76 @@ def test_prepare_prompt_data_rejects_unadmitted_package(tmp_path): from yeto.rl.rlf.contracts import AdmissionError with pytest.raises(AdmissionError): prepare_prompt_data(_config(tmp_path), tmp_path / "prompt.jsonl") + + +# --- rl_sample_outcome tape collection (ports engine, S21 gate-7) ----------- + +def _outcome(group, index, reward, success, task="fixture-ctf-ezshell"): + return {"event": "rl_sample_outcome", "policy_version": 0, + "group_index": group, "index": index, + "task_id": task, "reward": reward, "success": success} + + +def test_tape_sample_outcomes_collect(tmp_path): + from yeto.rl.eval.rlf_rl_gate import collect_rollout_samples_from_tape + tape = _tape(tmp_path, [ + {"event": "rl_driver_start"}, + _outcome(0, 0, 1.0, True), + _outcome(0, 1, 0.0, False), + "garbage", + _outcome(1, 0, 0.0, False, task="fixture-ctf-ret2win"), + ]) + records = collect_rollout_samples_from_tape(tape) + assert len(records) == 3 + assert records[0]["reward"] == 1.0 and records[0]["success"] is True + assert records[2]["task_id"] == "fixture-ctf-ret2win" + + +def test_tape_sample_outcomes_fail_closed(tmp_path): + from yeto.rl.eval.rlf_rl_gate import collect_rollout_samples_from_tape + from yeto.rl.eval.rlf_rl_gate import GateEvidenceError + # no outcome records at all + tape = _tape(tmp_path, [{"event": "rl_driver_start"}]) + with pytest.raises(GateEvidenceError, match="no rl_sample_outcome"): + collect_rollout_samples_from_tape(tape) + # missing success must abort, never default to False + tape = _tape(tmp_path, [_outcome(0, 0, 1.0, None)]) + with pytest.raises(GateEvidenceError, match="lacks success"): + collect_rollout_samples_from_tape(tape) + # non-numeric reward aborts + bad = _outcome(0, 0, "1.0", True) + tape = _tape(tmp_path, [bad]) + with pytest.raises(GateEvidenceError, match="numeric reward"): + collect_rollout_samples_from_tape(tape) + + +def test_emit_sample_outcomes_writes_tape(tmp_path, monkeypatch): + from types import SimpleNamespace + from yeto.rl.adapters.miles.legacy import engine as legacy_engine + tape = tmp_path / "rl-island-0.jsonl" + monkeypatch.setattr("yeto.rl.wandb_rl.tee", lambda *a, **k: None) + args = SimpleNamespace(yeto_rl_event_tape=str(tape), yeto_rl_learner_id=0, + yeto_rl_policy_version=0) + + class _Sample: + def __init__(self, reward, metadata): + self._d = {"reward": reward, "metadata": metadata} + + def to_dict(self): + return self._d + + groups = [ + [_Sample(1.0, {"success": True, "rlf_task_id": "fixture-ctf-ezshell"}), + _Sample(0.0, {"success": False})], + [_Sample(None, {"success": False})], # no numeric reward -> skipped + ] + legacy_engine._emit_sample_outcomes(args, groups) + events = [json.loads(line) for line in tape.read_text().splitlines()] + outcomes = [e for e in events if e["event"] == "rl_sample_outcome"] + assert len(outcomes) == 2 + assert outcomes[0]["reward"] == 1.0 and outcomes[0]["success"] is True + assert outcomes[0]["task_id"] == "fixture-ctf-ezshell" + assert outcomes[1]["success"] is False + # no tape configured: no-op, no crash + legacy_engine._emit_sample_outcomes( + SimpleNamespace(yeto_rl_event_tape=None, yeto_rl_policy_version=0), groups) diff --git a/yeto/rl/adapters/miles/legacy/engine.py b/yeto/rl/adapters/miles/legacy/engine.py index 3d41a019d..555f9d28f 100644 --- a/yeto/rl/adapters/miles/legacy/engine.py +++ b/yeto/rl/adapters/miles/legacy/engine.py @@ -815,9 +815,48 @@ def queue_completed_groups(args, all_samples, data_source) -> None: "dropped_groups": dropped, "drop_reasons": drop_reasons, } + _emit_sample_outcomes(args, complete) source.add_samples(kept) +def _emit_sample_outcomes(args, complete: list[list[Any]]) -> None: + """rlf-codex-rl 5.8: per-sample reward/success onto the event tape. + + The ports bridge can only persist an empty completed-group queue, so the + RL gate's sample-level evidence must ride the tape (which is mirrored to + the Modal volume and survives the island). Fail-closed: a sample without + a numeric reward is skipped, and the gate rejects a tape with gaps. + """ + if not getattr(args, "yeto_rl_event_tape", None): + return + from yeto.rl.engine.events import write_event + + policy_version = int(getattr(args, "yeto_rl_policy_version", -1)) + for group_index, group in enumerate(complete): + for index, sample in enumerate(group): + first = sample[0] if isinstance(sample, list) else sample + to_dict = getattr(first, "to_dict", None) + if not callable(to_dict): + continue + data = to_dict() + reward = data.get("reward") + if not isinstance(reward, (int, float)) or isinstance(reward, bool): + continue + metadata = data.get("metadata") or {} + write_event( + args, + { + "event": "rl_sample_outcome", + "policy_version": policy_version, + "group_index": group_index, + "index": index, + "task_id": metadata.get("rlf_task_id") or metadata.get("task_id"), + "reward": float(reward), + "success": bool(metadata["success"]) if "success" in metadata else None, + }, + ) + + def _serialize_completed_groups(groups: list[list[Any]]) -> list[list[dict[str, Any]]]: serialized = [] for group in groups: diff --git a/yeto/rl/eval/rlf_rl_gate.py b/yeto/rl/eval/rlf_rl_gate.py index f1089a5b4..3df4c501d 100644 --- a/yeto/rl/eval/rlf_rl_gate.py +++ b/yeto/rl/eval/rlf_rl_gate.py @@ -190,6 +190,49 @@ def collect_rollout_samples( return records +def collect_rollout_samples_from_tape( + tape_path: str | Path, +) -> list[dict[str, Any]]: + """Real reward/success collection from ``rl_sample_outcome`` tape events. + + rlf-codex-rl 5.8 代拍板: on the ports engine the island checkpoint only + ever holds an empty completed-group queue, so sample-level evidence rides + the event tape (mirrored to the Modal volume, survives the island). The + same fail-closed rules as the checkpoint path: any gap in reward/success + aborts the gate. + """ + outcomes: list[dict[str, Any]] = [] + for line in Path(tape_path).expanduser().read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(record, dict) or record.get("event") != "rl_sample_outcome": + continue + outcomes.append(record) + if not outcomes: + raise GateEvidenceError("event tape has no rl_sample_outcome records") + records: list[dict[str, Any]] = [] + for order, outcome in enumerate(outcomes): + reward = outcome.get("reward") + if not isinstance(reward, (int, float)) or isinstance(reward, bool): + raise GateEvidenceError(f"tape sample {order} lacks a numeric reward") + if outcome.get("success") is None: + raise GateEvidenceError( + f"tape sample {order} lacks success " + "(rlf_success reward must write it)" + ) + records.append({"group_index": int(outcome.get("group_index", order)), + "index": int(outcome.get("index", 0)), + "task_id": outcome.get("task_id"), + "reward": float(reward), + "success": bool(outcome["success"])}) + return records + + # ------------------------------------------------------- policy update evidence def _summarize_params_real(path: str | Path) -> dict[str, float]: @@ -386,7 +429,9 @@ def main(argv: list[str] | None = None) -> None: pd.add_argument("--flags-json", default=None, help="path to the trusted flags JSON (never logged)") gate = sub.add_parser("gate", help="collect island artifacts, judge, write evidence") - gate.add_argument("--completed-groups", required=True) + gate.add_argument("--completed-groups", default=None, + help="island checkpoint with completed groups; optional when " + "--tape-events carries rl_sample_outcome records (ports engine)") gate.add_argument("--adapter-before", default=None) gate.add_argument("--adapter-after", default=None) gate.add_argument("--tape-events", default=None, @@ -409,7 +454,12 @@ def main(argv: list[str] | None = None) -> None: config = GateConfig(tasks_dir=args.tasks_dir, flags=flags, repeat=args.repeat) prepare_prompt_data(config, args.out) return - samples = collect_rollout_samples(args.completed_groups) + if args.completed_groups: + samples = collect_rollout_samples(args.completed_groups) + elif args.tape_events: + samples = collect_rollout_samples_from_tape(args.tape_events) + else: + raise SystemExit("rollout samples need --completed-groups or --tape-events") adapter_pair = (args.adapter_before, args.adapter_after) if args.tape_events and any(adapter_pair): raise SystemExit("--tape-events is mutually exclusive with --adapter-before/--adapter-after") @@ -424,6 +474,7 @@ def main(argv: list[str] | None = None) -> None: rows = build_gate_rows(samples, update) evidence = run_gate(rows, collector="miles_island", update=update) evidence["policy_update_source"] = update_source + evidence["sample_source"] = "completed_groups" if args.completed_groups else "event_tape" write_evidence(evidence, args.out, secrets=secrets) if not evidence["verdict"]: raise SystemExit(f"RL gate failed: {evidence['gate']}")