diff --git a/docs/RLF_CODEX_RUN.md b/docs/RLF_CODEX_RUN.md index a3c20b45..1c82a2d5 100644 --- a/docs/RLF_CODEX_RUN.md +++ b/docs/RLF_CODEX_RUN.md @@ -110,3 +110,68 @@ Known unverified (first GPU run must confirm): Miles `resolve_serving` coverage of the `qwen3` tito family, LoRA-free `SessionServerConfig` (`lora_rank=None`), and A100-80GB scheduling latency on Modal. + +## Real-GPU RL gate (5.8): Miles GRPO on `qwen35_08b` + +The gate proves the RLF -> Miles GRPO -> GAE -> policy-update chain on real +GPU with the small `qwen35_08b` profile (0.8B, the S17-verified harness +stack; LoRA, 1xGPU). The 30B capability surface is covered separately by the +5.6 baseline; 30B RL training is a follow-up item (design decision +2026-10-11). Tasks: `fixture-ctf-ezshell` (solvable in ~2 turns, provides +success) and `fixture-ctf-ret2win` (provides reward variance). Codex-only; +AgentFlow is offline-only and refused at admission. + +### Runner and commands + +The gate runner is `yeto.rl.eval.rlf_rl_gate`. Island side, in order: + + # 1. admit the packages, write Miles --prompt-data (flags never logged) + python -m yeto.rl.eval.rlf_rl_gate prompt-data \ + --tasks-dir "$YETO_RLF_CTF_TASKS_DIR" --flags-json /trusted/flags.json \ + --out /run/rlf_gate_prompt.jsonl --repeat 2 + + # 2. launch the Miles GRPO island (qwen35_08b, LoRA, 1xGPU): + # --data /run/rlf_gate_prompt.jsonl --codex-backend-profile qwen35_08b + # --reward-function rlf_success (-> --custom-rm-path yeto.rl.adapters.miles.rewards.rlf_success) + # --advantage-estimator grpo (fixed by island_entry) + # env: YETO_HARNESS_ENVIRONMENT_PROVIDER=yeto.cloud.modal_reward_env:rlf_modal_provider + # YETO_RLF_CTF_TASKS_DIR / YETO_RLF_CTF_FLAGS_JSON (Modal Secret, never in logs) + + # 3. after training, collect real artifacts, judge, write evidence + python -m yeto.rl.eval.rlf_rl_gate gate \ + --completed-groups /run/completed_groups.pt \ + --adapter-before /run/adapter_init --adapter-after /run/adapter_final \ + --step 2 --flags-json /trusted/flags.json --out /run/rlf_gate_evidence.json + +### Hard criteria (`yeto.rl.rlf.contracts.rl_gate`) + +Non-empty reward, `reward_std > 0`, `success_count > 0`, +`advantage_abs_sum > 0`, `policy_update == true`. All-zero or constant +reward MUST fail. The verdict is announced only for evidence collected +from the real island artifacts (`collector="miles_island"`); CPU/injected +collections exercise the orchestration but can never announce a pass. + +### Evidence chain (`yeto.rlf.rl_gate.evidence/v1`) + +- reward/success: the island completed-groups checkpoint + (`--completed-groups-path`) — the samples the learner actually trained + on, each with the `rlf_success` reward and `metadata.success` (W4 grade); +- advantage: recomputed per GRPO group with `contracts.group_advantages` + (the designated computation of record; never hand-filled); +- policy update: `contracts.policy_update_evidence` over a per-tensor + summary of the initial vs final LoRA adapter (代拍板: the Miles fork + exports no per-step snapshot hook; this proves parameters changed without + a fork patch, at the cost of not attributing the change to a specific + optimizer step, which the gate does not require); +- the JSON carries reward detail, success, `advantage_abs_sum`, and the + before/after policy digests; trusted flags are scrubbed + (`rlf.modal.scrub`) and only their sha256 appears in `config_digest`. + +### Stop conditions + +Stop and treat as infrastructure (never scored): serving not healthy within +15 min, or per-episode HARD deadline (30 min) exceeded repeatedly. Stop +with gate failure (the chain is broken, investigate before re-running): +`rl_gate` raising on any criterion, a sample missing `metadata.success`, +or unchanged adapter digests after training. A passing run archives the +evidence JSON next to the island artifacts. diff --git a/evidence/rlf/acceptance-status.json b/evidence/rlf/acceptance-status.json index d4331689..c858cd9c 100644 --- a/evidence/rlf/acceptance-status.json +++ b/evidence/rlf/acceptance-status.json @@ -6,7 +6,7 @@ "trusted_reference_flag": "pass", "codex_trajectory_metadata": "pass_cpu_tests", "cpu_reward_gae_gate": "pass", - "modal_gpu_rl_gate": "pending", + "modal_gpu_rl_gate": "attempted_failed_trainer_cutlass", "qwen3_baseline": "pass", "failure_class": "resolved_transport_single_recv", "swe": "deferred", @@ -16,5 +16,6 @@ "ctf_lease_evidence": "evidence/rlf/modal-ctf-lease-e2e.log (vm sb-01M4MEKE66M30GA31EZBGEBRR0, destroyed, describe=gone)", "qwen3_baseline_evidence": "run10 (s1-runs/s21-rlf-baseline-20261011a/run10-out): ezshell success reward=1.0 (2 turns), ret2win failed reward=0.0 (9 turns, wrong win addr); success_count=1, reward mean=0.5 std>0; run8: ret2win 36-turn 31K-token max_seq_len truncation (5.7 trigger evidence)", "compaction_real_firing": "pass", - "compaction_evidence": "run18 (s1-runs/.../run18-out): 12K window/4K reserve, compaction fired at 10320 tokens, summary_ok, context 8210->2176, episode continued to 12 turns + submit; run14: qwen3_cyber 24K profile active with correct no-fire gating" + "compaction_evidence": "run18 (s1-runs/.../run18-out): 12K window/4K reserve, compaction fired at 10320 tokens, summary_ok, context 8210->2176, episode continued to 12 turns + submit; run14: qwen3_cyber 24K profile active with correct no-fire gating", + "modal_gpu_rl_gate_evidence": "s1-runs/s21-rlf-baseline-20261011a/gate-launch7.log (rlf-gate-20261011-3): rollout+RLF reward path live; Megatron train step cutlass crd2idx compile error; artifacts lost with island (pull-before-stop lesson recorded)" } \ No newline at end of file diff --git a/evidence/rlf/decision-log.md b/evidence/rlf/decision-log.md index 84fdfda1..af0593ae 100644 --- a/evidence/rlf/decision-log.md +++ b/evidence/rlf/decision-log.md @@ -40,3 +40,12 @@ - 机制验证另用 12K 窗口/4K reserve/单轮 4096（触发线 8192）：run18 第 8 轮 10320 tokens 触发，summary_ok=true，segment 0 kept_steps=1，context 8210→2176 切换后继续 4 轮并以 submit 正常结束。segment identity 与共享 reward 合并由 CPU 测试锁定（test_codex_bridge_compaction / test_rlf_rollout_entry）。 - 途中修复：baseline runner 未预创建 segment sessions（已补，含 teardown 清理）；for_profile 与覆盖参数冲突（kwargs 优先）；budget==max_seq_len 桥约束（profile 24576 需配 context_length 24576）。 - 观察：该模型 thinking 轮常超 2K tokens，单轮截断（harness TITO 完整性规则）是长 episode 的主要死因；8K 窗口对该模型不可行。 + +## 2026-10-11 5.8 GPU RL gate 首跑（失败：训练端 cutlass 编译，非 RLF 链路） + +- 配置：L40S×1，qwen35_08b，GRPO 2 轮×4 组×4 样本，LoRA r16，rlf_success reward，rlf_modal_provider，任务 ezshell+ret2win。 +- 进展：岛启动、RLF provider 环境装配、rollout 正常产样本（prefill batch 4 并发）——RLF→Codex rollout→reward 链路在训练路径上真实工作。 +- 失败点：Megatron 训练步 cutlass MLIR 编译错误 `unable to compute crd2idx ... cute.layout "(?):(?{i64 div=8})"`，island rank 0 exit 1，launcher 按固定名单 0s 恢复超时停跑（exit 4）。 +- 教训：岛停止前 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。 diff --git a/tests/test_rlf_rl_gate.py b/tests/test_rlf_rl_gate.py new file mode 100644 index 00000000..38f0de38 --- /dev/null +++ b/tests/test_rlf_rl_gate.py @@ -0,0 +1,214 @@ +"""RLF real-GPU RL gate runner orchestration (task 5.8): fake learner / fake +provider only — no GPU, no Miles, no torch, no Modal. + +These tests cover the orchestration (collection -> gate rows -> rl_gate -> +evidence). The injected collection can never announce a passing verdict; +only the real ``miles_island`` collector on the GPU run may do that. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from yeto.rl.eval.rlf_rl_gate import ( + EVIDENCE_SCHEMA, + GateConfig, + GateEvidenceError, + build_gate_rows, + collect_policy_update, + collect_rollout_samples, + prepare_prompt_data, + run_gate, + write_evidence, +) + +FLAGS = {"fixture-ctf-ezshell": "flag{ezshell-secret}", "fixture-ctf-ret2win": "flag{ret2win-secret}"} + + +def _sample(group, index, reward, success, task_id="fixture-ctf-ezshell"): + return {"reward": reward, + "metadata": {"success": success, "rlf_task_id": task_id, + "rlf_episode_id": f"ep-{group}-{index}", + "rlf_trajectory_id": f"traj-{group}-{index}", + "grader_evidence": {"trusted": True}}} + + +def _payload(groups): + return {"schema_version": 1, "completed_groups": groups} + + +def _passing_groups(): + # ezshell group: one success (variance + success_count); ret2win group: all fail. + return [[_sample(0, 0, 1.0, True), _sample(0, 1, 0.0, False), + _sample(0, 2, 0.0, False), _sample(0, 3, 0.0, False)], + [_sample(1, 0, 0.0, False, "fixture-ctf-ret2win"), + _sample(1, 1, 0.0, False, "fixture-ctf-ret2win")]] + + +def _update_loader(before, after): + tables = {"before": before, "after": after} + return lambda path: tables[str(path)] + + +def _update(updated=True): + before = {"lora_a": 0.5, "lora_b": -0.25} + after = {"lora_a": 0.5 + (0.01 if updated else 0.0), "lora_b": -0.25} + return collect_policy_update("before", "after", step=2, + loader=_update_loader(before, after)) + + +def _config(tmp_path): + return GateConfig(tasks_dir=str(tmp_path), flags=FLAGS) + + +# -- collection from the completed-groups checkpoint ------------------------- + +def test_collect_rollout_samples_reads_real_checkpoint_shape(): + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups())) + assert len(samples) == 6 + assert samples[0] == {"group_index": 0, "index": 0, "task_id": "fixture-ctf-ezshell", + "reward": 1.0, "success": True} + assert sum(s["success"] for s in samples) == 1 + + +def test_collect_flattens_compaction_segments_sharing_one_outcome(): + group = [[_sample(0, 0, 1.0, True), _sample(0, 0, 1.0, True)], _sample(0, 1, 0.0, False)] + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload([group])) + assert [s["reward"] for s in samples] == [1.0, 0.0] + + +def test_collect_fails_closed_on_missing_success_or_reward(): + bad = _sample(0, 0, 1.0, True) + del bad["metadata"]["success"] + with pytest.raises(GateEvidenceError, match="metadata.success"): + collect_rollout_samples("ckpt", loader=lambda _p: _payload([[bad]])) + bad = _sample(0, 0, 1.0, True) + bad["reward"] = None + with pytest.raises(GateEvidenceError, match="numeric reward"): + collect_rollout_samples("ckpt", loader=lambda _p: _payload([[bad]])) + with pytest.raises(GateEvidenceError, match="no completed groups"): + collect_rollout_samples("ckpt", loader=lambda _p: _payload([])) + + +# -- advantages and policy update are computed, never hand-filled ------------ + +def test_gate_rows_compute_grpo_group_advantages(): + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups())) + rows = build_gate_rows(samples, _update()) + first = [r["advantage"] for r in rows if r["metadata"]["group_index"] == 0] + assert first == [0.75, -0.25, -0.25, -0.25] # reward minus group mean + assert all(r["policy_update"] for r in rows) + + +def test_policy_update_unchanged_params_marks_rows_not_updated(): + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups())) + rows = build_gate_rows(samples, _update(updated=False)) + assert not any(r["policy_update"] for r in rows) + + +# -- the five gate criteria --------------------------------------------------- + +def test_all_five_criteria_pass_but_injected_collection_never_announces(tmp_path): + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups())) + update = _update() + rows = build_gate_rows(samples, update) + evidence = run_gate(rows, collector="injected", config=_config(tmp_path), update=update) + gate = evidence["gate"] + assert gate["passed"] and gate["result"]["reward_std"] > 0 + assert gate["result"]["success_count"] == 1 + assert gate["result"]["advantage_abs_sum"] > 0 + # CPU orchestration only: the 5.8 gate is never announced off the real run. + assert evidence["verdict"] is False + assert "never announced" in evidence["verdict_note"] + + +def test_all_zero_rewards_fail(): + groups = [[_sample(0, i, 0.0, False) for i in range(4)]] + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(groups)) + evidence = run_gate(build_gate_rows(samples, _update()), collector="miles_island") + assert evidence["verdict"] is False and evidence["gate"]["passed"] is False + assert "reward_std_positive" in evidence["gate"]["error"] + assert "success_count_positive" in evidence["gate"]["error"] + assert "advantage_abs_sum_positive" in evidence["gate"]["error"] + + +def test_constant_nonzero_rewards_fail(): + groups = [[_sample(0, i, 1.0, True) for i in range(4)]] + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(groups)) + evidence = run_gate(build_gate_rows(samples, _update()), collector="miles_island") + assert evidence["verdict"] is False + assert "reward_std_positive" in evidence["gate"]["error"] + + +def test_no_policy_update_fails_even_with_real_collector(): + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups())) + update = _update(updated=False) + rows = build_gate_rows(samples, update) + evidence = run_gate(rows, collector="miles_island", update=update) + assert evidence["verdict"] is False + assert "policy_update" in evidence["gate"]["error"] + pu = evidence["policy_update_evidence"] + assert pu["updated"] is False and pu["before_digest"] == pu["after_digest"] + + +def test_real_collector_passing_run_announces_verdict(tmp_path): + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups())) + update = _update() + rows = build_gate_rows(samples, update) + evidence = run_gate(rows, collector="miles_island", config=_config(tmp_path), update=update) + assert evidence["verdict"] is True and evidence["schema"] == EVIDENCE_SCHEMA + assert evidence["policy_update_evidence"]["before_digest"] != \ + evidence["policy_update_evidence"]["after_digest"] + + +# -- evidence on disk: content and flag hygiene -------------------------------- + +def test_write_evidence_scrubs_flags_and_keeps_detail(tmp_path): + config = _config(tmp_path) + samples = collect_rollout_samples("ckpt", loader=lambda _p: _payload(_passing_groups())) + rows = build_gate_rows(samples, _update()) + rows[0]["metadata"]["note"] = f"grader saw {FLAGS['fixture-ctf-ezshell']} then submitted" + evidence = run_gate(rows, collector="injected", config=config, update=_update()) + out = write_evidence(evidence, tmp_path / "evidence.json", secrets=config.flags.values()) + text = out.read_text() + for flag in FLAGS.values(): + assert flag not in text + assert "***" in text + loaded = json.loads(text) + assert loaded["schema"] == EVIDENCE_SCHEMA + assert {r["reward"] for r in loaded["rows"]} == {0.0, 1.0} + assert "flag{" not in json.dumps(loaded["config_digest"]) # digest only + + +def test_config_requires_flags_for_gate_tasks(tmp_path): + with pytest.raises(GateEvidenceError, match="no trusted flag"): + GateConfig(tasks_dir=str(tmp_path), flags={"fixture-ctf-ezshell": "flag{x}"}) + + +# -- prompt data preparation ---------------------------------------------------- + +def _package(root, task_id): + (root / "public").mkdir(parents=True) + (root / "public" / "README.md").write_text(f"# {task_id}\nObtain the flag.") + (root / "task.yaml").write_text(f"task_id: {task_id}\ngrader_mode: flag\n") + (root / "MANIFEST.json").write_text(json.dumps({"task_id": task_id, "grader_mode": "flag"})) + + +def test_prepare_prompt_data_admits_gate_tasks(tmp_path): + for task_id in FLAGS: + _package(tmp_path / task_id, task_id) + out = prepare_prompt_data(_config(tmp_path), tmp_path / "prompt.jsonl") + rows = [json.loads(line) for line in Path(out).read_text().splitlines() if line.strip()] + assert len(rows) == 2 * len(FLAGS) # repeat=2 per task + assert {r["metadata"]["rlf_task_id"] for r in rows} == set(FLAGS) + assert all(r["metadata"]["benchmark"] == "rlf" for r in rows) + + +def test_prepare_prompt_data_rejects_unadmitted_package(tmp_path): + _package(tmp_path / "fixture-ctf-ezshell", "fixture-ctf-ezshell") + from yeto.rl.rlf.contracts import AdmissionError + with pytest.raises(AdmissionError): + prepare_prompt_data(_config(tmp_path), tmp_path / "prompt.jsonl") diff --git a/yeto/rl/eval/rlf_rl_gate.py b/yeto/rl/eval/rlf_rl_gate.py new file mode 100644 index 00000000..c33bd604 --- /dev/null +++ b/yeto/rl/eval/rlf_rl_gate.py @@ -0,0 +1,357 @@ +"""Real-GPU RLF RL gate runner and evidence chain (rlf-codex-rl task 5.8). + +One gate run, in this order: + +1. admit the CTF task packages (``fixture-ctf-ezshell``: solvable in ~2 + turns, provides success; ``fixture-ctf-ret2win``: provides reward + variance) into the RLF catalog and render Miles ``--prompt-data`` rows + (``yeto.rl.rlf.catalog.write_prompt_rows`` via :func:`write_prompt_data`); +2. launch one Miles GRPO island on the small ``qwen35_08b`` profile (0.8B, + the S17-verified harness stack; LoRA, 1xGPU) with + ``--custom-rm-path yeto.rl.adapters.miles.rewards.rlf_success`` and + ``--advantage-estimator grpo`` (both already fixed by + ``island_entry.build_miles_argv``) and the W4 RLF environment provider + (``YETO_HARNESS_ENVIRONMENT_PROVIDER=yeto.cloud.modal_reward_env:rlf_modal_provider`` + + ``YETO_RLF_CTF_TASKS_DIR`` / ``YETO_RLF_CTF_FLAGS_JSON``). The 30B + capability surface is covered separately by the 5.6 baseline; 30B RL + training is a follow-up item (design decision 2026-10-11); +3. after training, collect the gate evidence from real learner/rollout + artifacts — never hand-filled: + - rewards/success: the island's completed-groups checkpoint + (``--completed-groups-path``), i.e. the serialized Miles samples the + learner actually trained on, each carrying the ``rlf_success`` + reward value and ``metadata.success``; + - advantages: recomputed per GRPO group with + ``yeto.rl.rlf.contracts.group_advantages`` (the computation of record + designated by W3 for paths where the learner does not export + per-sample advantages); + - policy update: ``contracts.policy_update_evidence`` over a parameter + summary of the initial adapter vs the final trained adapter + (see "代拍板" note below); +4. judge with ``contracts.rl_gate`` (hard criteria: non-empty reward, + reward_std>0, success_count>0, advantage_abs_sum>0, policy_update) and + write the evidence JSON (reward detail, success, advantage_abs_sum, + policy-update digest before/after). All-zero or constant reward MUST + fail the gate. + +代拍板 (policy-update collection, pending main-agent sign-off): the Miles +fork exports no per-step parameter snapshot hook, so the minimal viable +collection is a parameter summary (per-tensor ``float(sum)``) of the +initial LoRA adapter (``--initial-adapter`` snapshot, or the base snapshot +at t0) versus the final saved adapter, fed to +``contracts.policy_update_evidence``. Rationale: it reuses W3's digest +contract unchanged, proves the optimizer step changed policy parameters, +and needs no fork patch; it cannot attribute the change to a specific +optimizer step, which the gate does not require. + +Flag hygiene: trusted flags live only in ``YETO_RLF_CTF_FLAGS_JSON`` (the +Modal Secret). Every persisted evidence string is scrubbed through +``yeto.rl.rlf.modal.scrub`` with the flag values; flags are never logged. + +The verdict is announced only for evidence collected from the real island +artifacts (``collector="miles_island"``). CPU tests inject fake learners +and only exercise the orchestration: an injected collection can never +produce a passing verdict, so the gate cannot be declared green off GPU. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Mapping, Sequence + +from yeto.rl.rlf.catalog import import_catalog, write_prompt_data +from yeto.rl.rlf.contracts import ( + AdmissionError, + PolicyUpdateEvidence, + group_advantages, + policy_update_evidence, + rl_gate, +) +from yeto.rl.rlf.modal import scrub + +EVIDENCE_SCHEMA = "yeto.rlf.rl_gate.evidence/v1" +DEFAULT_PROFILE = "qwen35_08b" +DEFAULT_TASK_IDS = ("fixture-ctf-ezshell", "fixture-ctf-ret2win") +GATE_TASKS_DIR_ENV = "YETO_RLF_CTF_TASKS_DIR" +GATE_FLAGS_ENV = "YETO_RLF_CTF_FLAGS_JSON" + + +class GateEvidenceError(RuntimeError): + """Collected artifacts cannot support a gate judgment (fail closed).""" + + +@dataclass +class GateConfig: + """One 5.8 gate run. Flags never leave the trusted side (never logged).""" + + tasks_dir: str + flags: Mapping[str, str] # task_id -> trusted flag + profile: str = DEFAULT_PROFILE + task_ids: tuple[str, ...] = DEFAULT_TASK_IDS + repeat: int = 2 # prompt rows per task + rounds: int = 2 # Miles --num-rollout + groups_per_round: int = 4 # --rollout-batch-size + samples_per_group: int = 4 # --n-samples-per-prompt + lora_rank: int = 16 + evidence: Mapping[str, Any] = field( + default_factory=lambda: {"runtime_quality_gate": {"healthy": True}}) + + def __post_init__(self) -> None: + missing = [t for t in self.task_ids if t not in self.flags] + if missing: + raise GateEvidenceError(f"no trusted flag for tasks: {missing}") + + def digest(self) -> str: + """Config digest WITHOUT the flags (safe to persist).""" + blob = json.dumps( + {"profile": self.profile, "task_ids": sorted(self.task_ids), + "repeat": self.repeat, "rounds": self.rounds, + "groups_per_round": self.groups_per_round, + "samples_per_group": self.samples_per_group, + "lora_rank": self.lora_rank, + "flag_sha256": {t: hashlib.sha256(self.flags[t].encode()).hexdigest() + for t in sorted(self.flags)}}, + sort_keys=True, separators=(",", ":")) + return hashlib.sha256(blob.encode()).hexdigest() + + +# ---------------------------------------------------------------- prompt data + +def prepare_prompt_data(config: GateConfig, out_path: str | Path) -> Path: + """Admit the gate task packages and write Miles ``--prompt-data`` JSONL.""" + roots = [Path(config.tasks_dir) / task_id for task_id in config.task_ids] + catalog = import_catalog(roots, evidence=dict(config.evidence)) + return write_prompt_data(catalog, out_path, repeat=config.repeat) + + +# ---------------------------------------------------- rollout sample collection + +def _load_island_checkpoint(path: str | Path) -> Mapping[str, Any]: + """Real loader: the island's completed-groups checkpoint (lazy torch).""" + import torch # GPU/learner side only; CPU tests inject ``loader`` + + payload = torch.load(Path(path).expanduser(), map_location="cpu", weights_only=True) + if not isinstance(payload, Mapping) or "completed_groups" not in payload: + raise GateEvidenceError(f"island checkpoint {path} lacks completed_groups") + return payload + + +def collect_rollout_samples( + checkpoint_path: str | Path, + *, + loader: Callable[[str | Path], Mapping[str, Any]] | None = None, +) -> list[dict[str, Any]]: + """Real reward/success collection from the learner's completed groups. + + Returns one record per sample: ``group_index``/``index`` (position in the + checkpoint), ``task_id``, ``reward`` and ``success``. A sample without a + graded ``reward`` or ``metadata.success`` aborts the gate (fail closed) — + silent zeros are exactly what this gate exists to reject. + """ + payload = (loader or _load_island_checkpoint)(checkpoint_path) + groups = payload.get("completed_groups") + if not isinstance(groups, list) or not groups: + raise GateEvidenceError("island checkpoint has no completed groups") + records: list[dict[str, Any]] = [] + for group_index, group in enumerate(groups): + if not isinstance(group, list): + raise GateEvidenceError(f"completed group {group_index} is not a sample list") + for index, sample in enumerate(group): + if isinstance(sample, list): # compaction segments share one outcome + if not sample: + raise GateEvidenceError("empty segment list in completed group") + sample = sample[0] + if not isinstance(sample, Mapping): + raise GateEvidenceError(f"sample {group_index}/{index} is not a mapping") + reward = sample.get("reward") + metadata = sample.get("metadata") or {} + if not isinstance(reward, (int, float)) or isinstance(reward, bool): + raise GateEvidenceError(f"sample {group_index}/{index} lacks a numeric reward") + if "success" not in metadata: + raise GateEvidenceError( + f"sample {group_index}/{index} lacks metadata.success " + "(rlf_success reward must write it)") + records.append({"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"])}) + return records + + +# ------------------------------------------------------- policy update evidence + +def _summarize_params_real(path: str | Path) -> dict[str, float]: + """Real loader: per-tensor ``float(sum)`` over a safetensors/torch file or dir.""" + tensors: dict[str, Any] = {} + root = Path(path).expanduser() + files = sorted(root.rglob("*.safetensors")) if root.is_dir() else [root] + if files and all(f.suffix == ".safetensors" for f in files): + from safetensors.torch import load_file # lazy, learner side only + + for file in files: + tensors.update(load_file(str(file))) + else: + import torch # lazy, learner side only + + payload = torch.load(root, map_location="cpu", weights_only=True) + state = payload.get("state_dict", payload) if isinstance(payload, Mapping) else payload + if not isinstance(state, Mapping): + raise GateEvidenceError(f"cannot read parameter tensors from {path}") + tensors = dict(state) + summary: dict[str, float] = {} + for name, tensor in tensors.items(): + try: + summary[str(name)] = float(tensor.float().sum()) + except (TypeError, AttributeError, RuntimeError) as error: + raise GateEvidenceError(f"non-tensor parameter {name!r} in {path}") from error + if not summary: + raise GateEvidenceError(f"no parameter tensors found in {path}") + return summary + + +def collect_policy_update( + before_path: str | Path, + after_path: str | Path, + *, + step: int, + loader: Callable[[str | Path], Mapping[str, float]] | None = None, +) -> PolicyUpdateEvidence: + """Digest the initial vs final adapter parameters (see module 代拍板 note).""" + load = loader or _summarize_params_real + return policy_update_evidence(dict(load(before_path)), dict(load(after_path)), step=step) + + +# ------------------------------------------------------------------ gate rows + +def build_gate_rows( + samples: Sequence[Mapping[str, Any]], + update: PolicyUpdateEvidence, +) -> list[dict[str, Any]]: + """Gate rows: real rewards/success, GRPO group advantages, real update flag. + + Advantages come from ``contracts.group_advantages`` per rollout group + (reward minus group mean) — the designated computation of record; the + ``policy_update`` flag comes from the collected PolicyUpdateEvidence, + never from a literal. + """ + by_group: dict[Any, list[Mapping[str, Any]]] = {} + for record in samples: + by_group.setdefault(record["group_index"], []).append(record) + rows: list[dict[str, Any]] = [] + for group_index in sorted(by_group): + group = by_group[group_index] + advantages = group_advantages([r["reward"] for r in group]) + for record, advantage in zip(group, advantages): + rows.append({"reward": float(record["reward"]), + "advantage": float(advantage), + "metadata": {"success": bool(record["success"]), + "task_id": record.get("task_id"), + "group_index": group_index}, + "policy_update": update.updated}) + return rows + + +# ---------------------------------------------------------------------- run + +def run_gate( + rows: Sequence[Mapping[str, Any]], + *, + collector: str, + config: GateConfig | None = None, + update: PolicyUpdateEvidence | None = None, +) -> dict[str, Any]: + """Judge the rows with ``rl_gate`` and assemble the evidence record. + + ``collector`` names the evidence source; only ``"miles_island"`` (the real + learner artifacts) may produce ``verdict=True``. Anything else (CPU fake + learners in tests, manual drills) exercises the orchestration but the + verdict stays False — the gate is never announced off the real run. + """ + try: + result = rl_gate(list(rows)) + gate: dict[str, Any] = {"passed": True, "result": result} + except AdmissionError as error: + gate = {"passed": False, "error": str(error)} + real = collector == "miles_island" + verdict = bool(gate["passed"]) and real + evidence: dict[str, Any] = { + "schema": EVIDENCE_SCHEMA, + "collected_at": time.time(), + "collector": collector, + "verdict": verdict, + "verdict_note": ( + "gate criteria evaluated on real Miles island artifacts" + if real else + "orchestration-only collection (injected/fake learner): " + "the 5.8 gate is never announced off the real GPU run"), + "gate": gate, + "rows": [dict(row) for row in rows], + } + if update is not None: + evidence["policy_update_evidence"] = { + "step": update.step, "updated": update.updated, + "before_digest": update.before_digest, "after_digest": update.after_digest, + "changed": list(update.changed)} + if config is not None: + evidence["config_digest"] = config.digest() + evidence["profile"] = config.profile + evidence["task_ids"] = sorted(config.task_ids) + return evidence + + +def write_evidence(evidence: Mapping[str, Any], out_path: str | Path, + *, secrets: Sequence[str] = ()) -> Path: + """Persist the evidence JSON with every trusted flag scrubbed.""" + text = json.dumps(evidence, ensure_ascii=False, indent=2, sort_keys=True, default=str) + path = Path(out_path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(scrub(text, *secrets) + "\n") + return path + + +def main(argv: list[str] | None = None) -> None: + """``python -m yeto.rl.eval.rlf_rl_gate {prompt-data|gate} ...`` (island side).""" + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + pd = sub.add_parser("prompt-data", help="admit packages, write Miles --prompt-data") + pd.add_argument("--tasks-dir", required=True) + pd.add_argument("--out", required=True) + pd.add_argument("--repeat", type=int, default=2) + 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("--adapter-before", required=True) + gate.add_argument("--adapter-after", required=True) + gate.add_argument("--step", type=int, required=True) + gate.add_argument("--out", required=True) + gate.add_argument("--flags-json", default=None) + args = parser.parse_args(argv) + + secrets: list[str] = [] + flags_path = getattr(args, "flags_json", None) + if flags_path: + flags = json.loads(Path(flags_path).read_text()) + secrets = [str(v) for v in flags.values()] + if args.command == "prompt-data": + if not flags_path: + raise SystemExit("--flags-json is required (fail closed)") + 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) + update = collect_policy_update(args.adapter_before, args.adapter_after, step=args.step) + rows = build_gate_rows(samples, update) + evidence = run_gate(rows, collector="miles_island", update=update) + write_evidence(evidence, args.out, secrets=secrets) + if not evidence["verdict"]: + raise SystemExit(f"RL gate failed: {evidence['gate']}") + + +if __name__ == "__main__": + main()