rlf-codex-rl 5.8：gate 磁带收集器同时接受 rl_trajectory_reward 事件（ports 引擎关键修复）

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

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,