diff --git a/tests/test_rlf_rl_gate.py b/tests/test_rlf_rl_gate.py index faa063dc..6a132f7e 100644 --- a/tests/test_rlf_rl_gate.py +++ b/tests/test_rlf_rl_gate.py @@ -311,6 +311,43 @@ def test_tape_sample_outcomes_fail_closed(tmp_path): collect_rollout_samples_from_tape(tape) +def _trajectory_reward(group, sample, reward, success, task="fixture-ctf-ezshell"): + return {"event": "rl_trajectory_reward", "rollout_id": 0, "policy_version": 0, + "group_index": group, "sample_index": sample, "trajectory_id": f"t-{sample}", + "task_id": task, "reward": reward, "success": success, "aborted": False} + + +def test_tape_trajectory_rewards_collect(tmp_path): + """rlf-codex-rl 5.8 attempt 8 lesson: on the ports engine the all-samples + hook is rollout_meta_hook (queue_completed_groups never runs); the driver + tapes its per-sample records as rl_trajectory_reward.""" + from yeto.rl.eval.rlf_rl_gate import collect_rollout_samples_from_tape + tape = _tape(tmp_path, [ + {"event": "rl_driver_start"}, + _trajectory_reward(0, 0, 1.0, True), + _trajectory_reward(0, 1, 0.0, False), + _trajectory_reward(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[0]["index"] == 0 and records[1]["index"] == 1 + assert records[2]["task_id"] == "fixture-ctf-ret2win" + + +def test_tape_trajectory_rewards_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 + # success None (no signed verdict) must abort, never default to False + tape = _tape(tmp_path, [_trajectory_reward(0, 0, 1.0, None)]) + with pytest.raises(GateEvidenceError, match="lacks success"): + collect_rollout_samples_from_tape(tape) + # reward None (non-finite) must abort + tape = _tape(tmp_path, [_trajectory_reward(0, 0, None, True)]) + 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 diff --git a/yeto/rl/eval/rlf_rl_gate.py b/yeto/rl/eval/rlf_rl_gate.py index 3df4c501..f682ba6c 100644 --- a/yeto/rl/eval/rlf_rl_gate.py +++ b/yeto/rl/eval/rlf_rl_gate.py @@ -200,6 +200,17 @@ def collect_rollout_samples_from_tape( 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. + + Two event shapes are accepted (rlf-codex-rl 5.8 attempt 8 lesson: the + ports launcher binds ``--rollout-all-samples-process-path`` to + ``rollout_meta_hook.extract_rollout_metadata``, so the legacy + ``queue_completed_groups`` hook never runs there; the meta hook already + reports per-sample records which the driver tapes as + ``rl_trajectory_reward`` behind ``--rl-observe-timeline``): + + - ``rl_sample_outcome`` (legacy engine hook, 054c20d5); + - ``rl_trajectory_reward`` (rl-fn-codex-rollout 1.0 observe-only pipeline; + ``sample_index`` fills ``index``). """ outcomes: list[dict[str, Any]] = [] for line in Path(tape_path).expanduser().read_text().splitlines(): @@ -210,11 +221,15 @@ def collect_rollout_samples_from_tape( record = json.loads(line) except json.JSONDecodeError: continue - if not isinstance(record, dict) or record.get("event") != "rl_sample_outcome": + if not isinstance(record, dict): + continue + if record.get("event") not in ("rl_sample_outcome", "rl_trajectory_reward"): continue outcomes.append(record) if not outcomes: - raise GateEvidenceError("event tape has no rl_sample_outcome records") + raise GateEvidenceError( + "event tape has no rl_sample_outcome/rl_trajectory_reward records" + ) records: list[dict[str, Any]] = [] for order, outcome in enumerate(outcomes): reward = outcome.get("reward") @@ -226,7 +241,7 @@ def collect_rollout_samples_from_tape( "(rlf_success reward must write it)" ) records.append({"group_index": int(outcome.get("group_index", order)), - "index": int(outcome.get("index", 0)), + "index": int(outcome.get("index", outcome.get("sample_index", 0))), "task_id": outcome.get("task_id"), "reward": float(reward), "success": bool(outcome["success"])}) @@ -431,7 +446,8 @@ def main(argv: list[str] | None = None) -> None: gate = sub.add_parser("gate", help="collect island artifacts, judge, write evidence") gate.add_argument("--completed-groups", default=None, help="island checkpoint with completed groups; optional when " - "--tape-events carries rl_sample_outcome records (ports engine)") + "--tape-events carries rl_sample_outcome/rl_trajectory_reward " + "records (ports engine)") gate.add_argument("--adapter-before", default=None) gate.add_argument("--adapter-after", default=None) gate.add_argument("--tape-events", default=None,