feat: add deterministic fully-async staleness control

examples/fully_async/README.md

@@ -60,6 +60,25 @@ work unchanged under fully-async:
 See `examples/coding_agent_rl/` for a non-trivial example that plugs in a
 multi-turn agent (Claude Code in a Docker-Proxy sandbox) this way.
 
+To bound off-policy staleness, set `--max-policy-version-lag N`. A completed
+group is accepted when `current_policy_version - group_policy_version <= N`;
+older groups are rejected and their admission-time prompt snapshots are
+requeued for regeneration. The default is unset (no stale rejection), while
+`--max-policy-version-lag 0` accepts only the current policy version.
+
+## Metrics
+
+Fully-async rollouts report fixed keys only—policy versions and sample IDs
+never become labels—so metric cardinality stays bounded:
+
+* `fully_async/version_lag/count_{0,1,2_to_3,4_plus,unknown}`
+* `fully_async/count/{stale_rejected,stale_requeued,aborted_requeued}`
+* `fully_async/queue/completed_groups`
+* `fully_async/policy/current_version`
+
+Counters cover the interval since the previous completed rollout; queue size
+and current version are gauges captured when that interval is emitted.
+
 ## Worker Internals (Very Short)
 
 * First call: create a process-wide `AsyncRolloutWorker` (thread + asyncio
@@ -70,6 +89,8 @@ multi-turn agent (Claude Code in a Docker-Proxy sandbox) this way.
 * Completed groups land on an output queue; each `generate_rollout` call
   drains until it has `rollout_batch_size` groups and returns them sorted
   by `sample.index`.
+* When a policy-version lag budget is configured, stale completed groups are
+  replaced by fresh attempts from their admission-time prompt snapshots.
 * Groups containing an `ABORTED` sample are pushed back into
   `data_buffer.add_samples` instead of being shipped to training.
 * Worker is stopped automatically at process exit via `atexit`.

slime/ray/rollout.py

@@ -1,4 +1,5 @@
 import dataclasses
+import importlib
 import itertools
 import logging
 import multiprocessing
@@ -504,6 +505,8 @@ def __init__(self, args, pg):
             runtime_env={"env_vars": add_default_ray_env_vars()},
         ).remote()
         self.rollout_id = -1
+        self.policy_version = 0
+        self.args.policy_version = self.policy_version
 
         self._health_monitors = []
         if not self.args.debug_train_only and self.args.use_fault_tolerance:
@@ -587,6 +590,37 @@ def get_num_rollout_per_epoch(self):
         assert self.args.rollout_global_dataset
         return len(self.data_source) // self.args.rollout_batch_size
 
+    def _call_generate_rollout_hook(self, hook_name: str, **kwargs):
+        """Call an optional lifecycle hook defined beside the rollout function."""
+
+        module = importlib.import_module(self.generate_rollout.__module__)
+        hook = getattr(module, hook_name, None)
+        if hook is None:
+            return None
+        return hook(**kwargs)
+
+    def before_weight_update(self) -> int:
+        """Notify the rollout implementation before actor weights are updated."""
+
+        self._call_generate_rollout_hook(
+            "before_weight_update",
+            policy_version=self.policy_version,
+        )
+        return self.policy_version
+
+    def after_weight_update(self, succeeded: bool) -> int:
+        """Publish a new policy version only after all weight updates succeed."""
+
+        if succeeded:
+            self.policy_version += 1
+            self.args.policy_version = self.policy_version
+        self._call_generate_rollout_hook(
+            "after_weight_update",
+            policy_version=self.policy_version,
+            succeeded=succeeded,
+        )
+        return self.policy_version
+
     def generate(self, rollout_id):
         start_time = time.time()
         self.rollout_id = rollout_id