diff --git a/examples/ar/qwen3_drpo_4b_base_dapo_sglang_async.yaml b/examples/ar/qwen3_drpo_4b_base_dapo_sglang_async.yaml
index 0679bcf5..ae6fa498 100644
--- a/examples/ar/qwen3_drpo_4b_base_dapo_sglang_async.yaml
+++ b/examples/ar/qwen3_drpo_4b_base_dapo_sglang_async.yaml
@@ -23,17 +23,12 @@ num_rollouts: 800
 
 # Train slab share (rollout gets the rest). 0.5 -> equal split.
 train_fraction: 0.5
-# Async knobs (one unified loop): max_inflight = concurrent generations (overlap
-# depth; 1 ≈ one-step pipeline). buffer_max_staleness = weight-syncs a buffered
-# group may cross: 0 = on-policy (ratio≈1), >0 = off-policy continuous buffer.
+# Async control. max_staleness counts whole rollout batches the behavior policy
+# may trail the train policy by; admission depth is max_staleness + 1.
+# max_staleness=0 keeps the behavior/train versions aligned at batch admission;
+# max_inflight remains a resource cap.
 max_inflight: 1
-buffer_max_staleness: 0
-
-# =1 is the on-policy reward-parity baseline vs the colocate ARTrainer (a sync
-# sits between every gen/train pair -> NO overlap). Set >1 to overlap
-# generate(N+1) with train(N): data is then off-policy by <= interval weight
-# versions, absorbed by DRPO's rollout-anchored ratio (old_logp_source: rollout).
-weight_sync_interval: 1
+max_staleness: 0
 
 eval_interval: 10
 adv_normalization_scope: group
diff --git a/examples/ar/qwen3_grpo_4b_base_dapo_sglang_async.yaml b/examples/ar/qwen3_grpo_4b_base_dapo_sglang_async.yaml
index 6ee77cbd..e29450b9 100644
--- a/examples/ar/qwen3_grpo_4b_base_dapo_sglang_async.yaml
+++ b/examples/ar/qwen3_grpo_4b_base_dapo_sglang_async.yaml
@@ -26,19 +26,15 @@ num_rollouts: 800
 # train-bound, lower for generation-bound workloads (mind DIVISIBILITY above).
 train_fraction: 0.5
 
-# Async knobs (one unified loop): max_inflight = concurrent generations (overlap
-# depth; 1 ≈ one-step pipeline). buffer_max_staleness = weight-syncs a buffered
-# group may cross: 0 = on-policy (ratio≈1), >0 = off-policy continuous buffer.
+# Async control. max_staleness counts whole rollout batches the behavior policy
+# may trail the train policy by; admission depth is max_staleness + 1.
+# max_staleness=0 means every batch starts training against the same train
+# version that generated it (rollout/replay numerical differences may still make
+# ratio only approximately 1) — note that with num_updates_per_batch=4 below,
+# updates 2-4 of a batch are still 1-3 updates off-policy by design, which the
+# frozen pi_old anchor and clip_range cover. max_inflight remains a resource cap.
 max_inflight: 1
-buffer_max_staleness: 0
-
-# Push the full dense weights into SGLang every N rollouts. =1 is the on-policy
-# reward-parity baseline vs the colocate ARTrainer (a sync sits between every
-# gen/train pair, so there is NO overlap). Set >1 to actually overlap
-# generate(N+1) with train(N): the data is then off-policy by <= interval
-# weight versions, absorbed by the rollout-anchored PPO ratio (old_logp =
-# segment.log_probs), exactly the slime train_async trade-off.
-weight_sync_interval: 1
+max_staleness: 0
 
 # AIME eval every 10 rollouts + a pre-train baseline (rollout -1).
 eval_interval: 10
diff --git a/examples/diffusion/bagel/bagel_vllmomni_async.yaml b/examples/diffusion/bagel/bagel_vllmomni_async.yaml
index e3b766d2..5b38b201 100644
--- a/examples/diffusion/bagel/bagel_vllmomni_async.yaml
+++ b/examples/diffusion/bagel/bagel_vllmomni_async.yaml
@@ -43,29 +43,23 @@ workers_per_device: 1
 layout: separate
 train_fraction: 0.5            # 4 train GPUs + 4 rollout GPUs on an 8-GPU pool
 
-# LoRA weight-sync cadence (loop concern, read by the trainer). >1 is required
-# for async overlap: interval=1 drains every step (every rollout is a cold sync
-# boundary, no generation can overlap a train step). interval=4 gives 3 of every
-# 4 rollouts a generation overlapped with training. old_logp_source=rollout keeps
-# the emitted π_old as a valid importance-sampling anchor; the ratio may move as
-# the resident rollout policy and current train policy diverge.
-weight_sync_interval: 4
-
-# ---- async knobs (AsyncDiffusionTrainer) ----
+# ---- async control ----
 # max_inflight: concurrent generations. MUST be 1: the trajectory-segment
 # cross-slab transfer (NCCL send) runs on the rollout worker; a second in-flight
 # generation co-tenanting that worker blocks the send behind it (~150s/rollout).
 # With max_inflight=1 the trainer polls (reaps) before topping up launches, so
 # each generation transfers in the idle window before the next launch, then
-# that next generation overlaps the train step. Real overlap needs
-# weight_sync_interval>1 (interval=1 drains every step).
-# buffer_max_staleness: how many weight syncs a buffered group may cross.
-#   0 = never crosses a regular rollout-weight sync (~174s/rollout on BAGEL 4+4).
-#   2 = throughput-optimal continuous buffer (~148s/rollout, matches vllm
-#       colocate); sync-boundary cold rollouts disappear. The emitted π_old
-#       remains a valid anchor, while the measured ratio may reflect policy lag.
+# that next generation overlaps the train step. Real overlap needs admission
+# depth > 1, i.e. max_staleness >= 1.
+#
+# max_staleness counts whole rollout batches, so admission depth is
+# max_staleness + 1 = 12 regardless of num_updates_per_batch. Note this is also
+# the weight-sync period: 11 batches (22 optimizer updates) elapse between
+# publications, which is a long way past AReaL's recommended eta <= 8 — and
+# FlowGRPO's rollout-anchored ratio is their "naive PPO" arm, with no decoupled
+# objective to absorb the drift. Lower this first if the ratio starts to swing.
 max_inflight: 1
-buffer_max_staleness: 2
+max_staleness: 11
 
 logging:
   report_to_wandb: false        # flip to true to enable wandb (rank-0/driver only)
@@ -220,7 +214,7 @@ sampling:
 # LoRA weight sync → vLLM-Omni rollout. Separate slabs ⇒ RemoteLoraWeightSync:
 # rank 0 ships the freshly-trained LoRA adapter to each cross-slab rollout Worker
 # by Ray RPC (no NCCL rendezvous, no name_remap — pushes the adapter directly).
-# The train loop drives cadence via weight_sync_interval.
+# The train loop publishes only when the optimizer-update lag budget binds.
 sync:
   _target_: unirl.distributed.weight_sync.lora.RemoteLoraWeightSync
   verify: true                         # checksum read-back asserts the synced LoRA landed; catches a wrong prefix
diff --git a/unirl/rollout/README.md b/unirl/rollout/README.md
index d06a0478..8490586c 100644
--- a/unirl/rollout/README.md
+++ b/unirl/rollout/README.md
@@ -64,14 +64,14 @@ wrong objective.
   plus a `sync:` block; *colocate* — a dedicated engine sharing GPUs with train,
   plus offload/onload and `sync:`.
 - **Driver-side async engines** (`engine/asynchronous.py`, the driver-side half next
-  to `engine/synchronous.py`'s worker-side sync contracts). Both engines expose the
-  same consumer verbs the async trainers program against: `poll` / `drain_freshest` /
-  `pop_evicted` / `quiesce` + engine-owned `weight_version`. `AsyncBatchRolloutEngine`
-  (batch granularity; non-blocking `Handle.launch_nowait` generations, stamps
-  versions at launch, used by `AsyncARTrainer`/`AsyncDiffusionTrainer`) and
-  `AsyncAgenticRolloutEngine` (trajectory granularity over the agentic rank-0
+  to `engine/synchronous.py`'s worker-side sync contracts). `AsyncBatchRolloutEngine`
+  (batch granularity; non-blocking `Handle.launch_nowait` generations, stamps the
+  synced train version at launch, and exposes completion-order FIFO train batches
+  to `AsyncARTrainer`/`AsyncDiffusionTrainer`) and `AsyncAgenticRolloutEngine`
+  (trajectory granularity over the agentic rank-0
   coordinator; normalizes the `[0]` unwraps, assembles n-sibling GRPO groups,
-  stamps versions at completion, used by the partial/async agentic trainers).
+  stamps sync-generation versions at completion, and retains the
+  `drain_freshest`/`pop_evicted` surface used by partial/async agentic trainers).
 
 **Extending it:** a new single-turn engine adds `engine/<name>/config.py` (a
 `BaseEngineConfig` whose `make_engine(**deps)` lazily imports and builds it) and
@@ -94,7 +94,9 @@ implements its weight-receive method and a matching `sync:` handler in
   engine also can't live on a `layout: separate` slab — `_build_rollout` raises.
 - **Quiesce before weight sync / eval / checkpoint on the batch async path** —
   `AsyncBatchRolloutEngine.quiesce()` drains every in-flight generation; a
-  weight + KV update corrupts one mid-flight. The agentic quiesce is a
+  weight + KV update corrupts one mid-flight. A batch weight publication also
+  requires the completed FIFO to be empty; hard-boundary admission guarantees
+  this rather than silently discarding data. The agentic quiesce is a
   turn-boundary `abort` + final poll, folded into
   `AsyncAgenticRolloutEngine.quiesce()`; its `sync_weights()` rejects a live
   drive, then pairs the weight push with the version bump and logs the sync.
diff --git a/unirl/rollout/engine/asynchronous.py b/unirl/rollout/engine/asynchronous.py
index 20018a8c..8ce47aff 100644
--- a/unirl/rollout/engine/asynchronous.py
+++ b/unirl/rollout/engine/asynchronous.py
@@ -5,18 +5,18 @@
 the driver side — everything is single-threaded, lock-free, and ray-free
 (non-blocking dispatch is ``Handle.launch_nowait``).
 
-Mechanisms (policy-free — launch ceilings, reap/launch ordering, and step
+Mechanisms (policy-free — launch admission, reap/launch ordering, and step
 loops live in the trainers):
 
 - :class:`VersionedBuffer` — payload-agnostic freshness/staleness buffer.
 - :class:`InflightPool` — non-blocking pool of distributed ``generate`` calls.
 
-Engines share one consumer surface (``poll`` / ``drain_freshest`` /
-``pop_evicted`` / ``quiesce`` + engine-owned ``weight_version``):
+Engines share launch/poll/quiesce mechanisms but own different queue semantics:
 
 - :class:`AsyncBatchRolloutEngine` — batch granularity over a single-turn
   engine slab; one ``submit`` is one non-blocking distributed ``generate``.
-  ``(weight_version, gen_id)`` are stamped at LAUNCH.
+  ``(behavior_version, gen_id)`` are stamped at LAUNCH and one complete
+  generation is consumed atomically in completion-order FIFO.
 - :class:`AsyncAgenticRolloutEngine` — trajectory granularity over the
   ``AgenticRolloutEngine`` rank-0 coordinator; ``submit`` fires a task-pool
   drive and completions stream in via ``poll``. ``(weight_version, gen_id)``
@@ -32,11 +32,13 @@
 from __future__ import annotations
 
 import logging
+from collections import deque
 from dataclasses import dataclass
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
+    Deque,
     Dict,
     Generic,
     Iterable,
@@ -106,22 +108,66 @@ def pop_evicted(self) -> List[T]:
         return evicted
 
 
-Complete = Callable[[int, int, Any], None]
+@dataclass(frozen=True)
+class RolloutBatch:
+    """One atomic train batch produced by one batch-generation launch."""
+
+    groups: List["Sample"]
+    behavior_version: int
+    gen_id: int
+
+
+class RolloutBatchQueue:
+    """Completion-order FIFO for single-turn batch generations."""
+
+    def __init__(self) -> None:
+        self._items: Deque[RolloutBatch] = deque()
+
+    def put(self, batch: RolloutBatch) -> None:
+        self._items.append(batch)
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+    def pop_next(
+        self,
+        *,
+        train_version: int,
+        staleness_budget: int,
+    ) -> Optional[RolloutBatch]:
+        if not self._items:
+            return None
+        item = self._items[0]
+        staleness = train_version - item.behavior_version
+        if staleness < 0:
+            raise RuntimeError(
+                f"generation {item.gen_id} has future behavior version "
+                f"{item.behavior_version} > train version {train_version}"
+            )
+        if staleness > staleness_budget:
+            raise RuntimeError(
+                f"generation {item.gen_id} exceeded its staleness budget: "
+                f"staleness={staleness} > budget={staleness_budget} optimizer updates"
+            )
+        return self._items.popleft()
+
+
+CompleteGeneration = Callable[[int, int, Any], None]
 
 
 @dataclass(frozen=True)
 class _InflightJob:
     gen_id: int
-    weight_version: int
+    behavior_version: int
     pending: Any
 
 
 class InflightPool:
     """Non-blocking pool of distributed ``generate`` launches on a rollout Handle.
 
-    Mechanism only: launch ceilings, reap/launch ordering, and step loops are
+    Mechanism only: launch admission, reap/launch ordering, and step loops are
     caller policy. Jobs are launched via ``Handle.launch_nowait`` and completed
-    through ``complete(gen_id, weight_version, payload)`` — all of ``complete``'s
+    through ``complete(gen_id, behavior_version, payload)`` — all of ``complete``'s
     fallible work must happen before it mutates caller state, because a job
     whose completion raises stays in flight for retry.
     """
@@ -138,14 +184,14 @@ def next_gen_id(self) -> int:
     def __len__(self) -> int:
         return len(self._jobs)
 
-    def launch(self, sample: Any, *, weight_version: int) -> int:
+    def launch(self, sample: Any, *, behavior_version: int) -> int:
         gen_id = self._next_gen_id
         pending = self._rollout.launch_nowait("generate", sample)
-        self._jobs.append(_InflightJob(gen_id, int(weight_version), pending))
+        self._jobs.append(_InflightJob(gen_id, behavior_version, pending))
         self._next_gen_id += 1
         return gen_id
 
-    def reap_ready(self, complete: Complete) -> int:
+    def reap_ready(self, complete: CompleteGeneration) -> int:
         """Complete every ready job; leave unresolved and failed jobs in flight.
 
         A job whose ``result()``/``complete`` raises stays in flight for retry;
@@ -161,7 +207,7 @@ def reap_ready(self, complete: Complete) -> int:
                 still.append(job)
                 continue
             try:
-                complete(job.gen_id, job.weight_version, job.pending.result())
+                complete(job.gen_id, job.behavior_version, job.pending.result())
                 completed += 1
             except Exception as exc:
                 still.append(job)
@@ -174,7 +220,7 @@ def reap_ready(self, complete: Complete) -> int:
             raise first_error
         return completed
 
-    def drain_all(self, complete: Complete) -> int:
+    def drain_all(self, complete: CompleteGeneration) -> int:
         """Quiesce: complete every job, blocking as needed. Same error contract as
         :meth:`reap_ready`."""
         jobs, self._jobs = self._jobs, []
@@ -182,7 +228,7 @@ def drain_all(self, complete: Complete) -> int:
         completed = 0
         for job in jobs:
             try:
-                complete(job.gen_id, job.weight_version, job.pending.result())
+                complete(job.gen_id, job.behavior_version, job.pending.result())
                 completed += 1
             except Exception as exc:
                 self._jobs.append(job)
@@ -201,12 +247,13 @@ def wait_oldest(self) -> None:
 
 
 class AsyncBatchRolloutEngine:
-    """Batch-granular async engine over a ``SyncRolloutEngine`` slab Handle; buffers ``Sample`` groups.
+    """Batch-granular async engine over a ``SyncRolloutEngine`` slab Handle.
 
-    ``complete(gen_id, completed) -> groups`` runs at reap time — scoring must
-    precede training, and on transfer-sensitive backends the next launch. All
-    of ``complete``'s fallible work happens before any buffer mutation, so a
-    failed job stays in flight for retry without double-inserting groups.
+    One generation must produce exactly ``groups_per_batch`` groups and enters a
+    completion-order FIFO atomically. ``behavior_version`` is captured at launch
+    from the exact train snapshot currently synced to the rollout engine, so
+    ``staleness_budget`` is compared in committed optimizer updates — the caller
+    resolves it from a batch count before passing it down.
 
     ``quiesce()`` (drain everything) is MANDATORY before a weight sync, eval, or
     checkpoint: a weight + KV update corrupts an in-flight generation.
@@ -216,21 +263,16 @@ def __init__(
         self,
         rollout: Any,
         *,
-        complete: Callable[[int, "Sample"], List["Sample"]],
+        process_completion: Callable[[int, "Sample"], List["Sample"]],
+        groups_per_batch: int,
         start_gen_id: int = 0,
     ) -> None:
-        self._complete = complete
+        if groups_per_batch < 1:
+            raise ValueError(f"groups_per_batch must be >= 1, got {groups_per_batch}")
+        self._process_completion = process_completion
+        self._groups_per_batch = groups_per_batch
         self._pool = InflightPool(rollout, start_gen_id=start_gen_id)
-        self._buffer: VersionedBuffer["Sample"] = VersionedBuffer()
-        self._weight_version = 0
-
-    @property
-    def weight_version(self) -> int:
-        return self._weight_version
-
-    def bump_weight_version(self) -> int:
-        self._weight_version += 1
-        return self._weight_version
+        self._ready = RolloutBatchQueue()
 
     @property
     def next_gen_id(self) -> int:
@@ -238,34 +280,52 @@ def next_gen_id(self) -> int:
         return self._pool.next_gen_id
 
     @property
-    def inflight(self) -> int:
+    def inflight_count(self) -> int:
         return len(self._pool)
 
-    def submit(self, sample: "Sample") -> int:
-        """Launch one non-blocking distributed ``generate``; stamps the CURRENT version."""
-        return self._pool.launch(sample, weight_version=self._weight_version)
+    @property
+    def ready_count(self) -> int:
+        return len(self._ready)
+
+    def submit(self, sample: "Sample", *, behavior_version: int) -> int:
+        """Launch one generation under the supplied rollout policy version."""
+
+        return self._pool.launch(sample, behavior_version=behavior_version)
 
     def poll(self) -> int:
         return self._pool.reap_ready(self._on_complete)
 
-    def drain_freshest(self, n: int, *, max_staleness: int) -> Optional[List["Sample"]]:
-        return self._buffer.drain_freshest(n, current_version=self._weight_version, max_staleness=max_staleness)
-
-    def pop_evicted(self) -> List["Sample"]:
-        return self._buffer.pop_evicted()
-
-    def quiesce(self) -> List["Sample"]:
+    def pop_next_batch(
+        self,
+        *,
+        train_version: int,
+        staleness_budget: int,
+    ) -> Optional[RolloutBatch]:
+        return self._ready.pop_next(
+            train_version=train_version,
+            staleness_budget=staleness_budget,
+        )
+
+    def quiesce(self) -> None:
         self._pool.drain_all(self._on_complete)
-        return []
 
     def wait_oldest(self) -> None:
         """Block until the oldest in-flight generation resolves (reap via ``poll``)."""
         self._pool.wait_oldest()
 
-    def _on_complete(self, gen_id: int, weight_version: int, completed: "Sample") -> None:
-        groups = self._complete(gen_id, completed)  # fallible (scoring) before any buffer put
-        for group in groups:
-            self._buffer.put(group, weight_version=weight_version, gen_id=gen_id)
+    def _on_complete(self, gen_id: int, behavior_version: int, completed: "Sample") -> None:
+        groups = self._process_completion(gen_id, completed)  # fallible before any queue mutation
+        if len(groups) != self._groups_per_batch:
+            raise RuntimeError(
+                f"generation {gen_id} produced {len(groups)} groups; expected groups_per_batch={self._groups_per_batch}"
+            )
+        self._ready.put(
+            RolloutBatch(
+                groups=groups,
+                behavior_version=behavior_version,
+                gen_id=gen_id,
+            )
+        )
 
 
 def root_of(traj: "Sample") -> str:
@@ -418,27 +478,9 @@ def _ingest(self, completed: List["Sample"]) -> int:
         return len(completed)
 
 
-def launch_ceiling(rollout_id: int, *, sync_interval: int, max_staleness: int, num_rollouts: int) -> int:
-    """The batch trainers' on-policy launch clamp — trainer POLICY, defined once.
-
-    A generation launched now is consumed later, so how far ahead the gen_id
-    allocator may run is bounded to ``max_staleness`` weight-sync windows:
-    ``max_staleness=0`` ⇒ never launch into a future sync-window ⇒ no
-    generation crosses a sync ⇒ ``ratio≈1`` (on-policy).
-
-    OWNERSHIP: this is trainer-side POLICY, not engine surface — its vocabulary
-    (``rollout_id`` / ``sync_interval`` / ``num_rollouts``) is the trainers',
-    the engine classes never call it, and it must never become an engine
-    method. It is hosted in this module only because it is the two batch
-    trainers' one shared torch-free home; the step loops that use it stay in
-    the trainers as visible statement order.
-    """
-    return min(num_rollouts, ((rollout_id // sync_interval) + 1 + max_staleness) * sync_interval)
-
-
 __all__ = [
     "AsyncAgenticRolloutEngine",
     "AsyncBatchRolloutEngine",
-    "launch_ceiling",
+    "RolloutBatch",
     "root_of",
 ]
diff --git a/unirl/rollout/engine/synchronous.py b/unirl/rollout/engine/synchronous.py
index 87966a9a..43d6ea1e 100644
--- a/unirl/rollout/engine/synchronous.py
+++ b/unirl/rollout/engine/synchronous.py
@@ -157,6 +157,19 @@ class SyncRolloutEngine(BaseRolloutEngine, ABC):
     def generate(self, sample: Sample) -> Sample:
         """Synchronously fill and return one request ``Sample``."""
 
+    @distributed(dispatch_mode=Dispatch.BROADCAST)
+    def set_policy_version(self, train_version: int) -> None:
+        """Assign the optimizer-update version of the currently loaded weights.
+
+        Weight transports may receive one model through several buckets, so
+        receiver-local API-call counters are not valid policy provenance. The
+        trainer calls this only after a full weight publication succeeds.
+        """
+
+        if train_version < 0:
+            raise ValueError(f"train_version must be >= 0, got {train_version}")
+        self._weight_version = train_version
+
     def _stamp_weight_version(self, sample: Sample) -> Sample:
         """Stamp ``self._weight_version`` onto the frontier (last) gen Part."""
         v = getattr(self, "_weight_version", None)
diff --git a/unirl/train/backend/base_backend.py b/unirl/train/backend/base_backend.py
index 7fe51d10..e1476fb1 100644
--- a/unirl/train/backend/base_backend.py
+++ b/unirl/train/backend/base_backend.py
@@ -370,6 +370,12 @@ def on_rollout_end(self) -> None:
         if self.ema is not None:
             self.ema.on_rollout_end(self._optimizer_step_count)
 
+    @distributed(dispatch_mode=Dispatch.BROADCAST)
+    def get_optimizer_step_count(self) -> int:
+        """Return the authoritative number of committed optimizer updates."""
+
+        return self._optimizer_step_count
+
     @property
     def rollout_adapter_name(self) -> str:
         """Adapter the rollout must sample under (single source of truth).
diff --git a/unirl/train/stack/base.py b/unirl/train/stack/base.py
index 1af3b0c4..065826e9 100644
--- a/unirl/train/stack/base.py
+++ b/unirl/train/stack/base.py
@@ -41,6 +41,7 @@
 from __future__ import annotations
 
 import logging
+import math
 from contextlib import nullcontext
 from dataclasses import dataclass, replace
 from typing import Dict, List, Mapping, Optional, Tuple
@@ -61,7 +62,7 @@
 
 @dataclass(frozen=True)
 class TrainStepResult:
-    """Result of one full optimizer step on this stage."""
+    """Result of one ``train_track`` call, possibly spanning multiple optimizer updates."""
 
     loss: float
     grad_norm: float
@@ -69,6 +70,9 @@ class TrainStepResult:
     has_backward: bool
     micros: List[AlgorithmStepResult]
     metrics: Mapping[str, object]
+    # Number of optimizer steps that actually committed. This is zero when an
+    # update had no backward or the backend rejected a non-finite grad norm.
+    optimizer_updates: int
     per_update: Tuple[Mapping[str, object], ...] = ()
 
 
@@ -92,6 +96,7 @@ def _aggregate_update_results(results: List["TrainStepResult"]) -> "TrainStepRes
         has_backward=any(r.has_backward for r in results),
         micros=micros,
         metrics=metrics,
+        optimizer_updates=sum(r.optimizer_updates for r in results),
     )
 
 
@@ -302,6 +307,7 @@ def _run_update(
             has_backward=has_backward,
             micros=micro_results,
             metrics=aggregated_metrics,
+            optimizer_updates=1 if has_backward and math.isfinite(grad_norm) else 0,
         )
 
     def on_rollout_end(self) -> None:
@@ -511,7 +517,13 @@ def _run_updates(
             return results[0]
         aggregated = _aggregate_update_results(results)
         per_update = tuple(
-            {**dict(r.metrics), "loss": float(r.loss), "grad_norm": float(r.grad_norm), "lr": float(r.lr)}
+            {
+                **r.metrics,
+                "loss": r.loss,
+                "grad_norm": r.grad_norm,
+                "lr": r.lr,
+                "optimizer_updates": r.optimizer_updates,
+            }
             for r in results
         )
         return replace(aggregated, per_update=per_update)
diff --git a/unirl/train/unified_model_stack.py b/unirl/train/unified_model_stack.py
index a43ad5ee..0592eb77 100644
--- a/unirl/train/unified_model_stack.py
+++ b/unirl/train/unified_model_stack.py
@@ -34,6 +34,7 @@
 from __future__ import annotations
 
 import logging
+import math
 from contextlib import nullcontext
 from dataclasses import replace
 from typing import Dict, List, Mapping, Tuple
@@ -215,6 +216,7 @@ def _backward_part(
             has_backward=has_backward,
             micros=micros,
             metrics=aggregated,
+            optimizer_updates=0,
         )
         return partial, has_backward
 
@@ -250,9 +252,16 @@ def _train_one_step(
             logger.warning("UnifiedModelTrainStack._train_one_step: no algorithm reported backward; skipping step.")
 
         lr = self._current_lr()
+        optimizer_updates = 1 if any_backward and math.isfinite(grad_norm) else 0
         for name, r in list(results.items()):
             results[name] = TrainStepResult(
-                loss=r.loss, grad_norm=grad_norm, lr=lr, has_backward=r.has_backward, micros=r.micros, metrics=r.metrics
+                loss=r.loss,
+                grad_norm=grad_norm,
+                lr=lr,
+                has_backward=r.has_backward,
+                micros=r.micros,
+                metrics=r.metrics,
+                optimizer_updates=optimizer_updates,
             )
         return results
 
diff --git a/unirl/train_async_ar.py b/unirl/train_async_ar.py
index 37c359ab..32526a08 100755
--- a/unirl/train_async_ar.py
+++ b/unirl/train_async_ar.py
@@ -16,9 +16,9 @@
     Constraint: ``train_fraction * num_devices`` and ``(1-train_fraction) * num_devices``
     must both be integers, AND ``batch_size * samples_per_prompt`` must be divisible
     by each slab size (DP_SCATTER divisibility).
-  * ``max_inflight`` — concurrent generations (overlap depth). ``1`` ≈ one-step pipeline.
-  * ``buffer_max_staleness`` — weight-syncs a buffered group may cross. ``0``/unset =
-    on-policy (``ratio≈1``); ``>0`` = off-policy continuous buffer.
+  * ``max_inflight`` — concurrent generations (resource/overlap limit).
+  * ``max_staleness`` — how many whole rollout batches the behavior policy may
+    trail the train policy by at batch admission.
 """
 
 from __future__ import annotations
@@ -56,11 +56,10 @@ def main(cfg: DictConfig) -> None:
         eval_temperature=float(cfg.get("eval_temperature", 1.0)),
         train_fraction=float(cfg.get("train_fraction", 0.5)),
         max_inflight=int(cfg.get("max_inflight", 1)),
-        buffer_max_staleness=cfg.get("buffer_max_staleness"),
+        max_staleness=cfg.get("max_staleness", 0),
     )
     trainer.train(
         num_rollouts=int(cfg.get("num_rollouts", 100)),
-        weight_sync_interval=int(cfg.get("weight_sync_interval", 1)),
         save_interval=int(cfg.get("save_interval", 0)),
         save_dir=cfg.get("save_dir"),
         load_dir=cfg.get("load_dir"),
diff --git a/unirl/train_async_diffusion.py b/unirl/train_async_diffusion.py
index 6b003f0c..a837f6d2 100755
--- a/unirl/train_async_diffusion.py
+++ b/unirl/train_async_diffusion.py
@@ -15,8 +15,8 @@
 
 Extra config knobs vs the synchronous separate recipe:
   * ``max_inflight`` — must be ``1``; other values fail during trainer initialization.
-  * ``buffer_max_staleness`` — regular rollout-weight syncs a buffered group may
-    cross. ``0``/unset never crosses a sync; ``>0`` enables bounded policy lag.
+  * ``max_staleness`` — how many whole rollout batches the behavior policy may
+    trail the train policy by at batch admission.
 ``layout`` is forced to ``separate`` (async needs disjoint train/rollout slabs).
 """
 
@@ -57,11 +57,10 @@ def main(cfg: DictConfig) -> None:
         eval_rewards_cfg=cfg.get("eval_rewards"),
         task_config=cfg.get("task_config"),
         max_inflight=int(cfg.get("max_inflight", 1)),
-        buffer_max_staleness=cfg.get("buffer_max_staleness"),
+        max_staleness=cfg.get("max_staleness", 0),
     )
     trainer.train(
         num_rollouts=cfg.get("num_rollouts", 100),
-        weight_sync_interval=cfg.get("weight_sync_interval", 1),
         save_interval=cfg.get("save_interval", 0),
         save_dir=cfg.get("save_dir"),
         load_dir=cfg.get("load_dir"),
diff --git a/unirl/trainer/README.md b/unirl/trainer/README.md
index 842ff191..27d20965 100644
--- a/unirl/trainer/README.md
+++ b/unirl/trainer/README.md
@@ -59,8 +59,8 @@ The current trainer surface is:
 | `DiffusionTrainer` | one diffusion `Part` → one `TrainStack` | Reference diffusion loop; supports trainside or dedicated rollout, optional separate reward GPUs, FSDP offload, and DiffusionNFT's EMA-adapter rollout. |
 | `ARTrainer` | one AR `Part` → one `TrainStack` | Text or multimodal AR rollout with group/global advantage normalization and optional token-balanced DP shards. |
 | `SFTTrainer` | dataset records → one standalone training `Part` | Reuses the RL TrainStack without rollout, reward, or advantages; owns exact epoch/cursor resume and full-set evaluation. |
-| `AsyncARTrainer` | buffered AR `Sample` groups → one `TrainStack` | Separate train/rollout slabs with resident generation, bounded staleness, and quiescence before sync, eval, or checkpoint. |
-| `AsyncDiffusionTrainer` | buffered diffusion `Sample` groups → one `TrainStack` | The same separate-slab async loop for DiT. Requires `max_inflight=1` and reaps each generation before launching the next, so the cross-slab trajectory transfer never queues behind a fresh generation. |
+| `AsyncARTrainer` | FIFO AR generation batch → one `TrainStack` | Separate train/rollout slabs with resident generation, batch-denominated `max_staleness` admission over an optimizer-update clock, and quiescence before weight sync, eval, or checkpoint. |
+| `AsyncDiffusionTrainer` | FIFO diffusion generation batch → one `TrainStack` | The same update-versioned separate-slab loop for DiT. Requires `max_inflight=1` and reaps each generation before launching the next, so the cross-slab trajectory transfer never queues behind a fresh generation. |
 | `PETrainer` | `ar` + `diffusion` Parts → two `TrainStack`s | Composed prompt-rewrite/image rollout; image rewards propagate to AR rewrites. `freeze_llm=true` trains and checkpoints diffusion only. |
 | `UnifiedModelTrainer` | whole `Sample` → one `UnifiedModelTrainStack` | AR and image losses accumulate into shared-backbone optimizer steps while prompt-tree lineage remains intact during DP scatter. |
 | `AgenticTrainer` / `AgenticEnvTrainer` | variable-depth `List[Sample]` → concatenated turn `Part` | Barrier multi-turn tool use. The base variant scores terminal answers; the env variant consumes per-trajectory environment returns. |
@@ -69,9 +69,9 @@ The current trainer surface is:
 
 The async variants program against the driver-side async engines in
 `unirl/rollout/engine/asynchronous.py`: `AsyncBatchRolloutEngine` (AR/diffusion — non-blocking
-batched generations, launch-time version stamps) and `AsyncAgenticRolloutEngine`
+batched generations, launch-time train-version stamps, completion-order FIFO batches) and `AsyncAgenticRolloutEngine`
 (partial/async agentic — trajectory drives, group assembly, completion-time stamps).
-The trainers keep the policy: launch ceilings, reap-vs-launch order, quiesce points,
+The trainers keep the policy: optimizer-update launch slots, reap-vs-launch order, quiesce points,
 and tail carry/drop.
 
 **Extending it:** a new domain is a new `<Domain>Trainer(BaseTrainer)` that builds its
@@ -180,8 +180,10 @@ so EMA decay schedules continue) and resumes the loop from the saved step.
 Synchronous Sample-based trainers continue `training_progress` and
 driver-authored x_T scheduling, fast-forward a deterministically seeded data
 stream, and force the restored weights into a freshly started rollout engine
-when needed. `AsyncARTrainer` also fast-forwards its deterministic input stream
-but rebuilds its rollout buffer. Partial-agentic resume can consume a different
+when needed. Async AR/diffusion resume reads the backend optimizer count as the
+train version, fast-forwards the deterministic input stream to the saved
+rollout step, and syncs those restored weights into the fresh engine.
+Partial-agentic resume can consume a different
 input sequence when an earlier over-sampled drive required refills, and ReFL
 does not currently fast-forward its data source.
 
@@ -243,9 +245,9 @@ an evaluation and checkpoint fall on the same step, evaluation runs first.
 - `DiffusionTrainer`, `PETrainer`, and `UnifiedModelTrainer` report image
   reward; optional `eval_rewards` suites can
   score the same generated samples or their own prompt sets. PE scores only the
-  diffusion/image frontier. `AsyncDiffusionTrainer` quiesces first and then scores
-  the policy already resident in its rollout engine, without a weight sync and
-  without offloading that engine afterwards.
+  diffusion/image frontier. `AsyncDiffusionTrainer` reaches an empty hard
+  boundary, syncs the current train version when needed, and then scores
+  the resident rollout engine without offloading it afterwards.
 - Agentic evaluation is not implemented. Barrier and partial variants raise if
   evaluation is enabled; async variants currently force it off.
 
diff --git a/unirl/trainer/agentic.py b/unirl/trainer/agentic.py
index 9013c646..ee4855c2 100644
--- a/unirl/trainer/agentic.py
+++ b/unirl/trainer/agentic.py
@@ -216,7 +216,7 @@ def _advantage_train_and_log(
         )
         if not train_parts:  # pathological: every sampled trajectory failed to generate
             logger.warning("AgenticTrainer rollout %d produced no trainable turns.", rollout_id)
-            return TrainStepResult(0.0, 0.0, 0.0, False, [], {}), mean_reward
+            return TrainStepResult(0.0, 0.0, 0.0, False, [], {}, optimizer_updates=0), mean_reward
 
         train_part = Part.concat(train_parts)
         train_part = self._pad_to_dp_multiple(train_part)
diff --git a/unirl/trainer/agentic_async.py b/unirl/trainer/agentic_async.py
index eb940da8..e742eea2 100644
--- a/unirl/trainer/agentic_async.py
+++ b/unirl/trainer/agentic_async.py
@@ -297,7 +297,7 @@ def _train_on_groups(
         depths = [len(tr.gen_parts()) for tr in trajs]
         if not train_parts:
             logger.warning("AsyncAgenticTrainer rollout %d produced no trainable turns.", rollout_id)
-            return TrainStepResult(0.0, 0.0, 0.0, False, [], {}), mean_reward
+            return TrainStepResult(0.0, 0.0, 0.0, False, [], {}, optimizer_updates=0), mean_reward
 
         train_part = self._pad_to_dp_multiple(Part.concat(train_parts))
         result = self.stack.train_track(train_part, training_progress=float(training_progress))
diff --git a/unirl/trainer/async_ar.py b/unirl/trainer/async_ar.py
index 85b72d33..c98bb4ca 100644
--- a/unirl/trainer/async_ar.py
+++ b/unirl/trainer/async_ar.py
@@ -9,20 +9,22 @@
 
 ONE single-threaded loop (slime's "one trainer loop; async-depth is a knob"
 principle, implemented with UniRL-native non-blocking Ray dispatch instead of
-slime's thread+asyncio). The async behavior is set by **two numeric knobs**:
+slime's thread+asyncio). Async freshness rides a clock of committed optimizer
+updates, but the budget is stated in whole rollout batches:
 
 * ``max_inflight`` — how many generations run concurrently (overlap/parallelism
   depth). ``1`` ≈ the classic one-step pipeline; higher fans out more.
-* ``buffer_max_staleness`` — how many weight-syncs a buffered group may cross
-  before it is evicted. ``0`` (default) = **on-policy**: the launch clamp never
-  lets a generation cross a weight sync, so ``ratio≈1`` (the colocate-parity
-  regime). ``>0`` = **off-policy continuous buffer**: generations may run ahead
-  across syncs, bounded by eviction; the rollout-anchored DRPO ratio absorbs it.
+* ``max_staleness`` — how many whole rollout batches the behavior policy may
+  trail the train policy by, checked at batch admission and consumption. ``0``
+  aligns policy versions; the rollout-anchored PPO ratio remains the numerical
+  source of truth. Batches are the unit because admission only ever runs at a
+  batch boundary; with ``num_updates_per_batch > 1`` the updates after the first
+  in a batch additionally drift by up to ``num_updates_per_batch - 1``.
 
 Generation runs through :class:`~unirl.rollout.engine.asynchronous.AsyncBatchRolloutEngine`
 (non-blocking Ray futures over the rollout Handle) on the single driver thread —
-no producer thread, no locks; the trainer's ``_next_step`` loop owns the policy
-(launch ceiling, launch-then-reap order). Draining all in-flight generations
+no producer thread, no locks; the trainer's ``_next_rollout_batch`` loop owns the policy
+(optimizer-update launch admission, launch-then-reap order). Draining all in-flight generations
 before each weight sync is **mandatory** (the engine corrupts an in-flight
 generation when weights + KV cache update mid-flight); this is the
 single-threaded ``_drain_all`` quiesce.
@@ -45,9 +47,16 @@
 from unirl.distributed.group.placement import placement, remote
 from unirl.distributed.tensor import hydrate
 from unirl.models.qwen3_5.validation import validate_qwen3_5_training_contract
-from unirl.rollout.engine.asynchronous import AsyncBatchRolloutEngine, launch_ceiling
+from unirl.rollout.engine.asynchronous import AsyncBatchRolloutEngine, RolloutBatch
 from unirl.train.stack import TrainStepResult
 from unirl.trainer.ar import ARTrainer
+from unirl.trainer.async_policy import (
+    AsyncBatchControl,
+    log_admission_notes,
+    next_hard_boundary,
+    sync_period_batches,
+    unwrap_replicated_int,
+)
 from unirl.trainer.base import BaseTrainer, build_sampling_dict
 from unirl.types.sample import Sample
 from unirl.types.sampling import BaseSamplingParams, total_samples_per_prompt
@@ -101,7 +110,7 @@ def __init__(
         eval_temperature: float = 1.0,
         train_fraction: float = 0.5,
         max_inflight: int = 1,
-        buffer_max_staleness: Optional[int] = None,
+        max_staleness: int = 0,
     ) -> None:
         validate_qwen3_5_training_contract(
             pipeline_cfg=pipeline_cfg,
@@ -138,7 +147,10 @@ def __init__(
 
         self._train_fraction = float(train_fraction)
         self._max_inflight = max(1, int(max_inflight))
-        self._buffer_max_staleness = buffer_max_staleness
+        self._control = AsyncBatchControl(
+            max_staleness=max_staleness,
+            num_updates_per_batch=stack_cfg.get("num_updates_per_batch", 1),
+        )
         self._train_devices = int(round(self.num_devices * self._train_fraction))
         if self._train_devices <= 0 or self._train_devices >= self.num_devices:
             raise ValueError(
@@ -177,13 +189,14 @@ def __init__(
         with placement(self.pool, fraction=1.0 - self._train_fraction, shared_workers=True):
             self.rollout = remote(**rollout_parsed)
 
-        if self.weight_sync is not None:
-            self._connect_separate(sync_cfg)
+        if self.weight_sync is None:
+            raise ValueError("AsyncARTrainer requires a cross-slab weight sync; add a `sync:` block.")
+        self._connect_separate(sync_cfg)
 
     def _prepare_rollout(self, *, sync_weights: bool) -> bool:
         """Sync a resident separate-slab engine without colocate handoffs."""
-        if sync_weights and self.weight_sync is not None:
-            self.weight_sync.sync()
+        if sync_weights:
+            self._control.sync_rollout(self._async_engine, self.rollout, self.weight_sync)
         return False
 
     def _finish_rollout(self, *, train_state_offloaded: bool) -> None:
@@ -251,8 +264,9 @@ def _advantage_and_train(
         training_progress: float,
         rollout_id: int,
         t0: Optional[float] = None,
+        extra_metrics: Optional[Dict[str, float]] = None,
     ) -> Tuple[TrainStepResult, float]:
-        """Advantage + optimizer step for a SCORED ``Sample`` (rewards already attached)."""
+        """Advantage + optimizer updates for a scored ``Sample`` (rewards already attached)."""
         if t0 is None:
             t0 = time.perf_counter()
         part = sample.parts[-1]
@@ -270,12 +284,16 @@ def _advantage_and_train(
         if self.balance_shards:
             train_part = part.balance_shards(self._train_devices)
         result = self.stack.train_track(train_part, training_progress=float(training_progress))
+        self._control.record_optimizer_updates(result.optimizer_updates)
+        if extra_metrics is not None:
+            extra_metrics.update(self._control.train_metrics(result.optimizer_updates))
         self.wandb_logger.log_rollout_step(
             rollout_id,
             result,
             sample,
             step_time_s=time.perf_counter() - t0,
             trunc_len=getattr(self.sampling_params.get("ar"), "max_new_tokens", None),
+            extra_metrics=extra_metrics,
         )
         self._reset_transport_buffers()
         return result, mean_reward
@@ -284,65 +302,99 @@ def train(
         self,
         *,
         num_rollouts: int,
-        weight_sync_interval: int = 1,
         save_interval: int = 0,
         save_dir: Optional[str] = None,
         load_dir: Optional[str] = None,
         save_mode: str = "full",
     ) -> None:
-        interval = max(1, weight_sync_interval)
-        stale = self._buffer_max_staleness if self._buffer_max_staleness is not None else 0
-        M = self._max_inflight
-
         start_rollout = self.maybe_load_checkpoint(load_dir, num_rollouts=num_rollouts)
         resumed = bool(load_dir)
+        train_version = unwrap_replicated_int(
+            self.backend.get_optimizer_step_count(),
+            name="backend optimizer step count",
+        )
+        self._control.restore(train_version)
         for _ in range(start_rollout):
             self.data_source.get_samples(self.batch_size)
         self._init_wandb(
             num_rollouts=num_rollouts,
             extra={
                 "adv_normalization_scope": self.adv_normalization_scope,
-                "max_inflight": M,
-                "buffer_max_staleness": stale,
-                "weight_sync_interval": interval,
+                "max_inflight": self._max_inflight,
+                "max_staleness": self._control.max_staleness,
+                "staleness_budget": self._control.staleness_budget,
+                "num_updates_per_batch": self._control.num_updates_per_batch,
+                "sync_period_batches": sync_period_batches(
+                    self._control,
+                    eval_interval=self.eval_interval,
+                    save_interval=save_interval,
+                ),
             },
         )
+        # Reported here rather than in __init__: save_interval only arrives with
+        # the train call, and it clamps the publication period just as the
+        # staleness budget does.
+        log_admission_notes(
+            self._control,
+            max_inflight=self._max_inflight,
+            eval_interval=self.eval_interval,
+            save_interval=save_interval,
+        )
 
         self._async_engine = AsyncBatchRolloutEngine(
             self.rollout,
-            complete=self._score_completed,
+            process_completion=self._score_completed,
+            groups_per_batch=self.batch_size,
             start_gen_id=start_rollout,
         )
 
-        if resumed and self.weight_sync is not None:
-            self.weight_sync.sync()
+        if resumed or self.eval_interval > 0:
+            self._control.sync_rollout(self._async_engine, self.rollout, self.weight_sync, force=True)
         if self.eval_interval > 0:
             self.evaluate(rollout_id=-1)
 
         try:
             for rollout_id in range(start_rollout, num_rollouts):
                 t0 = time.perf_counter()
-                picked = self._next_step(rollout_id, interval, M, stale, num_rollouts)
-                sample = Sample.concat(picked)
+                hard_boundary = next_hard_boundary(
+                    rollout_id,
+                    num_rollouts=num_rollouts,
+                    eval_interval=self.eval_interval,
+                    save_interval=save_interval,
+                )
+                batch = self._next_rollout_batch(
+                    rollout_id,
+                    num_rollouts=num_rollouts,
+                    hard_boundary=hard_boundary,
+                )
+                sample = Sample.concat(batch.groups)
                 training_progress = rollout_id / max(1, num_rollouts - 1)
                 result, mean_reward = self._advantage_and_train(
-                    sample, training_progress=training_progress, rollout_id=rollout_id, t0=t0
+                    sample,
+                    training_progress=training_progress,
+                    rollout_id=rollout_id,
+                    t0=t0,
+                    extra_metrics=self._control.behavior_metrics(batch.behavior_version),
                 )
                 self.wandb_logger.log_progress(rollout_id, num_rollouts, result, mean_reward, logger=logger)
 
                 step = rollout_id + 1
-                if self.eval_interval > 0 and step % self.eval_interval == 0:
-                    self._drain_all()
+                eval_due = self.eval_interval > 0 and step % self.eval_interval == 0
+                save_due = save_interval > 0 and (step % save_interval == 0 or step >= num_rollouts)
+                sync_due = step < num_rollouts and self._control.publish_lag > self._control.staleness_budget
+                if eval_due or save_due or sync_due:
+                    self._control.sync_rollout(self._async_engine, self.rollout, self.weight_sync)
+
+                if eval_due:
                     self.evaluate(rollout_id=rollout_id)
-                if save_interval > 0 and (step % save_interval == 0 or step >= num_rollouts):
-                    self._drain_all()
+                if save_due:
                     self.maybe_save_checkpoint(
-                        rollout_id, num_rollouts, save_interval=save_interval, save_dir=save_dir, save_mode=save_mode
+                        rollout_id,
+                        num_rollouts,
+                        save_interval=save_interval,
+                        save_dir=save_dir,
+                        save_mode=save_mode,
                     )
-                if step % interval == 0 and self.weight_sync is not None:
-                    self._drain_all()
-                    self.weight_sync.sync()
-                    self._async_engine.bump_weight_version()
         finally:
             active_exception = sys.exc_info()[0] is not None
             try:
@@ -354,34 +406,38 @@ def train(
             finally:
                 self._finish_wandb()
 
-    def _next_step(
+    def _next_rollout_batch(
         self,
         rollout_id: int,
-        interval: int,
-        M: int,
-        stale: int,
+        *,
         num_rollouts: int,
-    ) -> List[Sample]:
-        """Top up launches, reap completed generations, and return the freshest
-        ``batch_size`` scored group Samples for ``rollout_id`` (blocking on the
-        oldest in-flight generation if the buffer is short).
-
-        The launch clamp is the load-bearing on-policy guarantee: a generation
-        launched now is consumed later, so bound how far ahead we launch to
-        ``stale`` weight-syncs. ``stale=0`` ⇒ never launch into a future
-        sync-window ⇒ no generation crosses a sync ⇒ ``ratio≈1`` (on-policy).
-        """
+        hard_boundary: int,
+    ) -> RolloutBatch:
+        """Launch, reap, and consume one completion-order FIFO train batch."""
+
         engine = self._async_engine
         while True:
-            ceiling = launch_ceiling(rollout_id, sync_interval=interval, max_staleness=stale, num_rollouts=num_rollouts)
-            while engine.next_gen_id < ceiling and engine.inflight < M:
-                engine.submit(self._build_async_sample(engine.next_gen_id))
+            slots = self._control.launch_slots(
+                inflight_count=engine.inflight_count,
+                ready_count=engine.ready_count,
+                max_inflight=self._max_inflight,
+                trained_batches=rollout_id,
+                num_rollouts=num_rollouts,
+                hard_boundary=hard_boundary,
+            )
+            for _ in range(slots):
+                engine.submit(
+                    self._build_async_sample(engine.next_gen_id),
+                    behavior_version=self._control.rollout_version,
+                )
             engine.poll()
-            picked = engine.drain_freshest(self.batch_size, max_staleness=stale)
-            engine.pop_evicted()
-            if picked is not None:
-                return picked
-            if engine.inflight:
+            batch = engine.pop_next_batch(
+                train_version=self._control.train_version,
+                staleness_budget=self._control.staleness_budget,
+            )
+            if batch is not None:
+                return batch
+            if engine.inflight_count:
                 engine.wait_oldest()
             else:
-                raise RuntimeError("async rollout buffer underflow with no in-flight generations")
+                raise RuntimeError("async rollout queue is empty and the staleness budget admits no new generation")
diff --git a/unirl/trainer/async_diffusion.py b/unirl/trainer/async_diffusion.py
index 6b41db9f..f8662b56 100644
--- a/unirl/trainer/async_diffusion.py
+++ b/unirl/trainer/async_diffusion.py
@@ -17,17 +17,21 @@
 * ``_build_async_sample`` — one data batch → one request ``Sample``.
 * ``_score_completed`` — reward at reap time, then split into tree-complete
   groups. Generation overlaps training; reward scoring itself does not.
-* ``_advantage_and_train`` — advantage + FlowGRPO optimizer step over the
-  freshest ``batch_size`` groups; it never calls the reward.
+* ``_advantage_and_train`` — advantage + FlowGRPO optimizer step over the next
+  FIFO rollout batch; it never calls the reward.
 
-Two numeric knobs (identical semantics to AsyncARTrainer):
+Async control uses the same optimizer-update clock as AsyncARTrainer:
   * ``max_inflight`` — must be ``1`` so a reap-time transfer never competes with
     a queued generation on the rollout workers.
-  * ``buffer_max_staleness`` — regular rollout-weight syncs a buffered group may
-    cross. ``0`` (default) never crosses a sync; ``>0`` enables a bounded
-    policy-lag buffer.
+  * ``max_staleness`` — how many whole rollout batches the behavior policy may
+    trail the train policy by, checked at batch admission and consumption. The
+    clock underneath counts committed optimizer updates; the budget is stated in
+    batches because those are the only points where admission can react.
+    ``num_updates_per_batch > 1`` therefore also means the updates after the
+    first in a batch run at up to ``num_updates_per_batch - 1`` more updates of
+    drift than the admitted figure.
 
-``_next_step`` polls (reaps) BEFORE topping up launches, which is what makes the
+``_next_rollout_batch`` polls (reaps) BEFORE topping up launches, which is what makes the
 overlap fast here: reaping a generation pulls its trajectory segment off the
 rollout slab (the reward's cross-slab localize, an NCCL send issued on the
 rollout workers), so a generation launched ahead of that send blocks it —
@@ -50,8 +54,15 @@
 import torch
 
 from unirl.distributed.tensor import hydrate
-from unirl.rollout.engine.asynchronous import AsyncBatchRolloutEngine, launch_ceiling
+from unirl.rollout.engine.asynchronous import AsyncBatchRolloutEngine, RolloutBatch
 from unirl.train.stack import TrainStepResult
+from unirl.trainer.async_policy import (
+    AsyncBatchControl,
+    log_admission_notes,
+    next_hard_boundary,
+    sync_period_batches,
+    unwrap_replicated_int,
+)
 from unirl.trainer.diffusion import DiffusionTrainer
 from unirl.types.sample import Sample
 
@@ -65,7 +76,7 @@ def __init__(
         self,
         *,
         max_inflight: int = 1,
-        buffer_max_staleness: Optional[int] = None,
+        max_staleness: int = 0,
         **diffusion_kwargs: Any,
     ) -> None:
         layout = diffusion_kwargs.setdefault("layout", "separate")
@@ -86,7 +97,10 @@ def __init__(
             )
 
         self._max_inflight = max_inflight
-        self._buffer_max_staleness = buffer_max_staleness
+        self._control = AsyncBatchControl(
+            max_staleness=max_staleness,
+            num_updates_per_batch=diffusion_kwargs["stack_cfg"].get("num_updates_per_batch", 1),
+        )
 
     def _build_async_sample(self, gen_id: int) -> Sample:
         """Consume one data batch and build the request Sample for ``gen_id``."""
@@ -121,8 +135,9 @@ def _advantage_and_train(
         training_progress: float,
         rollout_id: int,
         t0: Optional[float] = None,
+        extra_metrics: Optional[dict[str, float]] = None,
     ) -> Tuple[TrainStepResult, float]:
-        """Advantage + optimizer step for a SCORED ``Sample`` (rewards already attached)."""
+        """Advantage + optimizer updates for a scored ``Sample`` (rewards already attached)."""
         if t0 is None:
             t0 = time.perf_counter()
         part = sample.parts[-1]
@@ -135,7 +150,16 @@ def _advantage_and_train(
         part = part.compute_advantages(normalize=True, use_global_std=self._adv_use_global_std)
         sample = sample.replace_frontier(part)
         result = self.stack.train_track(sample.parts[-1], training_progress=float(training_progress))
-        self.wandb_logger.log_rollout_step(rollout_id, result, sample, step_time_s=time.perf_counter() - t0)
+        self._control.record_optimizer_updates(result.optimizer_updates)
+        if extra_metrics is not None:
+            extra_metrics.update(self._control.train_metrics(result.optimizer_updates))
+        self.wandb_logger.log_rollout_step(
+            rollout_id,
+            result,
+            sample,
+            step_time_s=time.perf_counter() - t0,
+            extra_metrics=extra_metrics,
+        )
         self._reset_transport_buffers()
         return result, mean_reward
 
@@ -143,65 +167,99 @@ def train(
         self,
         *,
         num_rollouts: int,
-        weight_sync_interval: int = 1,
         save_interval: int = 0,
         save_dir: Optional[str] = None,
         load_dir: Optional[str] = None,
         save_mode: str = "auto",
     ) -> None:
-        interval = max(1, weight_sync_interval)
-        stale = self._buffer_max_staleness if self._buffer_max_staleness is not None else 0
-        M = self._max_inflight
-
         start_rollout = self.maybe_load_checkpoint(load_dir, num_rollouts=num_rollouts)
         resumed = bool(load_dir)
+        train_version = unwrap_replicated_int(
+            self.backend.get_optimizer_step_count(),
+            name="backend optimizer step count",
+        )
+        self._control.restore(train_version)
         for _ in range(start_rollout):
             self.data_source.get_samples(self.batch_size)
         self._init_wandb(
             num_rollouts=num_rollouts,
             extra={
-                "max_inflight": M,
-                "buffer_max_staleness": stale,
-                "weight_sync_interval": interval,
+                "max_inflight": self._max_inflight,
+                "max_staleness": self._control.max_staleness,
+                "staleness_budget": self._control.staleness_budget,
+                "num_updates_per_batch": self._control.num_updates_per_batch,
+                "sync_period_batches": sync_period_batches(
+                    self._control,
+                    eval_interval=self.eval_interval,
+                    save_interval=save_interval,
+                ),
                 "train_fraction": self._train_fraction,
             },
         )
+        # Reported here rather than in __init__: save_interval only arrives with
+        # the train call, and it clamps the publication period just as the
+        # staleness budget does.
+        log_admission_notes(
+            self._control,
+            max_inflight=self._max_inflight,
+            eval_interval=self.eval_interval,
+            save_interval=save_interval,
+        )
 
         self._async_engine = AsyncBatchRolloutEngine(
             self.rollout,
-            complete=self._score_completed,
+            process_completion=self._score_completed,
+            groups_per_batch=self.batch_size,
             start_gen_id=start_rollout,
         )
 
-        if resumed and self.weight_sync is not None:
-            self.weight_sync.sync()
+        if resumed:
+            self._control.sync_rollout(self._async_engine, self.rollout, self.weight_sync, force=True)
         if self.eval_interval > 0:
             self.evaluate(start_rollout, sync_weights=False, sleep_after=False)
 
         try:
             for rollout_id in range(start_rollout, num_rollouts):
                 t0 = time.perf_counter()
-                picked = self._next_step(rollout_id, interval, M, stale, num_rollouts)
-                sample = Sample.concat(picked)
+                hard_boundary = next_hard_boundary(
+                    rollout_id,
+                    num_rollouts=num_rollouts,
+                    eval_interval=self.eval_interval,
+                    save_interval=save_interval,
+                )
+                batch = self._next_rollout_batch(
+                    rollout_id,
+                    num_rollouts=num_rollouts,
+                    hard_boundary=hard_boundary,
+                )
+                sample = Sample.concat(batch.groups)
                 training_progress = rollout_id / max(1, num_rollouts - 1)
                 result, mean_reward = self._advantage_and_train(
-                    sample, training_progress=training_progress, rollout_id=rollout_id, t0=t0
+                    sample,
+                    training_progress=training_progress,
+                    rollout_id=rollout_id,
+                    t0=t0,
+                    extra_metrics=self._control.behavior_metrics(batch.behavior_version),
                 )
                 self.wandb_logger.log_progress(rollout_id, num_rollouts, result, mean_reward, logger=logger)
 
                 step = rollout_id + 1
-                if self.eval_interval > 0 and step % self.eval_interval == 0:
-                    self._drain_all()
+                eval_due = self.eval_interval > 0 and step % self.eval_interval == 0
+                save_due = save_interval > 0 and (step % save_interval == 0 or step >= num_rollouts)
+                sync_due = step < num_rollouts and self._control.publish_lag > self._control.staleness_budget
+                if eval_due or save_due or sync_due:
+                    self._control.sync_rollout(self._async_engine, self.rollout, self.weight_sync)
+
+                if eval_due:
                     self.evaluate(step, sync_weights=False, sleep_after=False)
-                if save_interval > 0 and (step % save_interval == 0 or step >= num_rollouts):
-                    self._drain_all()
+                if save_due:
                     self.maybe_save_checkpoint(
-                        rollout_id, num_rollouts, save_interval=save_interval, save_dir=save_dir, save_mode=save_mode
+                        rollout_id,
+                        num_rollouts,
+                        save_interval=save_interval,
+                        save_dir=save_dir,
+                        save_mode=save_mode,
                     )
-                if step % interval == 0 and self.weight_sync is not None:
-                    self._drain_all()
-                    self.weight_sync.sync()
-                    self._async_engine.bump_weight_version()
         finally:
             # Cleanup failures must not mask the exception that stopped training.
             active_exception = sys.exc_info()[0] is not None
@@ -214,38 +272,42 @@ def train(
             finally:
                 self._finish_wandb()
 
-    def _next_step(
+    def _next_rollout_batch(
         self,
         rollout_id: int,
-        interval: int,
-        M: int,
-        stale: int,
+        *,
         num_rollouts: int,
-    ) -> List[Sample]:
-        """Reap completed generations, top up launches, and return the freshest
-        ``batch_size`` scored group Samples for ``rollout_id`` (blocking on the
-        oldest in-flight generation if the buffer is short).
+        hard_boundary: int,
+    ) -> RolloutBatch:
+        """Reap, launch, and consume one completion-order FIFO train batch.
 
         Polls BEFORE topping up: reaping pulls the trajectory segment off the
         rollout slab, so it must not queue behind a freshly launched generation,
         and the post-reap launch is what overlaps this step (module docstring).
-
-        The launch clamp is the load-bearing on-policy guarantee: a generation
-        launched now is consumed later, so bound how far ahead we launch to
-        ``stale`` weight-syncs. ``stale=0`` ⇒ never launch into a future
-        sync-window ⇒ no generation crosses a regular rollout-weight sync.
         """
         engine = self._async_engine
         while True:
-            ceiling = launch_ceiling(rollout_id, sync_interval=interval, max_staleness=stale, num_rollouts=num_rollouts)
             engine.poll()
-            while engine.next_gen_id < ceiling and engine.inflight < M:
-                engine.submit(self._build_async_sample(engine.next_gen_id))
-            picked = engine.drain_freshest(self.batch_size, max_staleness=stale)
-            engine.pop_evicted()
-            if picked is not None:
-                return picked
-            if engine.inflight:
+            slots = self._control.launch_slots(
+                inflight_count=engine.inflight_count,
+                ready_count=engine.ready_count,
+                max_inflight=self._max_inflight,
+                trained_batches=rollout_id,
+                num_rollouts=num_rollouts,
+                hard_boundary=hard_boundary,
+            )
+            for _ in range(slots):
+                engine.submit(
+                    self._build_async_sample(engine.next_gen_id),
+                    behavior_version=self._control.rollout_version,
+                )
+            batch = engine.pop_next_batch(
+                train_version=self._control.train_version,
+                staleness_budget=self._control.staleness_budget,
+            )
+            if batch is not None:
+                return batch
+            if engine.inflight_count:
                 engine.wait_oldest()
             else:
-                raise RuntimeError("async rollout buffer underflow with no in-flight generations")
+                raise RuntimeError("async rollout queue is empty and the staleness budget admits no new generation")
diff --git a/unirl/trainer/async_policy.py b/unirl/trainer/async_policy.py
new file mode 100644
index 00000000..b846ed21
--- /dev/null
+++ b/unirl/trainer/async_policy.py
@@ -0,0 +1,247 @@
+"""Optimizer-update policy control shared by async AR and diffusion trainers.
+
+Two quantities ride the optimizer-update clock and only one of them is staleness:
+
+* ``staleness`` — updates between the behavior policy that generated a batch and
+  the train weights that batch starts training against, i.e. the off-policyness
+  of the data. This is AReaL's ``eta`` / ``max_head_offpolicyness``.
+* ``publish_lag`` — updates between the current train weights and the snapshot
+  last published to the rollout engine. This is sync debt: no batch is that
+  stale, but it is what makes a weight sync due.
+
+Recipes state the budget in whole rollout batches (``max_staleness``) because
+admission and consumption only ever run at a batch boundary. Stating it in raw
+updates instead would quantize it to ``num_updates_per_batch`` — 22 and 23 would
+both mean a 12-batch depth — and silently change meaning whenever that count
+changes. ``staleness_budget`` converts once into the clock the versions count in.
+
+Batch entry is also the only point the budget is enforced at, which matters once
+``num_updates_per_batch > 1``: the anchor is frozen for the whole batch while the
+weights keep moving, so update ``i`` trains at ``staleness + i - 1`` and the worst
+case any gradient step sees is ``staleness_budget + num_updates_per_batch - 1``.
+That extra span is the in-batch off-policyness PPO already assumes — the frozen
+anchor and ``clip_range`` cover it — so it is deliberately outside the budget.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class AsyncBatchControl:
+    """Track train/rollout versions and gate batch generation."""
+
+    max_staleness: int
+    num_updates_per_batch: int
+    train_version: int = 0
+    rollout_version: int = 0
+
+    def __post_init__(self) -> None:
+        if self.max_staleness < 0:
+            raise ValueError(f"max_staleness must be >= 0, got {self.max_staleness}")
+        if self.num_updates_per_batch < 1:
+            raise ValueError(f"num_updates_per_batch must be >= 1, got {self.num_updates_per_batch}")
+        if self.rollout_version > self.train_version:
+            raise ValueError(
+                f"rollout_version cannot be ahead of train_version: {self.rollout_version} > {self.train_version}"
+            )
+
+    @property
+    def staleness_budget(self) -> int:
+        """``max_staleness`` batches expressed in committed optimizer updates."""
+
+        return self.max_staleness * self.num_updates_per_batch
+
+    @property
+    def publish_lag(self) -> int:
+        """Sync debt: updates the published rollout snapshot trails train by."""
+
+        return self.train_version - self.rollout_version
+
+    @property
+    def admission_depth(self) -> int:
+        """Outstanding generations the budget allows, in whole batches."""
+
+        return self.max_staleness + 1
+
+    def restore(self, train_version: int) -> None:
+        self.train_version = train_version
+        self.rollout_version = 0
+
+    def staleness(self, behavior_version: int) -> int:
+        """Optimizer updates between train and a batch's behavior policy."""
+
+        stale = self.train_version - behavior_version
+        if stale < 0:
+            raise ValueError(
+                f"rollout batch has future behavior version {behavior_version} > train version {self.train_version}"
+            )
+        return stale
+
+    def record_optimizer_updates(self, optimizer_updates: int) -> None:
+        self.train_version += optimizer_updates
+
+    def launch_slots(
+        self,
+        *,
+        inflight_count: int,
+        ready_count: int,
+        max_inflight: int,
+        trained_batches: int,
+        num_rollouts: int,
+        hard_boundary: int,
+    ) -> int:
+        if self.publish_lag > self.staleness_budget:
+            return 0
+        # Floor division is what keeps a partially-committed step honest: a step
+        # that skipped updates moved the clock by less than a whole batch, and
+        # rounding down never admits a generation the budget cannot cover.
+        freshness = (self.staleness_budget - self.publish_lag) // self.num_updates_per_batch + 1
+        allowed = min(freshness, num_rollouts - trained_batches, hard_boundary - trained_batches)
+        return max(0, min(max_inflight - inflight_count, allowed - inflight_count - ready_count))
+
+    def sync_rollout(self, engine: Any, rollout: Any, weight_sync: Any, *, force: bool = False) -> bool:
+        if not force and self.publish_lag == 0:
+            return False
+        engine.quiesce()
+        if engine.ready_count:
+            raise RuntimeError(
+                f"cannot sync rollout weights with completed batches queued: ready_count={engine.ready_count}"
+            )
+        weight_sync.sync()
+        rollout.set_policy_version(self.train_version)
+        self.rollout_version = self.train_version
+        return True
+
+    def behavior_metrics(self, behavior_version: int) -> dict[str, float]:
+        staleness = self.staleness(behavior_version)
+        return {
+            "async/behavior_version": behavior_version,
+            "async/staleness_updates": staleness,
+            "async/staleness_batches": staleness / self.num_updates_per_batch,
+        }
+
+    def train_metrics(self, optimizer_updates: int) -> dict[str, int]:
+        return {
+            "async/train_version": self.train_version,
+            "async/publish_lag": self.publish_lag,
+            "async/optimizer_updates": optimizer_updates,
+        }
+
+
+def sync_period_batches(
+    control: AsyncBatchControl,
+    *,
+    eval_interval: int = 0,
+    save_interval: int = 0,
+) -> int:
+    """Batches between weight publications, once every admission limit is applied.
+
+    The staleness budget alone would publish every ``admission_depth`` batches,
+    but :func:`next_hard_boundary` clamps admission as well, so an eval or
+    checkpoint interval below that depth becomes the period instead.
+    """
+
+    period = control.admission_depth
+    for interval in (eval_interval, save_interval):
+        if interval > 0:
+            period = min(period, interval)
+    return period
+
+
+def log_admission_notes(
+    control: AsyncBatchControl,
+    *,
+    max_inflight: int,
+    eval_interval: int = 0,
+    save_interval: int = 0,
+) -> None:
+    """Report admission settings whose effect differs from what the value suggests.
+
+    All of these are legitimate configurations, so none is an error; each is a
+    case where the recipe's number does not buy what its name implies.
+    """
+
+    period = sync_period_batches(control, eval_interval=eval_interval, save_interval=save_interval)
+
+    if control.max_staleness == 0:
+        logger.warning(
+            "max_staleness=0 admits one generation at a time; generation cannot overlap the preceding train batch"
+        )
+    if max_inflight > control.admission_depth:
+        logger.warning(
+            "max_inflight=%d exceeds the staleness admission depth %d; the extra concurrency cannot be used",
+            max_inflight,
+            control.admission_depth,
+        )
+    # The loop reaps before it launches, so one completed batch can sit in the
+    # ready queue behind the in-flight ones; anything past that never becomes
+    # concurrency, it only defers the sync.
+    usable_depth = max_inflight + 1
+    if control.admission_depth > usable_depth:
+        logger.info(
+            "max_staleness=%d admits %d outstanding batches but the loop holds at most %d "
+            "(max_inflight=%d plus one reaped); the surplus does not deepen the pipeline, it "
+            "sets the weight-sync period to %d batches",
+            control.max_staleness,
+            control.admission_depth,
+            usable_depth,
+            max_inflight,
+            period,
+        )
+    if period < control.admission_depth:
+        logger.warning(
+            "eval/checkpoint boundaries publish every %d batches, below the %d the staleness "
+            "budget allows (eval_interval=%d, save_interval=%d), so max_staleness=%d is never "
+            "fully spent — data tops out at %d batches stale",
+            period,
+            control.admission_depth,
+            eval_interval,
+            save_interval,
+            control.max_staleness,
+            period - 1,
+        )
+
+
+def next_hard_boundary(
+    trained_batches: int,
+    *,
+    num_rollouts: int,
+    eval_interval: int = 0,
+    save_interval: int = 0,
+) -> int:
+    """Nearest eval/checkpoint/final boundary for launch admission."""
+
+    boundary = num_rollouts
+    for interval in (eval_interval, save_interval):
+        if interval > 0 and trained_batches < num_rollouts:
+            boundary = min(boundary, ((trained_batches // interval) + 1) * interval)
+    return boundary
+
+
+def unwrap_replicated_int(value: object, *, name: str) -> int:
+    """Normalize a BROADCAST return and verify all worker replicas agree."""
+
+    if isinstance(value, (list, tuple)):
+        if not value or any(not isinstance(item, int) for item in value):
+            raise TypeError(f"{name} returned invalid worker values: {value!r}")
+        if any(item != value[0] for item in value[1:]):
+            raise RuntimeError(f"{name} disagrees across workers: {value!r}")
+        return value[0]
+    if not isinstance(value, int):
+        raise TypeError(f"{name} returned {type(value).__name__}, expected int")
+    return value
+
+
+__all__ = [
+    "AsyncBatchControl",
+    "log_admission_notes",
+    "next_hard_boundary",
+    "sync_period_batches",
+    "unwrap_replicated_int",
+]