From 55e7cdc4f7a2634940bc9f3d5ab7c923f7760273 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 04:55:04 +0000 Subject: [PATCH 01/13] Lay out a single-role full-parameter stream without a role lane The dense full-parameter (grpo) actor stream passed stream_role="actor" to ParameterLayout.create, which accepts a stream role only for role-lane algorithms (SAO). Island initialization failed before the layout hash check (P0-1.11 G3-fp probe). layout_stream_role now maps a single-role algorithm to the plain layout, whose hash carries no role field. Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_full_parameter_sync.py | 50 +++++++++++++++++++ .../miles/full_parameter/role_stream.py | 4 +- yeto/rl/local_learner.py | 14 ++++++ 3 files changed, 66 insertions(+), 2 deletions(-) diff --git a/tests/test_rl_full_parameter_sync.py b/tests/test_rl_full_parameter_sync.py index d493513a..78b4835c 100644 --- a/tests/test_rl_full_parameter_sync.py +++ b/tests/test_rl_full_parameter_sync.py @@ -677,3 +677,53 @@ 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 diff --git a/yeto/rl/adapters/miles/full_parameter/role_stream.py b/yeto/rl/adapters/miles/full_parameter/role_stream.py index ce836254..1db44e33 100644 --- a/yeto/rl/adapters/miles/full_parameter/role_stream.py +++ b/yeto/rl/adapters/miles/full_parameter/role_stream.py @@ -32,7 +32,7 @@ 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 _LOWER_HEX = frozenset("0123456789abcdef") @@ -318,7 +318,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, 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.""" From 79a8140929737c515cb6427dfd3eb4a3ce70ec60 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 05:40:51 +0000 Subject: [PATCH 02/13] Bind reward-function trajectory evidence for non-grader full-parameter islands A dense full-parameter island hard-coded the SecRLEnv evidence kind, so an island scored by an ordinary reward function (GSM8K) failed its first round with UntrustedOutcome. Add the reward-function evidence kind, which binds sample.reward, the prompt and the tokens, and choose the kind from the island's reward function. SecRLEnv keeps its signed-verdict evidence, and a sample that carries a grader verdict cannot be bound as reward-function evidence. Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_launcher_dense_full.py | 21 ++++++ tests/test_rl_trajectory_evidence.py | 93 ++++++++++++++++++++++++++ yeto/rl/adapters/miles/island_entry.py | 20 +++++- yeto/rl/trajectory_evidence.py | 74 +++++++++++++++++--- 4 files changed, 199 insertions(+), 9 deletions(-) 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_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/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/trajectory_evidence.py b/yeto/rl/trajectory_evidence.py index 1d71d9c8..f036e9c3 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 = _verified_outcome(sample, metadata, evidence_kind) outcome_identity = ( verified.task_id, verified.episode_id, @@ -458,8 +470,10 @@ def read_trajectory_batch_evidence(path: str | Path) -> TrajectoryBatchEvidence: def _verified_outcome( - metadata: Mapping[str, Any], evidence_kind: str + sample: Any, metadata: Mapping[str, Any], evidence_kind: str ) -> _VerifiedOutcome: + if evidence_kind == REWARD_FUNCTION_EVIDENCE_KIND: + return _reward_function_outcome(sample, metadata) if evidence_kind == "terminal-bench-2.1": if "secrlenv_trusted_outcome" in metadata: raise ValueError("trajectory metadata mixes benchmark evidence kinds") @@ -503,6 +517,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 +698,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) From 9fc18685dd9a76c1824207d09026c0a40fdc70c0 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 05:41:15 +0000 Subject: [PATCH 03/13] Specify reward-function trajectory evidence in rl-drop-reward-tamper-hash Co-Authored-By: Claude Opus 5.5 --- .../changes/rl-drop-reward-tamper-hash/design.md | 1 + .../specs/rl-reward-integrity-scope/spec.md | 15 +++++++++++++++ .../changes/rl-drop-reward-tamper-hash/tasks.md | 2 ++ 3 files changed, 18 insertions(+) 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) From ed228721d51e89566306286311c7787aed791183 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 05:49:36 +0000 Subject: [PATCH 04/13] Keep the grader-verdict check's signature for its direct callers Co-Authored-By: Claude Opus 5.5 --- yeto/rl/trajectory_evidence.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/yeto/rl/trajectory_evidence.py b/yeto/rl/trajectory_evidence.py index f036e9c3..7ee7c67d 100644 --- a/yeto/rl/trajectory_evidence.py +++ b/yeto/rl/trajectory_evidence.py @@ -230,7 +230,7 @@ def build_trajectory_batch_evidence( "compaction-marked trajectory was not preserved as segments" ) - verified = _verified_outcome(sample, metadata, evidence_kind) + verified = _trajectory_outcome(sample, metadata, evidence_kind) outcome_identity = ( verified.task_id, verified.episode_id, @@ -469,11 +469,19 @@ def read_trajectory_batch_evidence(path: str | Path) -> TrajectoryBatchEvidence: return evidence -def _verified_outcome( +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") From 9cad3b077af9baf99c05d467671a09373f559bf1 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 06:24:14 +0000 Subject: [PATCH 05/13] Install the actor step recorders for a full-parameter trainer The per-step recorders (applied LR, grad norm) were installed only by the LoRA state plugins, which a full-parameter actor never calls. The first full-parameter round then found no applied learning rate and failed. A trainer of a FullParameterSync island now installs them once, before its first training step, and refuses to train when they cannot be installed. The LoRA path's RPC sequence is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../test_rl_miles_adapter_trainer_publish.py | 56 ++++++++++++++++++- yeto/rl/adapters/miles/entry.py | 11 +++- yeto/rl/adapters/miles/state_plugin.py | 14 ++++- yeto/rl/adapters/miles/trainer.py | 19 +++++++ 4 files changed, 94 insertions(+), 6 deletions(-) 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/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/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 From 62b2524850760fb06826a9f760bf66b08b0abab9 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 07:20:01 +0000 Subject: [PATCH 06/13] Report initial-cut upload progress and mirror trajectory evidence to the tape Run s23-g3-fp-main-20261014c timed out after 900 s waiting for the initial cut: learner 0 had queued all INIT_PARAMS chunks, but the syncer never completed the fragments, and no log showed how many bytes had left the island. The client now counts the bytes its sender threads wrote and logs each lost connection, redial and redial failure. A full-parameter stream prints one line every 30 s while it waits for the initial cut, with the fragments received, the bytes sent, the chunks still queued and the redials. A full-parameter island writes its trajectory evidence to a private directory under ~/yeto-rl, which ended with the Modal container. The tape now also mirrors those rollout files, flattened to ".", so the evidence can be read after the run. Co-Authored-By: Claude Opus 5.5 --- tests/test_modal_runner.py | 19 ++++++++ tests/test_protocol_finalization.py | 36 ++++++++++++++ tests/test_rl_full_parameter_sync.py | 39 +++++++++++++++ yeto/modal_runner.py | 45 ++++++++++------- yeto/protocol.py | 37 ++++++++++++++ .../miles/full_parameter/role_stream.py | 48 ++++++++++++++++++- 6 files changed, 206 insertions(+), 18 deletions(-) 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_full_parameter_sync.py b/tests/test_rl_full_parameter_sync.py index 78b4835c..c91dba3e 100644 --- a/tests/test_rl_full_parameter_sync.py +++ b/tests/test_rl_full_parameter_sync.py @@ -727,3 +727,42 @@ async def fake_initialize(group, **kwargs): with pytest.raises(Stop): asyncio.run(module.MilesFullParameterRoleStream(config).initialize(object())) assert seen["stream_role"] == expected + + +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/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/full_parameter/role_stream.py b/yeto/rl/adapters/miles/full_parameter/role_stream.py index 1db44e33..c54abcb3 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 @@ -35,6 +36,9 @@ 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") @@ -380,9 +384,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 +400,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( From 5de23ea8e3ededbafc95826e8c5cfe228fe0d81a Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 08:30:50 +0000 Subject: [PATCH 07/13] Print the observed full-parameter layout as an attestation record on refusal A launch with a placeholder layout hash is refused before any plan is installed. The refusal now also prints the observed layout, per role, in the role shape of a SAO streaming layout attestation, so a ports island can serve as the layout probe for sao-streaming-full contracts. Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_miles_chunked_full_parameter.py | 57 +++++++++++++++ .../miles/models/chunked_full_parameter.py | 69 ++++++++++++++++++- 2 files changed, 125 insertions(+), 1 deletion(-) 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/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}, " From 688180ca3bbb7167e221695f813676df0bc02f72 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 08:32:42 +0000 Subject: [PATCH 08/13] Allow the critic-family mechanisms under sao-streaming-full sao-streaming-full requires the SAO critic, and its critic role has its own syncer (P0-D8). The multi-island critic allowance was limited to strict-avg, so no SAO streaming run with more than one island could start. Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_critic_fp32_masters.py | 18 ++++++++++++++++++ yeto/rl/engine/algorithm.py | 13 ++++++++----- 2 files changed, 26 insertions(+), 5 deletions(-) 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/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( From 95d91d054118d58ac21241ba98e1411f3a91d869 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 08:41:56 +0000 Subject: [PATCH 09/13] Reload the SAO contract bundle on the head controller The head replays the launch args as JSON, which drops the loaded sao-streaming-full contract bundle, so rendering the role syncers failed with AttributeError rl_sao_manifest. cmd_head now reloads the bundle from --rl-sao-contracts. The SAO launcher tests no longer stub the unverified mechanism allowance, which sao-streaming-full now passes. Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_launcher_sao_streaming.py | 42 +++++++++++++++++++------ yeto/cli.py | 6 ++++ 2 files changed, 38 insertions(+), 10 deletions(-) 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/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) From 172e1246f04e92de51fdb5e2922b65ae4a430dd0 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 09:30:12 +0000 Subject: [PATCH 10/13] Wait for the initial cut off the event loop An island initializes its actor and critic streams concurrently, and each role's initial cut needs every island's INIT for that role. The wait polled with time.sleep on the event loop, so the role that got there first kept the other role from connecting. Two SAO islands blocked on different roles deadlocked (s23-g3-sao-main-20261014a: island 0 only reached the actor syncer, island 1 only the critic syncer). Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_full_parameter_sync.py | 65 +++++++++++++++++++ .../miles/full_parameter/role_stream.py | 9 ++- 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/tests/test_rl_full_parameter_sync.py b/tests/test_rl_full_parameter_sync.py index c91dba3e..265532a8 100644 --- a/tests/test_rl_full_parameter_sync.py +++ b/tests/test_rl_full_parameter_sync.py @@ -729,6 +729,71 @@ async def fake_initialize(group, **kwargs): 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. diff --git a/yeto/rl/adapters/miles/full_parameter/role_stream.py b/yeto/rl/adapters/miles/full_parameter/role_stream.py index c54abcb3..e1c0f6b2 100644 --- a/yeto/rl/adapters/miles/full_parameter/role_stream.py +++ b/yeto/rl/adapters/miles/full_parameter/role_stream.py @@ -361,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() From ddbefdb97689080d3e8b95fdc538a64c9dc6b381 Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 04:21:16 +0000 Subject: [PATCH 11/13] Tape each accepted group's generating policy under a policy-age limit Under --rl-max-policy-age > 0 the driver emits rl_group_policies (group id, policy label yeto::, version segments) so a round's trajectories can be matched against the publications. A cut restore also reports the ids of the resumed buffered groups. Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_agentic_suspend.py | 2 ++ tests/test_rl_policy_age.py | 15 +++++++++++++++ yeto/rl/adapters/miles/in_flight.py | 3 +++ yeto/rl/engine/driver.py | 8 ++++++++ 4 files changed, 28 insertions(+) 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_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/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/engine/driver.py b/yeto/rl/engine/driver.py index bc2ea7c0..f5a8241e 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 From fd52828be00ddf1fb5c6623844dd37953750b23e Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 05:15:18 +0000 Subject: [PATCH 12/13] Resume a round cut from the data cursor the cut recorded Under fully_async the round hook reports no data cursor and the background worker keeps drawing after the round; the cut puts the untrained groups back into the data buffer and records the data source position. A resume now seeks to that position (rl_data_cursor_restored source=round_cut) instead of refusing for a missing ledger cursor (GPU run s23-g3-ps-b-20261014a). Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_restart_data_cursor.py | 56 ++++++++++++++++++++++++++++ yeto/rl/adapters/miles/round_cut.py | 18 +++++++++ yeto/rl/engine/bridges.py | 6 +++ yeto/rl/engine/driver.py | 16 +++++++- 4 files changed, 95 insertions(+), 1 deletion(-) 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/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/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 f5a8241e..f44e29cc 100644 --- a/yeto/rl/engine/driver.py +++ b/yeto/rl/engine/driver.py @@ -1700,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 From 25fb3ceef4422fe9d50c18f3e66ffc12ff4368ae Mon Sep 17 00:00:00 2001 From: michael <78113554+chowhsu@users.noreply.github.com> Date: Wed, 14 Oct 2026 11:06:29 +0000 Subject: [PATCH 13/13] Drop the removed SAO allowance fixture from the real-syncer test import 120e8548 removed the _declared_sao_mechanisms fixture, but test_rl_sao_streaming_real_syncer still imported it, so collection of the whole suite failed. sao-streaming-full passes the allowance without the stub. Co-Authored-By: Claude Opus 5.5 --- tests/test_rl_sao_streaming_real_syncer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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")