PR 17: Full-parameter, SAO streaming and P0-S fixes from the S23 GPU runs

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:<hex>`), the logical sample (episode id `sample:<index>`), 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.

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.
 

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)

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")

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)

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)

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
 

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"
+    )

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",

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