rlf-codex-rl 5.8：样本级 reward/success 上事件磁带，gate 从磁带收集 outcome

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。

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)

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:

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']}")