diff --git a/openspec/changes/rl-drop-reward-tamper-hash/design.md b/openspec/changes/rl-drop-reward-tamper-hash/design.md index 39f9a602..ae60302a 100644 --- a/openspec/changes/rl-drop-reward-tamper-hash/design.md +++ b/openspec/changes/rl-drop-reward-tamper-hash/design.md @@ -38,3 +38,4 @@ 6. **secrlenv group key.** The group key of `check_group` used the mac to tell apart different evidence at the same index. It now uses the canonical outcome. 7. **Unchanged hashes.** The composition of `BackendIdentity`, `session_contract_hash`, `lr_schedule_sha256`, the algorithm hash, and `training_contract_hash` is unchanged. Note: `reward_contract_hash` is the sha256 of the reward source files. `reward.py` changed, so its value changes for new runs. Its composition does not change. The three source pin values were recomputed with the file changes. 8. **Island HMAC.** `YETO_ISLAND_HMAC_KEY` and the syncer seal/open are unchanged (confirmed with grep). +9. **Reward-function evidence kind (2026-10-14, P0-1.11 GPU run, third stage).** The dense full-parameter island hard-coded the `secrlenv` evidence kind, so a GSM8K island failed its first round with `UntrustedOutcome`. The main agent chose option A: add the kind `reward-function`. Its outcome is the prompt (task id `prompt-sha256:`), the logical sample (episode id `sample:`), and `sample.reward`; its outcome hash uses the domain `yeto-reward-function-evidence-v1`. The island chooses the kind from `--reward-function` (`island_entry._trajectory_evidence_kind`). Meaning change: evidence is either a trusted grader verdict or a reward function output. A sample that carries a grader verdict is rejected as `reward-function` evidence, so a grader run cannot fall back to the weaker kind. The schema number does not change: the kind is already part of the schema 3 and 4 batch hash. diff --git a/openspec/changes/rl-drop-reward-tamper-hash/specs/rl-reward-integrity-scope/spec.md b/openspec/changes/rl-drop-reward-tamper-hash/specs/rl-reward-integrity-scope/spec.md index 900f6c51..017b34f5 100644 --- a/openspec/changes/rl-drop-reward-tamper-hash/specs/rl-reward-integrity-scope/spec.md +++ b/openspec/changes/rl-drop-reward-tamper-hash/specs/rl-reward-integrity-scope/spec.md @@ -27,6 +27,21 @@ Newly written trajectory evidence SHALL use schema 3, and its outcome hash MUST - **WHEN** a batch of trajectory evidence is written - **THEN** the schema_version of the evidence is 3, and the hash input has no mac +### Requirement: Trajectory evidence names the source of its reward +Trajectory evidence SHALL record an evidence kind that names where each reward comes from. The kinds `secrlenv` and `terminal-bench-2.1` bind a trusted grader verdict, and the reward MUST equal that verdict. The kind `reward-function` binds the output of the island's reward function (`sample.reward`); the reward function itself is bound by `reward_contract_hash`. So evidence is either a trusted grader verdict or a reward function output, and it is not always a grader verdict. A dense full-parameter island SHALL choose the kind from its reward function: the SecRLEnv reward selects `secrlenv`, and every other reward function selects `reward-function`. Terminal-Bench 2.1 keeps the kind named in its SAO runtime contract. + +#### Scenario: A GSM8K island writes reward-function evidence +- **WHEN** a dense full-parameter island runs with `--reward-function yeto.rl.gsm8k_reward:score` +- **THEN** each round's evidence has kind `reward-function`, every trajectory reward equals `sample.reward`, and the full-parameter round record reads that evidence + +#### Scenario: A grader verdict is not downgraded +- **WHEN** a sample carries a SecRLEnv or Terminal-Bench verdict and the evidence kind is `reward-function` +- **THEN** building the evidence fails, so a trusted-grader run cannot drop its verdict check + +#### Scenario: A non-scalar reward is rejected +- **WHEN** `sample.reward` is missing, a boolean, not finite, or not a number +- **THEN** building `reward-function` evidence fails + ### Requirement: Compatibility hashes and island authentication are unchanged The composition and values of `BackendIdentity`, `session_contract_hash`, `lr_schedule_sha256`, the algorithm hash, `training_contract_hash`, and `reward_contract_hash` MUST NOT change. The island HELLO HMAC and the per-frame HMAC in elastic mode MUST stay. diff --git a/openspec/changes/rl-drop-reward-tamper-hash/tasks.md b/openspec/changes/rl-drop-reward-tamper-hash/tasks.md index 21b047fe..4bf14037 100644 --- a/openspec/changes/rl-drop-reward-tamper-hash/tasks.md +++ b/openspec/changes/rl-drop-reward-tamper-hash/tasks.md @@ -17,6 +17,8 @@ - [x] 2.7b Recompute the three source pin values: `agent.py`, `codex_generate.py`, `codex_harness_agent.py` (`pins.py`, `yeto/rl/__init__.py`). The files changed, so the pin values must change with them - [ ] 2.8 Run the local safe test set; the four tests backend_identity, elastic_backend_identity, hw_compat_group, and export MUST pass +- [x] 2.9 Add the `reward-function` evidence kind and choose the kind from the island's reward function (design.md implementation record item 9) + ## 3. Credential guard (per D1) - This section is not done in this change: D1 decided to move the credential guard out of this change (2026-10-13) diff --git a/tests/test_modal_runner.py b/tests/test_modal_runner.py index 42868120..91a79b14 100644 --- a/tests/test_modal_runner.py +++ b/tests/test_modal_runner.py @@ -488,6 +488,25 @@ def test_tape_sync_copies_changed_files_and_commits(tmp_path): assert len(commits) == sync.syncs >= 3 and not sync.errors +def test_tape_sync_mirrors_trajectory_evidence_outside_the_tape_dir(tmp_path): + # P0-1.11 run 20261014b: the full-parameter evidence stayed in a private + # directory of a stopped container and could not be read back. + src, dst = tmp_path / "out", tmp_path / "vol" / "rank0" + src.mkdir() + evidence = tmp_path / "rl" / "trajectory-evidence-0-abc" + evidence.mkdir(parents=True) + (evidence / "rollout-00000000.json").write_text('{"schema": 4}') + (evidence / "scratch.bin").write_text("x") + pattern = str(tmp_path / "rl" / "trajectory-evidence-*" / "rollout-*.json") + sync = mr.TapeSync(str(src), str(dst), lambda: None, interval_s=3600, + mirror_globs=(pattern,)) + assert sync.sync_once() == 1 + copied = dst / "trajectory-evidence-0-abc.rollout-00000000.json" + assert copied.read_text() == '{"schema": 4}' + assert sync.sync_once() == 0 + assert mr.TAPE_MIRROR_GLOBS == ("/root/yeto-rl/trajectory-evidence-*/rollout-*.json",) + + def test_tape_sync_errors_never_raise(tmp_path): def boom(): raise OSError("volume gone") diff --git a/tests/test_protocol_finalization.py b/tests/test_protocol_finalization.py index d7ea4f25..21f32ac2 100644 --- a/tests/test_protocol_finalization.py +++ b/tests/test_protocol_finalization.py @@ -4,6 +4,7 @@ import socket import struct import threading +import time import pytest import torch @@ -29,6 +30,7 @@ FinalFragment, FinalManifest, SyncerClient, + _Outbound, decode_final_manifest, encode_final_manifest, read_frame, @@ -288,3 +290,37 @@ def acknowledge_finalization(self, value): assert client.acknowledged == manifest for name in params: assert torch.equal(params[name], authoritative[name]) + + +def test_send_progress_counts_bytes_the_sender_thread_wrote(): + client_sock, server_sock = socket.socketpair() + client = SyncerClient(("unused", 0), 0, _layout(1), dtype=DTYPE_F32) + client._gen = 1 + client._connected.set() + client._queues = [queue.Queue()] + assert client.send_progress().bytes_sent == 0 + sent = threading.Event() + client._queues[0].put(_Outbound(b"x" * 100, sent)) + client._queues[0].put(_Outbound(b"y" * 20)) + progress = client.send_progress() + assert (progress.bytes_sent, progress.queued_chunks, progress.connected) == (0, 2, True) + sender = threading.Thread( + target=client._send_loop, + args=(1, client_sock, client._queues[0]), + daemon=True, + ) + sender.start() + try: + received = b"" + while len(received) < 120: + received += server_sock.recv(4096) + assert sent.wait(1.0) + deadline = time.monotonic() + 1.0 + while client.send_progress().bytes_sent < 120 and time.monotonic() < deadline: + time.sleep(0.01) + progress = client.send_progress() + assert (progress.bytes_sent, progress.queued_chunks, progress.reconnects) == (120, 0, 0) + finally: + client.close() + server_sock.close() + sender.join(timeout=1) diff --git a/tests/test_rl_agentic_suspend.py b/tests/test_rl_agentic_suspend.py index 11410e30..4dd1c671 100644 --- a/tests/test_rl_agentic_suspend.py +++ b/tests/test_rl_agentic_suspend.py @@ -358,6 +358,7 @@ def test_export_and_restore_buffered_groups_and_agentic_references(monkeypatch): report = in_flight.import_in_flight(target, exported["entries"], max_policy_age=1, current_version=4, completed_group_ids=["g2"], sample_from_dict=dict) assert report["resumed_groups"] == 1 and report["resumed"] == 2 + assert report["resumed_group_ids"] == ["g1"] assert report["reasons"] == {"group_already_completed": 2, in_flight.AGENTIC_SESSION_LOST: 1} restored = target.data_source.buffer[0] assert [m["index"] for m in restored] == [10, 11] and restored[0]["metadata"]["task_id"] == "task-1" @@ -366,6 +367,7 @@ def test_export_and_restore_buffered_groups_and_agentic_references(monkeypatch): report = in_flight.import_in_flight(target, exported["entries"], max_policy_age=1, current_version=5, sample_from_dict=dict) assert report["resumed_groups"] == 0 and report["reasons"]["policy_age_exceeded"] == 4 + assert report["resumed_group_ids"] == [] # restored under limit 0: everything discarded and reported report = in_flight.import_in_flight(target, exported["entries"], max_policy_age=0, current_version=4, sample_from_dict=dict) diff --git a/tests/test_rl_critic_fp32_masters.py b/tests/test_rl_critic_fp32_masters.py index f286b385..2c480638 100644 --- a/tests/test_rl_critic_fp32_masters.py +++ b/tests/test_rl_critic_fp32_masters.py @@ -270,6 +270,24 @@ def test_allowance_strict_avg_critic_on_two_islands_decoupled_still_refused(): assert critic_run_problems(ppo, {"sync_preset": "strict-avg"}) == [] +def test_allowance_sao_streaming_full_has_a_critic_channel(): + """sao-streaming-full requires the SAO critic, which has its own role + syncer (P0-D8); without the allowance no multi-island SAO run could start + (P0-1.11 GPU check, 2026-10-14).""" + + names = [ + "advantage_estimators:ppo", "execution:critic", "features:critic_multi_update", + "features:gae_decoupled", "features:gae_length_adaptive", "features:sao_dis", + "features:value_hl_gauss", + ] + assert check_unverified_allowance( + names, islands=2, outer_sync=True, sync_preset="sao-streaming-full" + ) == tuple(sorted(names)) + with pytest.raises(AlgorithmSpecError, match="single-island"): + check_unverified_allowance(["execution:critic", "losses:cispo"], islands=2, + outer_sync=True, sync_preset="sao-streaming-full") + + def test_launcher_two_island_critic_strict_avg_passes_decoupled_refused(tmp_path, _fake_sky, _no_modal_listing): # noqa: F811 from test_rl_algorithm_provenance import _launcher_args diff --git a/tests/test_rl_full_parameter_sync.py b/tests/test_rl_full_parameter_sync.py index d493513a..265532a8 100644 --- a/tests/test_rl_full_parameter_sync.py +++ b/tests/test_rl_full_parameter_sync.py @@ -677,3 +677,157 @@ def test_close_defaults_to_the_p5_finished_code(): # P5 code table: 0 is FINISHED (LeaveReason.FINISHED once P5 merges). assert inspect.signature(FullParameterSync.close).parameters["reason"].default == 0 + + +def test_layout_stream_role_splits_only_role_lane_algorithms(): + from yeto.rl.local_learner import layout_stream_role + + assert layout_stream_role("grpo", "actor") is None + assert layout_stream_role("sao", "actor") == "actor" + assert layout_stream_role("sao", "critic") == "critic" + with pytest.raises(ValueError, match="no trainable role"): + layout_stream_role("grpo", "critic") + with pytest.raises(ValueError, match="no trainable role"): + layout_stream_role("dpo", "actor") + + +@pytest.mark.parametrize(("algorithm", "role", "expected"), [ + ("grpo", "actor", None), + ("sao", "critic", "critic"), +]) +def test_role_stream_initializes_dense_with_the_plain_layout(monkeypatch, algorithm, role, expected): + # P0-1.11 G3-fp probe: the dense (grpo) actor stream passed stream_role="actor", + # which ParameterLayout.create refuses for non-SAO algorithms. + from yeto.rl.adapters.miles.full_parameter import role_stream as module + from yeto.rl.local_learner import ComponentIdentity, ParameterLayout, ParameterSpec + + seen = {} + + class Stop(Exception): + pass + + async def fake_initialize(group, **kwargs): + seen.update(kwargs) + ParameterLayout.create( + algorithm=kwargs["algorithm"], + components=kwargs["components"], + specs=(ParameterSpec(role, "w", (2,), "float32", 2, "tp0"),), + num_fragments=1, + fragment_strategy="owner_affine", + stream_role=kwargs["stream_role"], + ) + raise Stop + + monkeypatch.setattr(module.MilesChunkedFullParameterAdapter, "initialize", fake_initialize) + config = replace( + _role_stream_config(role, 30100), + algorithm=algorithm, + component=ComponentIdentity(role, "a" * 40, "c" * 64), + ) + with pytest.raises(Stop): + asyncio.run(module.MilesFullParameterRoleStream(config).initialize(object())) + assert seen["stream_role"] == expected + + +def test_initial_cut_wait_leaves_the_event_loop_to_the_other_role(monkeypatch): + # s23-g3-sao-main-20261014a: the actor stream of island 0 and the critic + # stream of island 1 each blocked the event loop while waiting for their + # initial cut, so neither island connected its other role: deadlock. + from yeto.rl.adapters.miles.full_parameter import role_stream as module + + seen = [] + + class Client: + def __init__(self, *_args, **_kwargs): + pass + + def install_inbound_chunk_sink(self, _sink): + pass + + def start(self): + pass + + def check_health(self): + pass + + def drain_updates(self): + seen.append("drain after peer" if "peer" in seen else "drain") + return [] + + def send_progress(self): + from yeto.protocol import SendProgress + + return SendProgress(bytes_sent=0, queued_chunks=0, reconnects=0, connected=True) + + def close(self): + pass + + adapter = SimpleNamespace( + layout=SimpleNamespace( + layout_hash="1" * 64, fragments=SimpleNamespace(num_fragments=2) + ), + fragment_sink=lambda _initial, ray_module=None: SimpleNamespace(release_all=lambda: 0), + release=lambda _initial: None, + ) + + async def fake_initialize(_group, **_kwargs): + return adapter, object() + + monkeypatch.setattr(module.MilesChunkedFullParameterAdapter, "initialize", fake_initialize) + monkeypatch.setattr(module, "SyncerClient", Client) + monkeypatch.setattr(module, "MilesRayInboundChunkSink", lambda sink: sink) + config = replace(_role_stream_config("actor", 30100), learner_id=1, wait_timeout=0.5) + + async def peer(): + await asyncio.sleep(0.05) + seen.append("peer") + + async def both(): + return await asyncio.gather( + module.MilesFullParameterRoleStream(config).initialize(object()), + peer(), + return_exceptions=True, + ) + + initialized, _ = asyncio.run(both()) + assert isinstance(initialized, TimeoutError) + assert "drain after peer" in seen + + +def test_role_stream_reports_initial_cut_upload_progress_on_timeout(monkeypatch, capsys): + # P0-1.11 run 20261014c: the initial cut timed out after 900 s with no + # sign of how much INIT_PARAMS had left the island; the wait now reports it. + from yeto.protocol import SendProgress + from yeto.rl.adapters.miles.full_parameter import role_stream as module + + class StalledClient: + def check_health(self): + pass + + def drain_updates(self): + return [] + + def send_progress(self): + return SendProgress( + bytes_sent=3 * 2**20, queued_chunks=7, reconnects=1, connected=True + ) + + monkeypatch.setattr(module, "INITIAL_CUT_REPORT_SECONDS", 0.0) + config = replace(_role_stream_config("actor", 30100), wait_timeout=0.05) + adapter = SimpleNamespace( + layout=SimpleNamespace(fragments=SimpleNamespace(num_fragments=2)) + ) + stream = module.MilesFullParameterRoleStream(config) + + with pytest.raises(TimeoutError, match="initial cut"): + stream._wait_for_initial_fragments(StalledClient(), adapter) + + lines = [ + line for line in capsys.readouterr().err.splitlines() + if line.startswith("[yeto-initial-cut]") + ] + assert lines[0].startswith("[yeto-initial-cut] learner 0 waiting: 0/2 fragment(s)") + assert "timed out: 0/2 fragment(s)" in lines[-1] + assert lines[-1].endswith( + "sent 3.0 MiB, 7 chunk(s) queued, 1 redial(s), connected=True" + ) diff --git a/tests/test_rl_launcher_dense_full.py b/tests/test_rl_launcher_dense_full.py index a246dac3..1ce3a062 100644 --- a/tests/test_rl_launcher_dense_full.py +++ b/tests/test_rl_launcher_dense_full.py @@ -132,6 +132,27 @@ def test_island_entry_accepts_the_dense_island_command(): island_entry._check_dense_full_sweep(parsed, yeto_policy_sync=True) +@pytest.mark.parametrize( + ("reward_function", "kind"), + ( + ("yeto.rl.harness.codex.reward:reward_func", "secrlenv"), + ("yeto.rl.gsm8k_reward:score", "reward-function"), + ("yeto.rl.math_reward:reward_func", "reward-function"), + ), +) +def test_dense_island_evidence_kind_follows_its_reward_function(reward_function, kind): + from types import SimpleNamespace + + from yeto.rl.adapters.miles import island_entry + from yeto.rl.trajectory_evidence import EVIDENCE_KINDS + + chosen = island_entry._trajectory_evidence_kind( + SimpleNamespace(reward_function=reward_function) + ) + assert chosen == kind + assert chosen in EVIDENCE_KINDS + + EXPERT_FULL = ( "--rl-sync-preset", "full-parameter", "--rl-placement", "fixed-partition", "--rl-rollout-gpus", "8", "--rl-parameter-layout-sha256", LAYOUT, "--fragments", "4", diff --git a/tests/test_rl_launcher_sao_streaming.py b/tests/test_rl_launcher_sao_streaming.py index 97bc625d..ddea7b4e 100644 --- a/tests/test_rl_launcher_sao_streaming.py +++ b/tests/test_rl_launcher_sao_streaming.py @@ -24,16 +24,6 @@ "features:value_hl_gauss", "features:critic_multi_update") -@pytest.fixture(autouse=True) -def _declared_sao_mechanisms(monkeypatch): - """The SAO mechanisms are not declared for outer-sync runs yet (only the - single-island allowance exists); these tests cover the launch wiring.""" - from yeto.rl.engine import algorithm - - monkeypatch.setattr(algorithm, "check_unverified_allowance", - lambda allowed, **_kwargs: tuple(allowed)) - - def _write(path, payload) -> str: path.parent.mkdir(parents=True, exist_ok=True) data = json.dumps(payload, sort_keys=True).encode() @@ -98,6 +88,38 @@ def _flag(command: str, flag: str) -> str: return tokens[tokens.index(flag) + 1] +def test_head_controller_reloads_the_contract_bundle(tmp_path, monkeypatch): + """The head replays the launch args as JSON, which drops the loaded + bundle; the head must reload it to render the role syncers + (s23-g3-sao-probe-20261014a: AttributeError rl_sao_manifest).""" + from yeto import cli, launcher + + args, plan = _plan(tmp_path) + seen = {} + + class Syncer: + def __init__(self, head_args, num_learners): + seen["command"] = launcher.syncer_command(head_args, num_learners) + + def start(self): + pass + + def start_log_forwarder(self): + pass + + def start_tape_forwarder(self, _args): + pass + + def stop(self): + pass + + monkeypatch.setattr(launcher, "LocalSyncer", Syncer) + monkeypatch.setattr(launcher, "run", lambda *_a, **_k: 0) + assert cli.cmd_head(json.dumps(cli._serializable_args(args))) == 0 + assert seen["command"] == launcher.syncer_command(args, 2) + assert "--require-profile-binding" in str(seen["command"]) + + def test_role_syncers_serve_exactly_the_frozen_profile(tmp_path): from yeto import launcher diff --git a/tests/test_rl_miles_adapter_trainer_publish.py b/tests/test_rl_miles_adapter_trainer_publish.py index 1db533c5..f60893dd 100644 --- a/tests/test_rl_miles_adapter_trainer_publish.py +++ b/tests/test_rl_miles_adapter_trainer_publish.py @@ -11,7 +11,12 @@ from yeto.rl.contracts import InferencePublicationManifest, LocalStepReceipt from yeto.rl.adapters.miles.publish import MilesPublisher, PublicationError from yeto.rl.adapters.miles.rollout import PolicyTokenMismatch, policy_token -from yeto.rl.adapters.miles.state_plugin import APPLIED_LRS, GRAD_NORM, STEP_LOSSES +from yeto.rl.adapters.miles.state_plugin import ( + ACTOR_RECORDERS, + APPLIED_LRS, + GRAD_NORM, + STEP_LOSSES, +) from yeto.rl.adapters.miles.trainer import MilesTrainerGroup, TrainStepError, batch_hash from yeto.rl.engine.ports import ( GroupMetadata, @@ -92,6 +97,55 @@ def test_train_step_receipt_and_release(): assert t.last_grad_norm == 0.3 +def test_full_parameter_trainer_installs_actor_recorders_once_before_training(): + """A full-parameter actor calls no LoRA state plugin, so nothing else + installs the per-step recorders; without them no applied LR is recorded.""" + + actor = FakeActorGroup() + t = MilesTrainerGroup( + args=SimpleNamespace(num_steps_per_rollout=1, offload_train=True), + actor_model=actor, learner_id=0, learner_generation=0, + parameter_layout_hash=lambda: L, + release_refs=lambda args, pack: None, + install_actor_recorders=True, + ) + t.train_step(handle()) + t.train_step(handle()) + trains = [i for i, call in enumerate(actor.calls) if call[0] == "train"] + assert actor.calls[: trains[0]] == [("plugin", ACTOR_RECORDERS)] + assert actor.calls.count(("plugin", ACTOR_RECORDERS)) == 1 + + +def test_full_parameter_trainer_refuses_to_train_without_recorders(): + class NoRecorderActor(FakeActorGroup): + async def run_plugin(self, fn_path, kwargs=None): + if fn_path == ACTOR_RECORDERS: + self.calls.append(("plugin", fn_path)) + return [False] + return await super().run_plugin(fn_path, kwargs) + + actor, released = NoRecorderActor(), [] + t = MilesTrainerGroup( + args=SimpleNamespace(num_steps_per_rollout=1, offload_train=True), + actor_model=actor, learner_id=0, learner_generation=0, + parameter_layout_hash=lambda: L, + release_refs=lambda args, pack: released.append(pack), + install_actor_recorders=True, + ) + with pytest.raises(TrainStepError, match="recorders could not be installed"): + t.train_step(handle()) + assert actor.calls == [("plugin", ACTOR_RECORDERS)] and released == ["PACK"] + + +def test_only_a_full_parameter_sync_installs_actor_recorders(): + from yeto.rl.adapters.miles.entry import _is_full_parameter_sync + from yeto.rl.engine.full_parameter_sync import FullParameterSync + + assert _is_full_parameter_sync(object.__new__(FullParameterSync)) + assert not _is_full_parameter_sync(object()) + assert not _is_full_parameter_sync(None) + + def test_token_mismatch_rejected_before_training_and_refs_released(): actor, released = FakeActorGroup(), [] with pytest.raises(PolicyTokenMismatch, match="g0"): diff --git a/tests/test_rl_miles_chunked_full_parameter.py b/tests/test_rl_miles_chunked_full_parameter.py index e9c0ec28..5cd7885f 100644 --- a/tests/test_rl_miles_chunked_full_parameter.py +++ b/tests/test_rl_miles_chunked_full_parameter.py @@ -4,6 +4,7 @@ import asyncio import hashlib +import json import struct import sys import types @@ -46,7 +47,9 @@ ) from yeto.rl.adapters.miles.full_parameter.inbound_sink import MilesRayInboundChunkSink from yeto.rl.adapters.miles.models.chunked_full_parameter import ( + LAYOUT_PROBE_LOG_PREFIX, MilesChunkedFullParameterAdapter, + layout_attestation_record, ) from yeto.rl.trajectory_evidence import TrajectoryBatchEvidence @@ -203,6 +206,60 @@ async def install_full_parameter_fragment_plans(self, _plans): assert group.installed is False +def test_layout_refusal_prints_the_observed_layout_as_an_attestation_record(capsys): + class Group: + async def full_parameter_shard_manifests(self): + return (manifest(1, (99, 1), role="critic"), manifest(0, (100, 1), role="critic")) + + async def install_full_parameter_fragment_plans(self, _plans): + raise AssertionError("layout drift must fail before plan installation") + + with pytest.raises(RuntimeError, match="differs from the frozen probe"): + asyncio.run( + MilesChunkedFullParameterAdapter.initialize( + Group(), + policy_version=0, + algorithm="sao", + stream_role="critic", + components=(component("critic"),), + minimum_fragments=2, + expected_fragments=2, + expected_layout_hash="c" * 64, + max_fragment_bytes=400, + max_chunk_bytes=64, + ) + ) + lines = [ + line + for line in capsys.readouterr().out.splitlines() + if line.startswith(LAYOUT_PROBE_LOG_PREFIX) + ] + assert len(lines) == 1 + record = json.loads(lines[0][len(LAYOUT_PROBE_LOG_PREFIX) :]) + expected = MilesChunkedFullParameterAdapter.create( + (manifest(0, (100, 1), role="critic"), manifest(1, (99, 1), role="critic")), + algorithm="sao", + stream_role="critic", + components=(component("critic"),), + minimum_fragments=2, + max_fragment_bytes=400, + max_chunk_bytes=64, + ) + assert record == layout_attestation_record(expected, (component("critic"),)) + assert record["role"] == "critic" + assert record["component"] == {"model_revision": "a" * 40, "config_sha256": "b" * 64} + assert record["parameter_layout_sha256"] == expected.layout.layout_hash + assert record["expected_fragments"] == expected.layout.fragments.num_fragments + assert [fragment["fragment_id"] for fragment in record["fragments"]] == list( + range(record["expected_fragments"]) + ) + assert sum(f["scalar_count"] for f in record["fragments"]) == 201 + assert record["parameter_scalar_count"] == 201 + assert record["parameter_tensor_count"] == 4 + assert all(f["payload_bytes"] == 4 * f["scalar_count"] for f in record["fragments"]) + assert len(record["manifests"]) == 2 + + def _cut(adapter, values, *, policy_version, generation, ray): shards = [] for plan in adapter.plans: diff --git a/tests/test_rl_policy_age.py b/tests/test_rl_policy_age.py index b3d15c24..9e1b3f78 100644 --- a/tests/test_rl_policy_age.py +++ b/tests/test_rl_policy_age.py @@ -184,6 +184,21 @@ def test_driver_limit_one_accepts_version_segments_within_the_window(tmp_path): _driver(tmp_path, 1, versions=lambda rid: (max(0, rid - 1), rid)).run() events = [json.loads(l) for l in (tmp_path / "e.jsonl").read_text().splitlines()] assert len([e for e in events if e["event"] == "rl_round_trained"]) == 3 + # P0-S evidence: each accepted group's generating policy, labelled with the + # published hash of its version + labels = {e["policy_version"]: e["rl/policy_token"] for e in events if e["event"] == "rl_publication"} + policies = [e for e in events if e["event"] == "rl_group_policies"] + assert [e["rollout_id"] for e in policies] == [0, 1, 2] + for e in policies: + first = e["groups"][0] + assert first[1] == labels[e["rollout_id"]] + assert first[2] == sorted({max(0, e["rollout_id"] - 1), e["rollout_id"]}) + assert all(g[1] == labels[e["rollout_id"]] and g[2] is None for g in e["groups"][1:]) + + +def test_driver_limit_zero_tapes_no_group_policies(tmp_path): + _driver(tmp_path, 0).run() + assert '"rl_group_policies"' not in (tmp_path / "e.jsonl").read_text() def test_driver_limit_one_refuses_a_group_two_versions_old(tmp_path): diff --git a/tests/test_rl_restart_data_cursor.py b/tests/test_rl_restart_data_cursor.py index a95af7a2..5de12e3d 100644 --- a/tests/test_rl_restart_data_cursor.py +++ b/tests/test_rl_restart_data_cursor.py @@ -317,3 +317,59 @@ def test_strict_fresh_start_without_ledger_does_not_seek(tmp_path): driver.run() assert pool.seeks == [] and syncer.version == rounds _close(driver, ctl) + + +# ------------------------------------------------- P0-S: resume from a round cut +_CUT = {"sample_offset": 35, "epoch_id": 0, "sample_group_index": 35, "sample_index": 280} + + +def _cut_resume_driver(*, ledger_cursor, landed_shift=0): + from yeto.rl.engine.bridges import LocalOnlySync + from yeto.rl.engine.driver import IslandDriver + + sync = LocalOnlySync(6) + sync.round_cuts = SimpleNamespace(resume_data_cursor=dict(_CUT)) + seeks, events = [], [] + + def seek(cursor): + seeks.append(dict(cursor)) + return {**cursor, "sample_offset": cursor["sample_offset"] + landed_shift} + + fake = SimpleNamespace( + sync=sync, rollout=SimpleNamespace(seek_data_cursor=seek), + ledger=SimpleNamespace(restart_cursor=lambda rid: ledger_cursor, state=lambda rid: "outer_recorded"), + emit=lambda event, **fields: events.append((event, fields))) + return IslandDriver._restore_data_cursor, fake, seeks, events + + +def test_round_cut_resume_seeks_to_the_cut_cursor_without_a_ledger_cursor(): + """GPU s23-g3-ps-b-20261014a: under fully_async the ledger recorded no cursor for + rollout 2 and the resume was refused; the cut's own cursor is the resume point.""" + restore, fake, seeks, events = _cut_resume_driver(ledger_cursor=None) + restore(fake, 3) + assert seeks == [_CUT] + assert events == [("rl_data_cursor_restored", + {"rollout_id": 3, "source": "round_cut", "data_cursor": _CUT})] + + +def test_round_cut_cursor_wins_over_the_ledger_record(): + """The background worker drew past the round's record; the cut put those groups + back into the buffer, so continuing from the ledger cursor would re-draw them.""" + restore, fake, seeks, _ = _cut_resume_driver(ledger_cursor={**_CUT, "sample_offset": 30}) + restore(fake, 3) + assert seeks == [_CUT] + + +def test_round_cut_resume_refuses_a_seek_that_lands_elsewhere(): + restore, fake, _, _ = _cut_resume_driver(ledger_cursor=None, landed_shift=1) + with pytest.raises(DriverError, match="the cut's cursor"): + restore(fake, 3) + + +def test_cut_data_cursor_needs_all_four_fields(): + from yeto.rl.adapters.miles.round_cut import _cut_data_cursor + from yeto.rl.engine.bridges import LocalOnlySync + + assert _cut_data_cursor(SimpleNamespace(data={**_CUT, "buffer_length": 23})) == _CUT + assert _cut_data_cursor(SimpleNamespace(data={"sample_offset": 35})) is None + assert LocalOnlySync(6).resume_data_cursor() is None # no round cuts wired diff --git a/tests/test_rl_sao_streaming_real_syncer.py b/tests/test_rl_sao_streaming_real_syncer.py index 7cf75e77..0758914e 100644 --- a/tests/test_rl_sao_streaming_real_syncer.py +++ b/tests/test_rl_sao_streaming_real_syncer.py @@ -29,7 +29,7 @@ _wait, syncer_binary, ) -from test_rl_launcher_sao_streaming import _declared_sao_mechanisms, _manifest, _sao_args # noqa: F401 +from test_rl_launcher_sao_streaming import _manifest, _sao_args from test_rl_sao_streaming_runtime import _syncer_profile pytestmark = pytest.mark.skipif(not Path(CARGO).exists(), reason="needs cargo to build the syncer") diff --git a/tests/test_rl_trajectory_evidence.py b/tests/test_rl_trajectory_evidence.py index fdcb275d..105c81c5 100644 --- a/tests/test_rl_trajectory_evidence.py +++ b/tests/test_rl_trajectory_evidence.py @@ -241,3 +241,96 @@ def test_schema_3_records_every_layered_field(): item = _batch().envelopes[0] assert set(asdict(item)) == set(trajectory_evidence._LAYERED_ENVELOPE_FIELDS) assert item.active_token_count == item.response_token_count + + +def _reward_function_sample(*, group_index, index, value, prompt="2+2?"): + """A sample scored by an ordinary reward function (e.g. GSM8K): no grader + verdict in the metadata, the reward is ``sample.reward``.""" + + return SimpleNamespace( + group_index=group_index, + index=index, + status=SimpleNamespace(value="completed"), + metadata={}, + prompt=[{"role": "user", "content": prompt}], + reward=value, + tokens=[101, 102 + index, 103 + index], + response_length=2, + rollout_log_probs=[-0.25, -0.5], + ) + + +def test_reward_function_evidence_binds_the_reward_function_output(tmp_path): + groups = [ + [ + _reward_function_sample(group_index=4, index=0, value=1.0), + _reward_function_sample(group_index=4, index=1, value=0.0), + ] + ] + evidence = build_trajectory_batch_evidence( + groups, + rollout_id=2, + behavior_policy_hash="a" * 64, + reward_contract_hash="b" * 64, + evidence_kind="reward-function", + behavior_policy_of=lambda _segments: (2, "a" * 64), + max_policy_age=0, + ) + + assert evidence.schema_version == 4 + assert evidence.evidence_kind == "reward-function" + assert [item.reward for item in evidence.envelopes] == [1.0, 0.0] + assert [item.behavior_policy_version for item in evidence.envelopes] == [2, 2] + # one prompt group shares one task identity + assert len({item.task_id for item in evidence.envelopes}) == 1 + assert evidence.envelopes[0].task_id.startswith("prompt-sha256:") + path = write_trajectory_batch_evidence(tmp_path / "private", evidence) + assert read_trajectory_batch_evidence(path) == evidence + + +def test_reward_function_evidence_hash_changes_with_the_reward(): + def build(value): + return build_trajectory_batch_evidence( + [[_reward_function_sample(group_index=0, index=0, value=value)]], + rollout_id=1, + behavior_policy_hash="a" * 64, + reward_contract_hash="b" * 64, + evidence_kind="reward-function", + ) + + assert build(1.0).input_batch_hash != build(0.0).input_batch_hash + + +@pytest.mark.parametrize("value", (None, True, float("nan"), {"score": 1.0})) +def test_reward_function_evidence_requires_a_finite_scalar_reward(value): + with pytest.raises(ValueError, match="finite scalar reward"): + build_trajectory_batch_evidence( + [[_reward_function_sample(group_index=0, index=0, value=value)]], + rollout_id=1, + behavior_policy_hash="a" * 64, + reward_contract_hash="b" * 64, + evidence_kind="reward-function", + ) + + +def test_grader_verdict_cannot_be_downgraded_to_reward_function_evidence(): + # A SecRLEnv sample keeps its trusted-grader evidence kind. + with pytest.raises(ValueError, match="grader verdict"): + build_trajectory_batch_evidence( + [[_sample(group_index=0, index=0, value=1.0)]], + rollout_id=1, + behavior_policy_hash="a" * 64, + reward_contract_hash="b" * 64, + evidence_kind="reward-function", + ) + + +def test_secrlenv_evidence_still_rejects_a_sample_without_a_verdict(): + with pytest.raises(reward.UntrustedOutcome): + build_trajectory_batch_evidence( + [[_reward_function_sample(group_index=0, index=0, value=1.0)]], + rollout_id=1, + behavior_policy_hash="a" * 64, + reward_contract_hash="b" * 64, + evidence_kind="secrlenv", + ) diff --git a/yeto/cli.py b/yeto/cli.py index 3b6a22fc..89b73686 100644 --- a/yeto/cli.py +++ b/yeto/cli.py @@ -1528,6 +1528,12 @@ def cmd_head(payload: str) -> int: from .gpu_spec import parse_gpu_spec args = argparse.Namespace(**json.loads(payload)) + if getattr(args, "rl_sao_contracts", None): + # The loaded SAO contract bundle is not JSON: reload it on the head + # from the workdir copy (its host paths are relative to the workdir). + from .rl.sao_streaming_manifest import load_sao_streaming_manifest + + args.rl_sao_manifest = load_sao_streaming_manifest(args.rl_sao_contracts) # Same count launcher.run uses: --gpu islands plus --external-learners # seats; without the seats the syncer rejects every manual joiner. num_learners = len(parse_gpu_spec(args.gpu)) + (getattr(args, "external_learners", 0) or 0) diff --git a/yeto/modal_runner.py b/yeto/modal_runner.py index d2becb1a..892b06e9 100644 --- a/yeto/modal_runner.py +++ b/yeto/modal_runner.py @@ -95,6 +95,10 @@ TAPE_MOUNT = "/yeto-tape" TAPE_SOURCE_DIR = "/root/yeto-output" TAPE_SYNC_INTERVAL_S = 30.0 +# Files outside the tape directory that the tape also mirrors (flattened to +# "."): a full-parameter island's trajectory evidence lives +# in a private mkdtemp directory under ~/yeto-rl that ends with the container. +TAPE_MIRROR_GLOBS = ("/root/yeto-rl/trajectory-evidence-*/rollout-*.json",) TAPE_SUFFIXES = (".jsonl", ".json", ".csv", ".log", ".txt", ".pt") # .pt: verl raw log-prob dumps TAPE_MAX_FILE_BYTES = 256 * 1024 * 1024 MODAL_CPU_CORES_PER_GPU = 4 @@ -410,8 +414,10 @@ class TapeSync: Copies whole files that changed (size or mtime); never deletes.""" def __init__(self, src: str, dst: str, commit, interval_s: float = TAPE_SYNC_INTERVAL_S, - extra: dict[str, str] | None = None) -> None: + extra: dict[str, str] | None = None, + mirror_globs: tuple[str, ...] = ()) -> None: self.src, self.dst, self.commit, self.interval_s = Path(src), Path(dst), commit, interval_s + self.mirror_globs = tuple(mirror_globs) # absolute globs, flattened into dst self.extra = dict(extra or {}) # file name -> text written once self._seen: dict[str, tuple[int, int]] = {} self._stop = threading.Event() @@ -454,24 +460,31 @@ def sync_once(self) -> int: copied = 0 if self.src.is_dir(): for f in sorted(self.src.iterdir()): - if not f.is_file() or not f.name.endswith(TAPE_SUFFIXES): - continue - st = f.stat() - if st.st_size > TAPE_MAX_FILE_BYTES: - continue - key = (st.st_size, st.st_mtime_ns) - if self._seen.get(f.name) == key: - continue - target = self._target_name(f.name) - tmp = self.dst / f".{target}.tmp" - tmp.write_bytes(f.read_bytes()) - os.replace(tmp, self.dst / target) - self._seen[f.name] = key - copied += 1 + if f.is_file() and f.name.endswith(TAPE_SUFFIXES): + copied += self._copy_if_changed(f, f.name) + for pattern in self.mirror_globs: + root = Path(pattern).anchor or "." + for f in sorted(Path(root).glob(str(Path(pattern).relative_to(root)))): + if f.is_file() and f.name.endswith(TAPE_SUFFIXES): + copied += self._copy_if_changed(f, f"{f.parent.name}.{f.name}") self.commit() self.syncs += 1 return copied + def _copy_if_changed(self, f: Path, name: str) -> int: + st = f.stat() + if st.st_size > TAPE_MAX_FILE_BYTES: + return 0 + key = (st.st_size, st.st_mtime_ns) + if self._seen.get(name) == key: + return 0 + target = self._target_name(name) + tmp = self.dst / f".{target}.tmp" + tmp.write_bytes(f.read_bytes()) + os.replace(tmp, self.dst / target) + self._seen[name] = key + return 1 + def _safe_sync(self) -> None: try: self.sync_once() @@ -736,7 +749,7 @@ def island_main(cfg_json: str) -> int: os.makedirs(TAPE_SOURCE_DIR, exist_ok=True) tape = TapeSync( TAPE_SOURCE_DIR, f"{TAPE_MOUNT}/{cfg.tape_subdir}/rank{rank}", - _volume_commit(cfg.tape_volume_name), + _volume_commit(cfg.tape_volume_name), mirror_globs=TAPE_MIRROR_GLOBS, ) hostmem = None interval = hostmem_interval_s(cfg) diff --git a/yeto/protocol.py b/yeto/protocol.py index e99ff521..7beb2254 100644 --- a/yeto/protocol.py +++ b/yeto/protocol.py @@ -428,6 +428,20 @@ class _StreamedInboundFragment: payload: StreamedInboundPayload +@dataclass(frozen=True) +class SendProgress: + """Outbound progress of one SyncerClient, for upload diagnostics. + + ``bytes_sent`` counts bytes handed to the kernel by ``sendall`` across + every connection group; ``queued_chunks`` is what the live group's send + queues still hold; ``reconnects`` counts redial attempts.""" + + bytes_sent: int + queued_chunks: int + reconnects: int + connected: bool + + @dataclass class _Outbound: data: bytes | bytearray @@ -546,6 +560,8 @@ def __init__( self._failure = threading.Event() # set by any thread of the live group self._closed = threading.Event() self._reconnects_used = 0 + self._bytes_sent = 0 + self._bytes_sent_lock = threading.Lock() self._reset_bcasts_on_reconnect = False self._supervisor: threading.Thread | None = None @@ -689,6 +705,8 @@ def _supervise(self) -> None: self._failure.wait() if self._closed.is_set() or self.shutdown.is_set(): return + _connect_log(f"connection to syncer {self.addr[0]}:{self.addr[1]} lost " + f"(learner {self.learner_id}): {self._last_err!r}; redialing") self._teardown_group() backoff = RECONNECT_BACKOFF_START while True: @@ -706,9 +724,13 @@ def _supervise(self) -> None: backoff = min(backoff * 2.0, RECONNECT_BACKOFF_CAP) try: self._connect_group(patient=False) + _connect_log(f"reconnected to syncer {self.addr[0]}:{self.addr[1]} " + f"(redial {self._reconnects_used})") break except (OSError, ConnectionError) as e: self._last_err = e + _connect_log(f"redial {self._reconnects_used} to syncer failed: " + f"{type(e).__name__}: {e}") def close(self) -> None: self._closed.set() @@ -743,6 +765,19 @@ def check_health(self) -> None: if self._err is not None: raise RuntimeError(f"syncer connection failed: {self._err}") from self._err + def send_progress(self) -> SendProgress: + with self._lock: + queued = sum(q.qsize() for q in self._queues) + connected = self._connected.is_set() + with self._bytes_sent_lock: + sent = self._bytes_sent + return SendProgress( + bytes_sent=sent, + queued_chunks=queued, + reconnects=self._reconnects_used, + connected=connected, + ) + # -- learner-facing API ------------------------------------------------------ def send_init(self, fragment_id: int, tensor_bytes: bytes) -> None: @@ -1309,6 +1344,8 @@ def _send_loop(self, gen: int, sock: socket.socket, q: queue.Queue) -> None: if item is None: return sock.sendall(item.data) + with self._bytes_sent_lock: + self._bytes_sent += len(item.data) if item.sent is not None: item.sent.set() except BaseException as e: diff --git a/yeto/rl/adapters/miles/entry.py b/yeto/rl/adapters/miles/entry.py index ef6093a3..76455ef1 100644 --- a/yeto/rl/adapters/miles/entry.py +++ b/yeto/rl/adapters/miles/entry.py @@ -666,15 +666,19 @@ def build_sync( return StrictAvgSync(miles_args.yeto_rl_bridge_config, progress=progress), progress +def _is_full_parameter_sync(sync: Any) -> bool: + from yeto.rl.engine.full_parameter_sync import FullParameterSync + + return isinstance(sync, FullParameterSync) + + def _policy_state_for(sync: Any, **lora: Any) -> Any: """The full-parameter actor cut for a FullParameterSync island, else LoRA state.""" - from yeto.rl.engine.full_parameter_sync import FullParameterSync - from .full_parameter.session import MilesFullParameterPolicyState from .state import MilesPolicyState - if isinstance(sync, FullParameterSync): + if _is_full_parameter_sync(sync): return MilesFullParameterPolicyState(sync) return MilesPolicyState(**lora) @@ -781,6 +785,7 @@ def expected_policy() -> tuple[int, str]: critic_model=critic_model, release_refs=release_refs, runner=runner, + install_actor_recorders=_is_full_parameter_sync(sync), ), policy_state=policy_state, publisher=MilesPublisher( diff --git a/yeto/rl/adapters/miles/full_parameter/role_stream.py b/yeto/rl/adapters/miles/full_parameter/role_stream.py index ce836254..e1c0f6b2 100644 --- a/yeto/rl/adapters/miles/full_parameter/role_stream.py +++ b/yeto/rl/adapters/miles/full_parameter/role_stream.py @@ -13,6 +13,7 @@ import asyncio import hashlib import json +import sys import time from collections.abc import Mapping, Sequence from dataclasses import asdict, dataclass, replace @@ -32,9 +33,12 @@ from yeto.rl.adapters.miles.full_parameter.inbound_sink import MilesRayInboundChunkSink from yeto.rl.engine.policy_digest import PolicyDigest, TrainerResidentState, WeightsChanged from yeto.rl.engine.trainable_state import LAYOUT_FRAGMENTS, UnsupportedLayoutError -from yeto.rl.local_learner import ComponentIdentity +from yeto.rl.local_learner import ComponentIdentity, layout_stream_role from yeto.rl.trajectory_evidence import TrajectoryBatchEvidence +# Seconds between progress lines while an island waits for its initial cut. +INITIAL_CUT_REPORT_SECONDS = 30.0 + _LOWER_HEX = frozenset("0123456789abcdef") @@ -318,7 +322,7 @@ async def initialize(self, group) -> None: group, policy_version=0, algorithm=self.config.algorithm, - stream_role=self.config.role, + stream_role=layout_stream_role(self.config.algorithm, self.config.role), components=(self.config.component,), minimum_fragments=self.config.expected_fragments, expected_fragments=self.config.expected_fragments, @@ -357,7 +361,14 @@ async def initialize(self, group) -> None: raise RuntimeError( f"{self.config.algorithm} {self.config.role} initial fragment was not queued" ) - authoritative = self._wait_for_initial_fragments(client, adapter) + # Wait off the event loop: an island initializes its roles + # concurrently, and each role's initial cut needs every island's + # INIT for that role. A blocking wait here would stop this + # island's other role from connecting, and two islands blocked + # on different roles deadlock (s23-g3-sao-main-20261014a). + authoritative = await asyncio.to_thread( + self._wait_for_initial_fragments, client, adapter + ) adapter.verify_initial_fragments(initial, authoritative) except BaseException: client.close() @@ -380,9 +391,12 @@ def _wait_for_initial_fragments( client, adapter: MilesChunkedFullParameterAdapter, ) -> tuple[StoredAuthoritativeFragment, ...]: - deadline = time.monotonic() + self.config.wait_timeout + started = time.monotonic() + deadline = started + self.config.wait_timeout + next_report = started + INITIAL_CUT_REPORT_SECONDS observed: dict[int, StoredAuthoritativeFragment] = {} count = adapter.layout.fragments.num_fragments + self._report_initial_cut(client, observed, count, started, "waiting") while len(observed) < count: client.check_health() for update in client.drain_updates(): @@ -393,13 +407,50 @@ def _wait_for_initial_fragments( ) observed[stored.fragment_id] = stored if len(observed) < count: - if time.monotonic() >= deadline: + now = time.monotonic() + if now >= next_report: + self._report_initial_cut(client, observed, count, started, "waiting") + next_report = now + INITIAL_CUT_REPORT_SECONDS + if now >= deadline: + self._report_initial_cut(client, observed, count, started, "timed out") raise TimeoutError( f"{self.config.algorithm} {self.config.role} timed out waiting for initial cut" ) time.sleep(self.config.poll_seconds) + self._report_initial_cut(client, observed, count, started, "received") return tuple(observed[index] for index in range(count)) + def _report_initial_cut( + self, + client, + observed: Mapping[int, StoredAuthoritativeFragment], + count: int, + started: float, + state: str, + ) -> None: + """One stderr line on the initial cut: how much of INIT_PARAMS left + this island and how much of the broadcast came back.""" + + progress_of = getattr(client, "send_progress", None) + progress = progress_of() if callable(progress_of) else None + sent = ( + "" + if progress is None + else ( + f"; sent {progress.bytes_sent / 2**20:.1f} MiB, " + f"{progress.queued_chunks} chunk(s) queued, " + f"{progress.reconnects} redial(s), " + f"connected={progress.connected}" + ) + ) + print( + f"[yeto-initial-cut] learner {self.config.learner_id} {state}: " + f"{len(observed)}/{count} fragment(s) after " + f"{time.monotonic() - started:.0f}s{sent}", + file=sys.stderr, + flush=True, + ) + @staticmethod def _stored_update(update) -> StoredAuthoritativeFragment: if not getattr(update, "stored", False) or not isinstance( diff --git a/yeto/rl/adapters/miles/in_flight.py b/yeto/rl/adapters/miles/in_flight.py index 8f1290cd..44a32207 100644 --- a/yeto/rl/adapters/miles/in_flight.py +++ b/yeto/rl/adapters/miles/in_flight.py @@ -170,6 +170,7 @@ def import_in_flight(executor: Any, entries: Sequence[Mapping[str, Any]], *, max groups.setdefault(raw["group_id"], []).append((int(state.get("group_position", 0)), state.get("miles_sample"))) sizes[raw["group_id"]] = int(state.get("group_size", 0)) restored: list[list[Any]] = [] + restored_ids: list[str] = [] if groups: if sample_from_dict is None: from miles.utils.types import Sample @@ -181,6 +182,7 @@ def import_in_flight(executor: Any, entries: Sequence[Mapping[str, Any]], *, max discarded_partial += len(members) continue restored.append([sample_from_dict(m) for _, m in sorted(members, key=lambda pm: pm[0])]) + restored_ids.append(group_id) source = getattr(executor, "data_source", None) source = getattr(source, "__self__", source) if restored: @@ -193,6 +195,7 @@ def import_in_flight(executor: Any, entries: Sequence[Mapping[str, Any]], *, max return { "resumed": report["resumed"] - discarded_partial, "resumed_groups": len(restored), + "resumed_group_ids": restored_ids, "discarded": report["discarded"] + len(agentic) + discarded_partial, "discarded_tokens": report["discarded_tokens"], "reasons": reasons, diff --git a/yeto/rl/adapters/miles/island_entry.py b/yeto/rl/adapters/miles/island_entry.py index b20634a5..2532d2e1 100644 --- a/yeto/rl/adapters/miles/island_entry.py +++ b/yeto/rl/adapters/miles/island_entry.py @@ -657,6 +657,24 @@ def _check_dense_full_sweep(args, *, yeto_policy_sync: bool) -> None: raise ValueError("dense full-parameter mode requires --syncer-profile-sha256") +def _trajectory_evidence_kind(args) -> str: + """The evidence kind of a dense full-parameter island, chosen by its reward. + + The SecRLEnv reward binds the grader's signed verdict as evidence. Any + other reward function (for example a GSM8K answer check) binds its own + output, ``sample.reward``; a sample that still carries a grader verdict is + rejected there. Terminal-Bench 2.1 runs the SAO sync, whose runtime + contract names its own evidence kind. + """ + + from yeto.rl import SECRLENV_REWARD + from yeto.rl.trajectory_evidence import REWARD_FUNCTION_EVIDENCE_KIND + + if args.reward_function == SECRLENV_REWARD: + return "secrlenv" + return REWARD_FUNCTION_EVIDENCE_KIND + + def _private_trajectory_evidence_dir(args) -> Path: """A fresh owner-only directory for this island's trajectory evidence.""" @@ -1839,7 +1857,7 @@ def run_miles( miles_args.yeto_rl_trajectory_evidence_dir = str( _private_trajectory_evidence_dir(args) ) - miles_args.yeto_rl_trajectory_evidence_kind = "secrlenv" + miles_args.yeto_rl_trajectory_evidence_kind = _trajectory_evidence_kind(args) miles_args.yeto_rl_trajectory_evidence_schema_version = 1 setattr( miles_args, diff --git a/yeto/rl/adapters/miles/models/chunked_full_parameter.py b/yeto/rl/adapters/miles/models/chunked_full_parameter.py index 64e82454..9b7e6c99 100644 --- a/yeto/rl/adapters/miles/models/chunked_full_parameter.py +++ b/yeto/rl/adapters/miles/models/chunked_full_parameter.py @@ -3,8 +3,9 @@ from __future__ import annotations import hashlib +import json from collections.abc import Callable, Sequence -from dataclasses import dataclass, replace +from dataclasses import asdict, dataclass, replace from typing import Any, Protocol from yeto.rl.contracts import LocalStepReceipt @@ -135,6 +136,60 @@ def release_all(self) -> int: return sum(fragment.release() for fragment in self._stored.values()) +LAYOUT_PROBE_LOG_PREFIX = "[yeto-layout-probe] " + + +def layout_attestation_record( + adapter: MilesChunkedFullParameterAdapter, + components: Sequence[ComponentIdentity], +) -> dict[str, object]: + """One role's layout in the role shape of a SAO layout attestation. + + The fields are those ``yeto.rl.sao_streaming_runtime`` checks for each + role (``miles.sao-streaming-layouts.v2``); ``component`` is the single + component the layout was derived for. + """ + + if len(components) != 1: + raise ValueError("a layout attestation record names exactly one component") + (component,) = components + layout = adapter.layout + fragments = [] + for fragment_id, fragment in enumerate(layout.fragments.fragments): + role, shard_id = layout.fragment_owner(fragment_id) + fragments.append( + { + "fragment_id": fragment_id, + "role": role, + "shard_id": shard_id, + "parameter_count": len(layout.fragment_specs(fragment_id)), + "scalar_count": fragment.numel, + "payload_bytes": fragment.numel * 4, + } + ) + return { + "role": component.role, + "component": { + "model_revision": component.model_revision, + "config_sha256": component.config_hash, + }, + "parameter_layout_sha256": layout.layout_hash, + "expected_fragments": layout.fragments.num_fragments, + "parameter_tensor_count": adapter.expected_parameter_tensor_count, + "parameter_scalar_count": adapter.expected_parameter_scalar_count, + "fragments": fragments, + "manifests": [ + { + "topology": asdict(manifest.topology), + "source_layout_sha256": manifest.layout_hash, + "parameter_count": len(manifest.specs), + "scalar_count": sum(spec.numel for spec in manifest.specs), + } + for manifest in adapter.manifests + ], + } + + @dataclass(frozen=True) class MilesChunkedFullParameterAdapter: layout: ParameterLayout @@ -185,6 +240,18 @@ async def initialize( fragment_bytes = tuple( fragment.numel * 4 for fragment in adapter.layout.fragments.fragments ) + # The refusal doubles as a layout probe: print the observed layout + # in the shape of a layout attestation, so a probe run with a + # placeholder hash yields the real values to freeze. + print( + LAYOUT_PROBE_LOG_PREFIX + + json.dumps( + layout_attestation_record(adapter, components), + sort_keys=True, + separators=(",", ":"), + ), + flush=True, + ) raise RuntimeError( "Miles full-parameter layout differs from the frozen probe: " f"expected_fragments={expected_fragments}, " diff --git a/yeto/rl/adapters/miles/round_cut.py b/yeto/rl/adapters/miles/round_cut.py index f7167b9e..48880222 100644 --- a/yeto/rl/adapters/miles/round_cut.py +++ b/yeto/rl/adapters/miles/round_cut.py @@ -70,6 +70,8 @@ def __init__(self, *, controller: Any, source: Any, trainer: Any, every: int = 1 self.allow_config_change = bool(allow_config_change) self.next_lr = next_lr # callable(driver) -> list[float] | None self.pending_lr_check: dict[str, Any] | None = None + # set by resume(): where the rollout data source continues (the cut's cursor) + self.resume_data_cursor: dict[str, int] | None = None # ---------------------------------------------------------------- pointer @property @@ -247,6 +249,7 @@ def resume(self, driver: Any) -> dict[str, Any] | None: in_flight = self._restore_in_flight(driver, manifest, rid) if in_flight is not None: checks["in_flight"] = in_flight + self.resume_data_cursor = _cut_data_cursor(manifest) driver.local_step = int(info["local_step"]) self.resumed = dict(info) if info.get("lr_at_next_round") is not None: @@ -315,6 +318,21 @@ def check_first_round(self, driver: Any, *, rollout_id: int, applied_lrs: Any) - raise RoundCutError(problem) +def _cut_data_cursor(manifest: Any) -> dict[str, int] | None: + """The data source position recorded at the cut (None when incomplete). + + Under fully_async the background worker keeps drawing prompts after the + round's trained groups were reported, and the cut puts the untrained ones + back into the data buffer. The position to resume from is therefore the + one read at the cut, not the one the batch ledger recorded for the round.""" + from .rollout_meta_hook import _CURSOR_FIELDS + + data = dict(getattr(manifest, "data", None) or {}) + if not all(isinstance(data.get(f), int) for f in _CURSOR_FIELDS): + return None + return {f: int(data[f]) for f in _CURSOR_FIELDS} + + def wire_round_cuts(driver: Any, *, controller: Any, source: Any, **options: Any) -> RoundCutCheckpoint | None: """Give a :class:`~..bridges.LocalOnlySync` driver round cuts when a checkpoint store is configured (other syncs own the policy: their syncer is authoritative).""" diff --git a/yeto/rl/adapters/miles/state_plugin.py b/yeto/rl/adapters/miles/state_plugin.py index f8fc58e4..60f92f6f 100644 --- a/yeto/rl/adapters/miles/state_plugin.py +++ b/yeto/rl/adapters/miles/state_plugin.py @@ -70,6 +70,9 @@ def weights_version(actor: Any) -> dict[str, Any] | None: ALIGN_WEIGHT_VERSION = f"{_PLUGIN_MODULE}.align_weight_version" APPLIED_LRS = f"{_PLUGIN_MODULE}.applied_lrs" STEP_LOSSES = f"{_PLUGIN_MODULE}.step_losses" +# Run in the actor's processes before its first train: a full-parameter actor +# never calls a LoRA state plugin, which would otherwise install the recorders. +ACTOR_RECORDERS = f"{_PLUGIN_MODULE}.install_actor_recorders" # rl-algo-critic-family 3.2: run in the critic's processes before its train. CRITIC_RECORDERS = f"{_PLUGIN_MODULE}.install_critic_recorders" EXPLAINED_VARIANCE_KEY = "explained_variance" @@ -1096,8 +1099,8 @@ def _apply_state( def install_grad_norm_recorder() -> bool: """Wrap upstream ``train_one_step`` to record the grad norm it returns. - Idempotent; called from every state plugin, and the driver always exports - the trainable state before the first training step. + Idempotent; called from every state plugin, and the trainer installs it in + every actor process before the first training step (ACTOR_RECORDERS). """ global _RECORDER_INSTALLED @@ -1632,6 +1635,13 @@ def restore_critic_cut(actor: Any, **kwargs: Any) -> dict[str, Any]: return _restore_critic_cut(actor, **kwargs) +def install_actor_recorders(actor: Any) -> bool: + """Plugin: the per-step recorders in an actor process (idempotent).""" + + del actor + return install_grad_norm_recorder() + + def install_critic_recorders(actor: Any) -> bool: """Plugin: the per-step recorders in a critic process (idempotent).""" diff --git a/yeto/rl/adapters/miles/trainer.py b/yeto/rl/adapters/miles/trainer.py index ac91a4f4..f8a62240 100644 --- a/yeto/rl/adapters/miles/trainer.py +++ b/yeto/rl/adapters/miles/trainer.py @@ -31,6 +31,7 @@ from . import LoopRunner from .rollout import policy_token, require_policy_tokens from .state_plugin import ( + ACTOR_RECORDERS, APPLIED_LRS, CRITIC_RECORDERS, CRITIC_STATE_SUMMARY, @@ -264,6 +265,7 @@ def __init__( spec: Any = None, critic_model: Any = None, release_outputs: Callable[[Any], None] | None = None, + install_actor_recorders: bool = False, ) -> None: self._spec = spec # rl-algo-critic-family 3.1: shared actor/critic PPO (Miles train.py @@ -289,6 +291,10 @@ def __init__( # (``last_step_losses`` stays gated on the mechanisms that consume it). self.last_round_step_losses: list[dict[str, Any]] | None = None self.optimizer_steps_total = 0 + # A LoRA actor installs its step recorders through the state plugins + # the publisher calls before the first step; a full-parameter actor + # calls none of them, so the trainer installs the recorders itself. + self._actor_recorders_installed = not install_actor_recorders def train_step(self, batch: RolloutBatchHandle) -> LocalStepReceipt: if batch.payload is None: @@ -305,6 +311,7 @@ def train_step(self, batch: RolloutBatchHandle) -> LocalStepReceipt: if self._check_tokens: # spec: reject before training require_policy_tokens(batch, policy_token(batch.rollout_id, batch.policy_hash)) critic_outputs = self._train_critic(batch) if self._critic is not None else None + self._install_actor_recorders() if critic_outputs is None: outputs = self._run(self._actor.train(batch.rollout_id, batch.payload)) else: @@ -387,6 +394,18 @@ def train_step(self, batch: RolloutBatchHandle) -> LocalStepReceipt: parameter_layout_hash=self._layout_hash(), ) + def _install_actor_recorders(self) -> None: + """Record every optimizer step's applied LR and grad norm in the actor + processes, once per trainer; without them ``_applied_lrs`` finds no + steps.""" + + if self._actor_recorders_installed: + return + installed = list(self._run(self._actor.run_plugin(ACTOR_RECORDERS, {}))) + if not installed or not all(installed): + raise TrainStepError(f"actor step recorders could not be installed: {installed}") + self._actor_recorders_installed = True + def _workers(self) -> int: return int(getattr(self._args, "actor_num_nodes", 1) or 1) * int( getattr(self._args, "actor_num_gpus_per_node", 1) or 1 diff --git a/yeto/rl/engine/algorithm.py b/yeto/rl/engine/algorithm.py index d9574fa8..4ef3d3cd 100644 --- a/yeto/rl/engine/algorithm.py +++ b/yeto/rl/engine/algorithm.py @@ -1792,10 +1792,13 @@ def resolve_ports_algorithm(args: Any) -> "AlgorithmSpec": # rl-algo-critic-family (user decision 2026-10-07): the critic mechanisms may be # allowed (still explicitly, via --rl-allow-unverified-mechanism) on several -# islands / with outer sync, but only under the strict-avg preset, whose -# critic channel is the second syncer (DualStrictAvgSync, design D4). +# islands / with outer sync, but only under a preset with a critic channel: +# strict-avg, whose critic channel is the second syncer (DualStrictAvgSync, +# design D4), and sao-streaming-full, whose critic role has its own syncer +# (P0-D8, 2026-10-13; the preset requires the critic). # decoupled + critic stays refused (critic_run_problems). Every other # unverified mechanism keeps D11 as written. +CRITIC_CHANNEL_SYNC_PRESETS = frozenset({"strict-avg", "sao-streaming-full"}) CRITIC_STRICT_AVG_ALLOWANCES = frozenset({ "advantage_estimators:ppo", "execution:critic", @@ -1815,8 +1818,8 @@ def check_unverified_allowance( ) -> tuple[str, ...]: """D11 as written: refused with multiple islands *or* any outer sync. - Exception (critic family, design D4): when ``sync_preset == "strict-avg"`` - and every name is in :data:`CRITIC_STRICT_AVG_ALLOWANCES` (with + Exception (critic family, design D4): when ``sync_preset`` has a critic + channel (:data:`CRITIC_CHANNEL_SYNC_PRESETS`) and every name is in :data:`CRITIC_STRICT_AVG_ALLOWANCES` (with ``execution:critic`` among them), the allowance is accepted on several islands / with outer sync. ``sync_preset=None`` (caller does not know the preset) keeps D11 as written. @@ -1834,7 +1837,7 @@ def check_unverified_allowance( f"(known: {sorted(allowance_names())})" ) if islands != 1 or outer_sync: - if (sync_preset == "strict-avg" and "execution:critic" in names + if (sync_preset in CRITIC_CHANNEL_SYNC_PRESETS and "execution:critic" in names and set(names) <= CRITIC_STRICT_AVG_ALLOWANCES): return names raise AlgorithmSpecError( diff --git a/yeto/rl/engine/bridges.py b/yeto/rl/engine/bridges.py index 892001a8..324ade1d 100644 --- a/yeto/rl/engine/bridges.py +++ b/yeto/rl/engine/bridges.py @@ -131,6 +131,12 @@ def at_safe_point(self, driver: IslandDriver, *, rollout_id: int, final: bool = "policy_hash", "store_synced", "latest_seq", "pruned", "incarnation_index", "final", "trainer_onloaded_for_cut")}) + def resume_data_cursor(self) -> dict[str, int] | None: + """The data cursor recorded by the round cut this run resumed from (None: + no resume, or the cut recorded none). The driver seeks to it first.""" + cursor = getattr(self.round_cuts, "resume_data_cursor", None) + return None if cursor is None else dict(cursor) + def _end(self) -> int: return self.num_rollout if self.stop_after is None else min(self.num_rollout, int(self.stop_after)) diff --git a/yeto/rl/engine/driver.py b/yeto/rl/engine/driver.py index bc2ea7c0..f44e29cc 100644 --- a/yeto/rl/engine/driver.py +++ b/yeto/rl/engine/driver.py @@ -925,6 +925,14 @@ def _generate(self, rollout_id: int) -> RolloutBatchHandle: ) if not batch.groups: raise PolicyIdentityError(f"rollout {rollout_id} produced no complete group") + if limit > 0: + # P0-S evidence: the generating policy of every accepted group (its + # label carries the weight hash of that version), so a round's + # trajectories can be matched against the publications. + self.emit("rl_group_policies", rollout_id=rollout_id, groups=[ + [g.group_id, g.policy_token, + None if g.policy_versions is None else sorted(set(g.policy_versions))] + for g in batch.groups]) if self.eval_overlap is not None: self.eval_overlap.after_generate( rollout_id, token=self.expected_token, published_version=self.published_version @@ -1692,8 +1700,22 @@ def _restore_data_cursor(self, start_rollout_id: int) -> None: pool that reports cursors -> DriverError (no silent re-training).""" if start_rollout_id <= 0: return - cursor = self.ledger.restart_cursor(start_rollout_id) if self.ledger is not None else None seek = getattr(self.rollout, "seek_data_cursor", None) + from_cut = getattr(self.sync, "resume_data_cursor", None) + cut_cursor = from_cut() if callable(from_cut) else None + if cut_cursor is not None and callable(seek): + # Resume from a round cut: the cut read the data source at the safe + # point (P0-S: after fully_async put its untrained groups back), so + # its cursor wins over the ledger's per-round record. + landed = seek(cut_cursor) + landed = None if landed is None else {k: int(v) for k, v in dict(landed).items()} + if landed != cut_cursor: + raise DriverError(f"restart at rollout {start_rollout_id}: data source seek " + f"to the cut's cursor {cut_cursor} landed on {landed}") + self.emit("rl_data_cursor_restored", rollout_id=start_rollout_id, source="round_cut", + data_cursor=dict(landed)) + return + cursor = self.ledger.restart_cursor(start_rollout_id) if self.ledger is not None else None fallback = getattr(self.sync, "restart_cursor_fallback", None) # The whole-round fallback is for an island with NO record of rollout v-1 # (empty/absent ledger: fresh container, joined at base_version > 0). A diff --git a/yeto/rl/local_learner.py b/yeto/rl/local_learner.py index 5995bcdf..d4c7ba34 100644 --- a/yeto/rl/local_learner.py +++ b/yeto/rl/local_learner.py @@ -32,6 +32,20 @@ _SHA256 = re.compile(r"[0-9a-f]{64}\Z") +def layout_stream_role(algorithm: str, role: str) -> str | None: + """The ``stream_role`` a per-role parameter stream lays out with. + + Only role-lane algorithms (SAO) split their roles into separate layouts. + A single-role algorithm streams its one role with the plain layout, whose + hash carries no role field; ``role`` must still be that algorithm's role. + """ + + roles = _ROLES_BY_ALGORITHM.get(algorithm) + if roles is None or role not in roles: + raise ValueError(f"{algorithm!r} has no trainable role {role!r}") + return role if algorithm in _ROLE_LANE_ALGORITHMS else None + + @dataclass(frozen=True, order=True) class ComponentIdentity: """Immutable identity of one trainable algorithm component.""" diff --git a/yeto/rl/trajectory_evidence.py b/yeto/rl/trajectory_evidence.py index 1d71d9c8..7ee7c67d 100644 --- a/yeto/rl/trajectory_evidence.py +++ b/yeto/rl/trajectory_evidence.py @@ -17,6 +17,16 @@ ports rollout hook) and schema 3 otherwise. Schema 1 and 2 files remain readable: their structure and batch hash are checked, but the HMAC that was mixed into their outcome hashes is not (and cannot be) re-verified. + +Evidence kinds name where a trajectory's reward comes from: + +* ``secrlenv`` and ``terminal-bench-2.1`` -- a trusted grader verdict carried + in the sample metadata; the reward must equal the verdict. +* ``reward-function`` -- the output of the island's hash-bound reward function + (``sample.reward``), e.g. a GSM8K answer check. There is no separate + verdict to verify: the evidence binds the reward, the prompt and the tokens, + and the reward function itself is bound by ``reward_contract_hash``. + Trajectories of this kind must not carry a grader verdict. """ from __future__ import annotations @@ -40,7 +50,9 @@ _LAYERED_SCHEMAS = frozenset({_SIGNED_LAYERED_SCHEMA, _SCHEMA, _PER_TRAJECTORY_POLICY_SCHEMA}) _READABLE_SCHEMAS = frozenset({_LEGACY_SCHEMA, *_LAYERED_SCHEMAS}) _SHA256 = frozenset("0123456789abcdef") -EVIDENCE_KINDS = frozenset({"secrlenv", "terminal-bench-2.1"}) +REWARD_FUNCTION_EVIDENCE_KIND = "reward-function" +EVIDENCE_KINDS = frozenset({"secrlenv", "terminal-bench-2.1", REWARD_FUNCTION_EVIDENCE_KIND}) +_GRADER_VERDICT_KEYS = ("secrlenv_trusted_outcome", "tbench_trusted_outcome") # The launch context still names the signed-era layout (1 or 2) it was # attested with; it must be bound, but new evidence is schema 3 or 4. LAUNCH_CONTEXT_SCHEMA_VERSIONS = frozenset({_LEGACY_SCHEMA, _SIGNED_LAYERED_SCHEMA}) @@ -218,7 +230,7 @@ def build_trajectory_batch_evidence( "compaction-marked trajectory was not preserved as segments" ) - verified = _verified_outcome(metadata, evidence_kind) + verified = _trajectory_outcome(sample, metadata, evidence_kind) outcome_identity = ( verified.task_id, verified.episode_id, @@ -457,9 +469,19 @@ def read_trajectory_batch_evidence(path: str | Path) -> TrajectoryBatchEvidence: return evidence +def _trajectory_outcome( + sample: Any, metadata: Mapping[str, Any], evidence_kind: str +) -> _VerifiedOutcome: + if evidence_kind == REWARD_FUNCTION_EVIDENCE_KIND: + return _reward_function_outcome(sample, metadata) + return _verified_outcome(metadata, evidence_kind) + + def _verified_outcome( metadata: Mapping[str, Any], evidence_kind: str ) -> _VerifiedOutcome: + """The trusted grader verdict of a ``secrlenv`` or Terminal-Bench sample.""" + if evidence_kind == "terminal-bench-2.1": if "secrlenv_trusted_outcome" in metadata: raise ValueError("trajectory metadata mixes benchmark evidence kinds") @@ -503,6 +525,51 @@ def _verified_outcome( ) +def _reward_function_outcome( + sample: Any, metadata: Mapping[str, Any] +) -> _VerifiedOutcome: + """The outcome of a trajectory scored by the island's reward function. + + The reward is ``sample.reward`` as the reward function wrote it. The task + is the prompt (by content hash) and the episode is the logical sample. + """ + + if any(key in metadata for key in _GRADER_VERDICT_KEYS): + raise ValueError( + "reward-function trajectory carries a grader verdict; " + "bind the grader's evidence kind instead" + ) + reward = getattr(sample, "reward", None) + if ( + isinstance(reward, bool) + or not isinstance(reward, (int, float)) + or not math.isfinite(float(reward)) + ): + raise ValueError("reward-function trajectory has no finite scalar reward") + prompt = json.dumps( + getattr(sample, "prompt", None), sort_keys=True, separators=(",", ":") + ).encode("utf-8") + task_id = "prompt-sha256:" + hashlib.sha256(prompt).hexdigest() + episode_id = f"sample:{getattr(sample, 'index', None)}" + canonical = json.dumps( + { + "kind": REWARD_FUNCTION_EVIDENCE_KIND, + "task_id": task_id, + "episode_id": episode_id, + "reward": float(reward), + }, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return _VerifiedOutcome( + task_id=task_id, + episode_id=episode_id, + sample_id=None, + reward=float(reward), + canonical=canonical, + ) + + def _compaction_identity( metadata: Mapping[str, Any], *, @@ -639,11 +706,10 @@ def _active_token_evidence( def _outcome_evidence_hash( evidence_kind: str, verified: _VerifiedOutcome ) -> str: - domain = ( - b"yeto-tbench-verdict-evidence-v2\0" - if evidence_kind == "terminal-bench-2.1" - else b"yeto-secrlenv-cleanup-evidence-v2\0" - ) + domain = { + "terminal-bench-2.1": b"yeto-tbench-verdict-evidence-v2\0", + REWARD_FUNCTION_EVIDENCE_KIND: b"yeto-reward-function-evidence-v1\0", + }.get(evidence_kind, b"yeto-secrlenv-cleanup-evidence-v2\0") return _hash_parts(domain, verified.canonical)