diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index b975cf4af..4e358d41d 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -28,7 +28,7 @@
 
 # --- rollout ---------------------------------------------------------------
 /unirl/rollout/                         @celve @zzhuoxin1508 @leviking98z-rgb
-/unirl/rollout/async_runtime.py         @CjhHa1 @celve
+/unirl/rollout/engine/asynchronous.py         @CjhHa1 @celve
 /unirl/rollout/engine/sglang/           @celve @leviking98z-rgb
 /unirl/rollout/engine/sglang_diffusion/ @celve @leviking98z-rgb
 /unirl/rollout/engine/vllm_omni/        @celve @zzhuoxin1508
diff --git a/examples/diffusion/bagel/bagel_vllmomni_async.yaml b/examples/diffusion/bagel/bagel_vllmomni_async.yaml
index f567f83ce..e3b766d2f 100644
--- a/examples/diffusion/bagel/bagel_vllmomni_async.yaml
+++ b/examples/diffusion/bagel/bagel_vllmomni_async.yaml
@@ -55,9 +55,9 @@ weight_sync_interval: 4
 # 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 shared async runtime (reap_before_launch) reaps and
-# transfers each generation in the idle window before launching the next, then
-# overlaps that next generation with the train step. Real overlap needs
+# 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).
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 000000000..996d605d4
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,6 @@
+import sys
+from pathlib import Path
+
+# Make `import unirl` work without installing the package (the CPU-only tests
+# import ray/torch-free modules only).
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
diff --git a/tests/test_async_engine.py b/tests/test_async_engine.py
new file mode 100644
index 000000000..d96ce682c
--- /dev/null
+++ b/tests/test_async_engine.py
@@ -0,0 +1,234 @@
+"""CPU-only tests for the driver-side async rollout engines (stubbed handles)."""
+
+import pytest
+
+from unirl.rollout.engine.asynchronous import AsyncAgenticRolloutEngine, AsyncBatchRolloutEngine, root_of
+
+# ── batch engine stubs ──
+
+
+class StubPending:
+    def __init__(self, value):
+        self.value = value
+
+    def ready(self):
+        return True
+
+    def wait(self):
+        pass
+
+    def result(self):
+        return self.value
+
+
+class StubBatchHandle:
+    """launch_nowait echoes the submitted sample back as the completed value."""
+
+    def __init__(self):
+        self.launched = []
+
+    def launch_nowait(self, method, sample):
+        self.launched.append((method, sample))
+        return StubPending(f"done:{sample}")
+
+
+# ── agentic stubs ──
+
+
+class _Part:
+    def __init__(self, root):
+        self.sample_ids = [root]
+
+
+class Traj:
+    def __init__(self, root):
+        self.parts = [_Part(root)]
+
+    def __repr__(self):
+        return f"Traj({self.parts[0].sample_ids[0]})"
+
+
+class StubAgenticHandle:
+    """Coordinator handle: every value returns BROADCAST+RANK_ZERO-shaped [value]."""
+
+    def __init__(self):
+        self.submitted = []
+        self.poll_batches = []
+        self.finalize_value = None
+        self.carried = []
+        self.abort_calls = 0
+
+    def submit(self, tasks):
+        self.submitted.append(list(tasks))
+
+    def poll(self):
+        batch = self.poll_batches.pop(0) if self.poll_batches else []
+        return [batch]
+
+    def finalize_if_drained(self):
+        return [self.finalize_value]
+
+    def abort(self):
+        self.abort_calls += 1
+        return [list(self.carried)]
+
+
+# ── AsyncBatchRolloutEngine ──
+
+
+def test_batch_stamps_weight_version_at_launch():
+    handle = StubBatchHandle()
+    engine = AsyncBatchRolloutEngine(handle, complete=lambda gen_id, done: [done])
+    engine.submit("s0")           # launched under version 0
+    engine.bump_weight_version()  # sync happens while s0 is in flight
+    assert engine.poll() == 1
+
+    # Launch-time stamp: at staleness 0 the group is one sync stale → evicted.
+    assert engine.drain_freshest(1, max_staleness=0) is None
+    assert engine.pop_evicted() == ["done:s0"]
+
+
+def test_batch_gen_id_seed_and_next_gen_id():
+    handle = StubBatchHandle()
+    engine = AsyncBatchRolloutEngine(handle, complete=lambda g, d: [d], start_gen_id=5)
+    assert engine.next_gen_id == 5
+    assert engine.submit("s") == 5
+    assert engine.next_gen_id == 6
+    assert engine.inflight == 1
+
+
+def test_batch_quiesce_drains_everything_and_returns_empty():
+    handle = StubBatchHandle()
+    seen = []
+    engine = AsyncBatchRolloutEngine(handle, complete=lambda g, d: seen.append(g) or [d])
+    engine.submit("s0")
+    engine.submit("s1")
+    assert engine.quiesce() == []
+    assert engine.inflight == 0
+    assert seen == [0, 1]
+    assert engine.drain_freshest(2, max_staleness=0) == ["done:s1", "done:s0"]
+
+
+def test_batch_complete_failure_causes_no_double_put_on_retry():
+    handle = StubBatchHandle()
+    fails = [True]
+
+    def complete(gen_id, done):
+        if fails and fails.pop():
+            raise RuntimeError("score boom")
+        return [done]
+
+    engine = AsyncBatchRolloutEngine(handle, complete=complete)
+    engine.submit("s0")
+    with pytest.raises(RuntimeError, match="score boom"):
+        engine.poll()
+    assert engine.inflight == 1  # retained for retry, nothing buffered
+
+    assert engine.poll() == 1
+    assert engine.drain_freshest(1, max_staleness=0) == ["done:s0"]
+    assert engine.drain_freshest(1, max_staleness=0) is None  # exactly once
+
+
+def test_batch_complete_may_split_into_multiple_groups():
+    handle = StubBatchHandle()
+    engine = AsyncBatchRolloutEngine(handle, complete=lambda g, d: [f"{d}#0", f"{d}#1"])
+    engine.submit("s0")
+    engine.poll()
+    assert engine.drain_freshest(2, max_staleness=0) == ["done:s0#0", "done:s0#1"]
+
+
+# ── AsyncAgenticRolloutEngine ──
+
+
+def test_agentic_poll_unwraps_and_assembles_n_sibling_groups():
+    handle = StubAgenticHandle()
+    engine = AsyncAgenticRolloutEngine(handle, group_size=2)
+    a0, a1, b0 = Traj("a"), Traj("a"), Traj("b")
+
+    handle.poll_batches.append([a0, b0])
+    assert engine.poll() == 2
+    assert engine.pending_groups() == 2  # both roots incomplete
+    assert engine.buffered_groups() == 0
+
+    handle.poll_batches.append([a1])
+    assert engine.poll() == 1
+    assert engine.buffered_groups() == 1
+    assert engine.pending_groups() == 1  # root b still waiting
+
+    assert engine.drain_freshest(1, max_staleness=0) == [[a0, a1]]
+
+
+def test_agentic_stamps_weight_version_at_completion():
+    handle = StubAgenticHandle()
+    engine = AsyncAgenticRolloutEngine(handle, group_size=2)
+    handle.poll_batches.append([Traj("a")])
+    engine.poll()                 # first sibling under version 0
+    engine.bump_weight_version()
+    handle.poll_batches.append([Traj("a")])
+    engine.poll()                 # group completes under version 1
+
+    # Completion-time stamp: fresh at current version, staleness 0.
+    assert engine.drain_freshest(1, max_staleness=0) is not None
+
+
+def test_agentic_quiesce_polls_after_abort_and_before_bump():
+    handle = StubAgenticHandle()
+    engine = AsyncAgenticRolloutEngine(handle, group_size=1)
+    handle.carried = [Traj("tail")]
+    handle.poll_batches.append([Traj("done-during-quiesce")])
+
+    carried = engine.quiesce()
+    engine.bump_weight_version()  # trainer syncs after the quiesce
+
+    assert handle.abort_calls == 1
+    assert [root_of(t) for t in carried] == ["tail"]
+    # The group that completed during the quiesce was stamped pre-bump …
+    assert engine.buffered_groups() == 1
+    # … so at staleness 0 it is one sync stale and must evict, proving the
+    # stamp happened before bump_weight_version.
+    assert engine.drain_freshest(1, max_staleness=0) is None
+    assert len(engine.pop_evicted()) == 1
+
+
+def test_agentic_finalize_if_drained_passthrough_and_ingest():
+    handle = StubAgenticHandle()
+    engine = AsyncAgenticRolloutEngine(handle, group_size=1)
+
+    handle.finalize_value = None
+    assert engine.finalize_if_drained() is None
+
+    handle.finalize_value = [Traj("a")]
+    assert engine.finalize_if_drained() == 1
+    assert engine.buffered_groups() == 1
+
+
+def test_agentic_discard_roots_drops_partial_buckets():
+    handle = StubAgenticHandle()
+    engine = AsyncAgenticRolloutEngine(handle, group_size=2)
+    handle.poll_batches.append([Traj("a"), Traj("b")])
+    engine.poll()
+
+    assert engine.discard_roots(["a"]) == 1
+    assert engine.pending_groups() == 1
+
+
+def test_agentic_submit_passthrough():
+    handle = StubAgenticHandle()
+    engine = AsyncAgenticRolloutEngine(handle, group_size=1)
+    tasks = [Traj("a")]
+    engine.submit(tasks)
+    assert handle.submitted == [tasks]
+
+
+def test_agentic_gen_id_orders_groups_by_completion():
+    handle = StubAgenticHandle()
+    engine = AsyncAgenticRolloutEngine(handle, group_size=1)
+    first, second = Traj("a"), Traj("b")
+    handle.poll_batches.append([first])
+    engine.poll()
+    handle.poll_batches.append([second])
+    engine.poll()
+
+    # Freshest-first: the later-completed group drains first.
+    assert engine.drain_freshest(1, max_staleness=0) == [[second]]
+    assert engine.drain_freshest(1, max_staleness=0) == [[first]]
diff --git a/tests/test_async_runtime.py b/tests/test_async_runtime.py
new file mode 100644
index 000000000..40a8a6609
--- /dev/null
+++ b/tests/test_async_runtime.py
@@ -0,0 +1,198 @@
+"""CPU-only tests for the driver-side async mechanisms (VersionedBuffer, InflightPool)."""
+
+import pytest
+
+from unirl.rollout.engine.asynchronous import InflightPool, VersionedBuffer
+
+
+class StubPending:
+    """Stands in for handle.PendingHandleCall."""
+
+    def __init__(self, value, *, ready=True, fail_results=0):
+        self.value = value
+        self._ready = ready
+        self._fail_results = fail_results
+        self.result_calls = 0
+        self.wait_calls = 0
+
+    def ready(self):
+        return self._ready
+
+    def wait(self):
+        self.wait_calls += 1
+
+    def result(self):
+        self.result_calls += 1
+        if self._fail_results > 0:
+            self._fail_results -= 1
+            raise RuntimeError("result boom")
+        return self.value
+
+
+class StubRollout:
+    """Hands out pre-built StubPending objects in launch order."""
+
+    def __init__(self, pendings):
+        self._pendings = list(pendings)
+        self.launched = []
+
+    def launch_nowait(self, method, sample):
+        self.launched.append((method, sample))
+        return self._pendings.pop(0)
+
+
+class Recorder:
+    def __init__(self, fail_times=0, raise_cls=RuntimeError):
+        self.calls = []
+        self._fail_times = fail_times
+        self._raise_cls = raise_cls
+
+    def __call__(self, gen_id, weight_version, payload):
+        if self._fail_times > 0:
+            self._fail_times -= 1
+            raise self._raise_cls("complete boom")
+        self.calls.append((gen_id, weight_version, payload))
+
+
+# ── VersionedBuffer ──
+
+
+def test_drain_freshest_orders_by_gen_id_desc():
+    buf = VersionedBuffer()
+    for gen_id in (1, 3, 2):
+        buf.put(f"g{gen_id}", weight_version=0, gen_id=gen_id)
+    assert buf.drain_freshest(2) == ["g3", "g2"]
+    assert buf.size() == 1
+    assert buf.drain_freshest(1) == ["g1"]
+
+
+def test_drain_freshest_stable_on_gen_id_ties():
+    buf = VersionedBuffer()
+    buf.put("first", weight_version=0, gen_id=7)
+    buf.put("second", weight_version=0, gen_id=7)
+    assert buf.drain_freshest(2) == ["first", "second"]
+
+
+def test_drain_freshest_returns_none_without_consuming():
+    buf = VersionedBuffer()
+    buf.put("only", weight_version=0, gen_id=0)
+    assert buf.drain_freshest(2) is None
+    assert buf.size() == 1
+
+
+def test_staleness_eviction_and_pop_evicted():
+    buf = VersionedBuffer()
+    buf.put("stale", weight_version=0, gen_id=0)
+    buf.put("fresh", weight_version=2, gen_id=1)
+    assert buf.drain_freshest(2, current_version=2, max_staleness=1) is None
+    assert buf.pop_evicted() == ["stale"]
+    assert buf.pop_evicted() == []
+    assert buf.drain_freshest(1, current_version=2, max_staleness=1) == ["fresh"]
+
+
+# ── InflightPool ──
+
+
+def test_launch_stamps_and_reap_completes():
+    rollout = StubRollout([StubPending("a"), StubPending("b")])
+    pool = InflightPool(rollout, start_gen_id=10)
+    complete = Recorder()
+
+    assert pool.launch("s0", weight_version=0) == 10
+    assert pool.launch("s1", weight_version=1) == 11
+    assert pool.next_gen_id == 12
+    assert len(pool) == 2
+    assert rollout.launched == [("generate", "s0"), ("generate", "s1")]
+
+    assert pool.reap_ready(complete) == 2
+    assert complete.calls == [(10, 0, "a"), (11, 1, "b")]
+    assert len(pool) == 0
+
+
+def test_reap_skips_unready_jobs():
+    rollout = StubRollout([StubPending("a", ready=False), StubPending("b")])
+    pool = InflightPool(rollout)
+    complete = Recorder()
+    pool.launch("s0", weight_version=0)
+    pool.launch("s1", weight_version=0)
+
+    assert pool.reap_ready(complete) == 1
+    assert complete.calls == [(1, 0, "b")]
+    assert len(pool) == 1
+
+
+def test_complete_failure_keeps_job_in_flight_for_retry():
+    rollout = StubRollout([StubPending("a")])
+    pool = InflightPool(rollout)
+    complete = Recorder(fail_times=1)
+    pool.launch("s0", weight_version=0)
+
+    with pytest.raises(RuntimeError, match="complete boom"):
+        pool.reap_ready(complete)
+    assert len(pool) == 1
+
+    assert pool.reap_ready(complete) == 1
+    assert complete.calls == [(0, 0, "a")]
+    assert len(pool) == 0
+
+
+def test_result_failure_keeps_job_in_flight_for_retry():
+    pending = StubPending("a", fail_results=1)
+    rollout = StubRollout([pending])
+    pool = InflightPool(rollout)
+    complete = Recorder()
+    pool.launch("s0", weight_version=0)
+
+    with pytest.raises(RuntimeError, match="result boom"):
+        pool.reap_ready(complete)
+    assert len(pool) == 1 and complete.calls == []
+
+    assert pool.reap_ready(complete) == 1
+    assert complete.calls == [(0, 0, "a")]
+
+
+def test_first_error_reraised_after_full_sweep():
+    rollout = StubRollout([StubPending("a"), StubPending("b")])
+    pool = InflightPool(rollout)
+    complete = Recorder(fail_times=2)
+    pool.launch("s0", weight_version=0)
+    pool.launch("s1", weight_version=0)
+
+    with pytest.raises(RuntimeError):
+        pool.reap_ready(complete)
+    assert len(pool) == 2  # both swept, both retained
+
+
+def test_keyboard_interrupt_propagates_immediately():
+    rollout = StubRollout([StubPending("a"), StubPending("b")])
+    pool = InflightPool(rollout)
+    complete = Recorder(fail_times=1, raise_cls=KeyboardInterrupt)
+    pool.launch("s0", weight_version=0)
+    pool.launch("s1", weight_version=0)
+
+    with pytest.raises(KeyboardInterrupt):
+        pool.reap_ready(complete)
+    assert complete.calls == []  # second job never attempted
+    assert len(pool) == 2  # sweep aborted before reassignment
+
+
+def test_drain_all_completes_unready_jobs_too():
+    rollout = StubRollout([StubPending("a", ready=False), StubPending("b")])
+    pool = InflightPool(rollout)
+    complete = Recorder()
+    pool.launch("s0", weight_version=0)
+    pool.launch("s1", weight_version=0)
+
+    assert pool.drain_all(complete) == 2
+    assert len(pool) == 0
+
+
+def test_wait_oldest_blocks_on_first_job_only():
+    p0, p1 = StubPending("a", ready=False), StubPending("b", ready=False)
+    rollout = StubRollout([p0, p1])
+    pool = InflightPool(rollout)
+    pool.launch("s0", weight_version=0)
+    pool.launch("s1", weight_version=0)
+
+    pool.wait_oldest()
+    assert (p0.wait_calls, p1.wait_calls) == (1, 0)
diff --git a/unirl/distributed/group/handle.py b/unirl/distributed/group/handle.py
index ce9f5961f..72721719f 100644
--- a/unirl/distributed/group/handle.py
+++ b/unirl/distributed/group/handle.py
@@ -185,6 +185,50 @@ class HandleRef:
     sp_size: int = 1
 
 
+class PendingHandleCall:
+    """Future-like result of :meth:`Handle.launch_nowait`: launched, not yet collected.
+
+    ``ready()`` probes without blocking; ``wait()`` blocks without collecting;
+    ``result()`` blocks if needed, then runs the rebind + collect half of
+    ``handle_fn`` and returns the method's collected value. The collect half
+    runs at most once — rebind registers GC finalizers on the result refs — so
+    a successful ``result()`` caches its value and later calls return it; a
+    ``result()`` that raised may be retried.
+    """
+
+    def __init__(self, handle: "Handle", method_name: str, refs: List[Any], worker_local: bool) -> None:
+        self._handle = handle
+        self._method_name = method_name
+        self._refs = refs
+        self._worker_local = worker_local
+        self._consumed = False
+        self._value: Any = None
+
+    def ready(self) -> bool:
+        """True once every worker's ref is resolved (non-blocking probe)."""
+        done, _ = ray.wait(self._refs, num_returns=len(self._refs), timeout=0)
+        return len(done) == len(self._refs)
+
+    def wait(self) -> None:
+        """Block until every worker finishes, without collecting; re-raises worker errors."""
+        ray.get(self._refs)
+
+    def result(self) -> Any:
+        """Block if needed, then rebind + collect: the method's collected return value."""
+        if self._consumed:
+            return self._value
+        handle = self._handle
+        results = ray.get(self._refs)
+        results = [
+            handle._rebind_tree(r, handle.workers[i], worker_local=self._worker_local)
+            for i, r in enumerate(results)
+        ]
+        _, _, collect_fn, _ = handle._method_configs[self._method_name]
+        self._value = collect_fn(handle, results)
+        self._consumed = True
+        return self._value
+
+
 class Handle:
     """Controller-side SPMD handle.
 
@@ -274,6 +318,7 @@ def __init__(
         )
 
         # Bind @distributed methods as handle functions
+        self._method_configs: Dict[str, tuple] = {}
         self._bind_methods(role_cls)
 
         # Counter for unique call_id generation within enable_grad contexts.
@@ -338,6 +383,7 @@ def _bind_methods(self, role_cls) -> None:
             else:
                 execute_fn = self._execute_rank_zero
 
+            self._method_configs[name] = (config["dispatch_mode"], dispatch_fn, collect_fn, execute_fn)
             bound = self._make_handle_fn(name, config["dispatch_mode"], dispatch_fn, collect_fn, execute_fn)
             setattr(self, name, bound)
 
@@ -355,6 +401,9 @@ def _make_handle_fn(
         TensorMetas and append an RPCBackwardNode for later auto-backward.
         grad_mode and call_id are passed as dedicated parameters to Worker.call
         (not via kwargs) so dispatch internals remain unaware of grad state.
+
+        The non-blocking twin is :meth:`launch_nowait` +
+        :meth:`PendingHandleCall.result` below — keep the halves in parity.
         """
 
         def handle_fn(*args, **kwargs):
@@ -424,6 +473,40 @@ def handle_fn(*args, **kwargs):
         handle_fn.__doc__ = f"SPMD handle: {method_name} (dispatch={dispatch_fn.__name__})"
         return handle_fn
 
+    # ── Non-blocking launch ──
+
+    def launch_nowait(self, method_name: str, *args, **kwargs) -> PendingHandleCall:
+        """Launch a @distributed method without blocking: the dispatch → localize →
+        execute half of ``handle_fn``, stopping before ``ray.get``.
+
+        Always ``grad_mode=False`` / ``call_id=None`` (a pending call is never
+        valid under a GradContext, so the ``_grad_call_counter`` single-thread
+        assumption is untouched). Kept in line-parity with ``handle_fn`` above —
+        same divisibility gate, same localize. ``result()`` on the returned
+        :class:`PendingHandleCall` runs the collect half.
+        """
+        try:
+            dispatch_mode, dispatch_fn, _, execute_fn = self._method_configs[method_name]
+        except KeyError:
+            raise AttributeError(
+                f"{method_name!r} is not a @distributed method of {_owning_class(self.role_cls).__name__}"
+            ) from None
+
+        batch_size = infer_batch_size(args, kwargs)
+        if (
+            dispatch_mode in (Dispatch.DP_SCATTER, Dispatch.DP_SCATTER_HEAD)
+            and batch_size is not None
+            and batch_size % self.dp_size != 0
+        ):
+            raise ValueError(f"batch_size={batch_size} not divisible by dp_size={self.dp_size}")
+
+        shards = dispatch_fn(self, args, kwargs, batch_size)
+        transport_cls = self.pool.transport_cls
+        worker_local = issubclass(transport_cls, WorkerLocalTransport)
+        shards = transport_cls.localize(shards, self.pool, self.device_ids, self.worker_ids)
+        refs = execute_fn(method_name, shards, grad_mode=False, call_id=None)
+        return PendingHandleCall(self, method_name, refs, worker_local)
+
     # ── Execute strategies ──
 
     def _execute_all(self, method_name: str, shards: List, grad_mode: bool = False, call_id=None) -> List:
diff --git a/unirl/rollout/README.md b/unirl/rollout/README.md
index 178e7843e..41475f52d 100644
--- a/unirl/rollout/README.md
+++ b/unirl/rollout/README.md
@@ -35,7 +35,7 @@ wrong objective.
 
 ## How it works
 
-- **One synchronous generation interface.** `BaseRolloutEngine` (`engine/base.py`)
+- **One synchronous generation interface.** `BaseRolloutEngine` (`engine/synchronous.py`)
   is a `Remote` whose concrete engines implement synchronous `generate(sample)`;
   each keeps its native batching/runtime path. Single-turn engines return one
   `Sample` and dispatch `generate` with `DP_SCATTER`; the agentic coordinator
@@ -63,10 +63,20 @@ wrong objective.
   ratio is 1 on the first update; *separate* — a dedicated engine on its own GPUs
   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). The `AsyncRolloutEngine`
+  protocol — the async sibling of `SyncRolloutEngine` — is what the async
+  trainers program against: `poll` / `drain_freshest` / `pop_evicted` / `quiesce` +
+  engine-owned `weight_version`. Two concretes: `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
+  coordinator; normalizes the `[0]` unwraps, assembles n-sibling GRPO groups,
+  stamps versions at completion, used by the 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
-`engine/<name>/engine.py` (subclass `BaseSingleTurnRolloutEngine`, implement
+`engine/<name>/engine.py` (subclass `SyncRolloutEngine`, implement
 synchronous generation over the whole-`Sample` contract — thread-safe for
 concurrent callers if it should serve as an agentic inner, else serialized
 internally — and dispatch `generate` with `DP_SCATTER`). A dedicated engine also
@@ -83,6 +93,12 @@ implements its weight-receive method and a matching `sync:` handler in
   intentional exception: `BROADCAST + RANK_ZERO` returns its trajectory list.
 - **Direct sampling forbids a `sync:` block; dedicated requires one.** The trainside
   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
+  turn-boundary `abort` + final poll, folded into
+  `AsyncAgenticRolloutEngine.quiesce()`. Reap-vs-launch ordering is trainer
+  statement order (diffusion polls before topping up; see its `_next_step`).
 - **Reward/advantage methods are not engine code** — `Part.compute_advantages` and
   `Sample.propagate_rewards` are called by the trainer after scoring. An engine
   fills generation fields such as `segment`, `conditions`, `primitive`, and
diff --git a/unirl/rollout/async_runtime.py b/unirl/rollout/async_runtime.py
deleted file mode 100644
index f0eac5ccc..000000000
--- a/unirl/rollout/async_runtime.py
+++ /dev/null
@@ -1,389 +0,0 @@
-"""Generic driver-side runtime for asynchronous rollout generation.
-
-The runtime is deliberately policy- and trainer-agnostic.  It owns the
-non-blocking Ray dispatch seam, in-flight generation bookkeeping, and the
-versioned buffer of complete rollout groups.  Callers retain responsibility for
-building request Samples, scoring completed Samples, and training on the
-selected groups.
-
-Everything here is single-threaded and lock-free.  A generation is always
-completed before its groups enter :class:`VersionedGroupBuffer`; partial
-trajectory scheduling belongs to a separate, resumable-engine abstraction.
-"""
-
-from __future__ import annotations
-
-import logging
-from dataclasses import dataclass
-from typing import Any, Callable, List, Optional, Protocol
-
-import ray
-
-from unirl.distributed.group.dispatch import DISPATCH_MODE_REGISTRY, Dispatch
-from unirl.distributed.tensor import WorkerLocalTransport
-from unirl.distributed.tensor.pytree import infer_batch_size
-from unirl.types.sample import Sample
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass(frozen=True)
-class BufferedRolloutGroup:
-    """One complete rollout group plus the policy version that produced it."""
-
-    sample: Sample
-    weight_version: int
-    gen_id: int
-
-
-class VersionedGroupBuffer:
-    """Freshness-ordered buffer of complete, tree-preserving rollout groups."""
-
-    def __init__(self) -> None:
-        self._items: List[BufferedRolloutGroup] = []
-
-    def put_all(self, items: List[BufferedRolloutGroup]) -> None:
-        """Append a prepared batch of groups in one mutation."""
-
-        self._items.extend(items)
-
-    def drain_freshest(
-        self,
-        n: int,
-        *,
-        current_version: Optional[int] = None,
-        max_staleness: Optional[int] = None,
-    ) -> Optional[List[BufferedRolloutGroup]]:
-        """Pop the ``n`` freshest eligible groups, carrying leftovers forward.
-
-        Stale groups are evicted first, then remaining groups are sorted by
-        descending generation id.
-
-        Returns ``None`` without consuming eligible groups when fewer than ``n``
-        remain after eviction.
-        """
-
-        if max_staleness is not None and current_version is not None:
-            self._items = [item for item in self._items if current_version - item.weight_version <= max_staleness]
-        if len(self._items) < n:
-            return None
-        self._items.sort(key=lambda item: item.gen_id, reverse=True)
-        picked, self._items = self._items[:n], self._items[n:]
-        return picked
-
-
-@dataclass(frozen=True)
-class InflightGeneration:
-    """One non-blocking distributed ``generate`` invocation."""
-
-    refs: List[Any]
-    worker_local: bool
-    gen_id: int
-    weight_version: int
-
-
-class GenerationDispatcher(Protocol):
-    """Minimal dispatcher contract used by :class:`AsyncRolloutScheduler`."""
-
-    def launch(
-        self,
-        sample: Sample,
-        *,
-        gen_id: int,
-        weight_version: int,
-    ) -> InflightGeneration: ...
-
-    def is_ready(self, job: InflightGeneration) -> bool: ...
-
-    def wait(self, job: InflightGeneration) -> None: ...
-
-    def collect(self, job: InflightGeneration) -> Sample: ...
-
-
-class RayGenerationDispatcher:
-    """Non-blocking ``DP_SCATTER`` dispatcher for a rollout ``Handle``.
-
-    This intentionally mirrors the dispatch/localize/execute and
-    rebind/collect halves of ``distributed/group/handle.py``'s ``handle_fn``.
-    It therefore depends on the Handle's private ``_execute_all`` and
-    ``_rebind_tree`` seams; changes to that implementation must update this
-    adapter in lockstep.
-    """
-
-    def __init__(self, rollout_handle: Any) -> None:
-        self._rollout = rollout_handle
-
-    def launch(
-        self,
-        sample: Sample,
-        *,
-        gen_id: int,
-        weight_version: int,
-    ) -> InflightGeneration:
-        rollout = self._rollout
-        dispatch_fn = DISPATCH_MODE_REGISTRY[Dispatch.DP_SCATTER]["dispatch_fn"]
-        batch_size = infer_batch_size((sample,), {})
-        if batch_size is not None and batch_size % rollout.dp_size != 0:
-            raise ValueError(
-                f"request Sample batch_size={batch_size} not divisible by rollout dp_size={rollout.dp_size}"
-            )
-        shards = dispatch_fn(rollout, (sample,), {}, batch_size)
-        worker_local = issubclass(
-            rollout.pool.transport_cls,
-            WorkerLocalTransport,
-        )
-        shards = rollout.pool.transport_cls.localize(
-            shards,
-            rollout.pool,
-            rollout.device_ids,
-            rollout.worker_ids,
-        )
-        refs = rollout._execute_all(
-            "generate",
-            shards,
-            grad_mode=False,
-            call_id=None,
-        )
-        return InflightGeneration(
-            refs=refs,
-            worker_local=worker_local,
-            gen_id=gen_id,
-            weight_version=weight_version,
-        )
-
-    def is_ready(self, job: InflightGeneration) -> bool:
-        ready, _ = ray.wait(
-            job.refs,
-            num_returns=len(job.refs),
-            timeout=0,
-        )
-        return len(ready) == len(job.refs)
-
-    def wait(self, job: InflightGeneration) -> None:
-        ray.get(job.refs)
-
-    def collect(self, job: InflightGeneration) -> Sample:
-        rollout = self._rollout
-        collect_fn = DISPATCH_MODE_REGISTRY[Dispatch.DP_SCATTER]["collect_fn"]
-        results = ray.get(job.refs)
-        results = [
-            rollout._rebind_tree(
-                result,
-                rollout.workers[index],
-                worker_local=job.worker_local,
-            )
-            for index, result in enumerate(results)
-        ]
-        return collect_fn(rollout, results)
-
-
-BuildSample = Callable[[int], Sample]
-CompleteGeneration = Callable[
-    [InflightGeneration, Sample],
-    List[Sample],
-]
-
-
-class AsyncRolloutScheduler:
-    """Single-threaded scheduler for complete, versioned rollout groups.
-
-    ``reap_before_launch`` picks the phase order inside :meth:`next_step`.
-    Launch-first (the default, used by the AR path) keeps the in-flight window as
-    full as possible. Reap-first instead guarantees that the reap-time work
-    (``collect`` plus the caller's ``on_complete``) runs while the rollout workers
-    hold no other queued generation — required when that work pulls a large payload
-    off those workers, because the transfer would otherwise queue behind a freshly
-    launched generation. It also keeps ``max_inflight=1`` overlapping: the post-reap
-    launch is still made before the step returns, so it runs during the caller's
-    train step.
-    """
-
-    def __init__(
-        self,
-        dispatcher: GenerationDispatcher,
-        *,
-        groups_per_step: int,
-        reap_before_launch: bool = False,
-    ) -> None:
-        if groups_per_step < 1:
-            raise ValueError(f"groups_per_step must be >= 1, got {groups_per_step}")
-        self._dispatcher = dispatcher
-        self._groups_per_step = groups_per_step
-        self._reap_before_launch = bool(reap_before_launch)
-        self._buffer = VersionedGroupBuffer()
-        self._inflight: List[InflightGeneration] = []
-        self._launch_id = 0
-
-    def reset(self, start_id: int = 0) -> None:
-        """Reset empty runtime state for a fresh or resumed trainer loop."""
-
-        if self._inflight:
-            raise RuntimeError("cannot reset AsyncRolloutScheduler with generations in flight")
-        self._buffer = VersionedGroupBuffer()
-        self._launch_id = start_id
-
-    def _launch_one(
-        self,
-        *,
-        build_sample: BuildSample,
-        weight_version: int,
-    ) -> None:
-        gen_id = self._launch_id
-        sample = build_sample(gen_id)
-        self._inflight.append(
-            self._dispatcher.launch(
-                sample,
-                gen_id=gen_id,
-                weight_version=weight_version,
-            )
-        )
-        self._launch_id += 1
-
-    def _top_up(
-        self,
-        *,
-        ceiling: int,
-        max_inflight: int,
-        build_sample: BuildSample,
-        current_version: int,
-    ) -> None:
-        """Launch generations until the launch ceiling or the in-flight cap binds."""
-
-        while self._launch_id < ceiling and len(self._inflight) < max_inflight:
-            self._launch_one(
-                build_sample=build_sample,
-                weight_version=current_version,
-            )
-
-    def _complete(
-        self,
-        job: InflightGeneration,
-        on_complete: CompleteGeneration,
-    ) -> None:
-        # Complete-or-nothing: collect + score first, then a single buffer
-        # mutation. If either step fails the job stays in-flight for retry
-        # without double-inserting groups from a partial put.
-        completed = self._dispatcher.collect(job)
-        groups = on_complete(job, completed)
-        self._buffer.put_all(
-            [
-                BufferedRolloutGroup(
-                    sample=group,
-                    weight_version=job.weight_version,
-                    gen_id=job.gen_id,
-                )
-                for group in groups
-            ]
-        )
-
-    def reap_ready(self, on_complete: CompleteGeneration) -> None:
-        """Collect every ready generation; leave unresolved / failed jobs in flight."""
-
-        still: List[InflightGeneration] = []
-        first_error: Optional[Exception] = None
-        for job in self._inflight:
-            if not self._dispatcher.is_ready(job):
-                still.append(job)
-                continue
-            try:
-                self._complete(job, on_complete)
-            except Exception as exc:
-                # Keep the failed job in-flight so finally/drain_all can retry
-                # collect+score. KeyboardInterrupt/SystemExit propagate immediately
-                # (not deferred behind remaining ready jobs).
-                still.append(job)
-                if first_error is None:
-                    first_error = exc
-                else:
-                    logger.error(
-                        "reap_ready: additional failure for gen_id=%s",
-                        job.gen_id,
-                        exc_info=exc,
-                    )
-        self._inflight = still
-        if first_error is not None:
-            raise first_error
-
-    def drain_all(self, on_complete: CompleteGeneration) -> None:
-        """Quiesce every generation and buffer all successfully completed groups."""
-
-        jobs, self._inflight = list(self._inflight), []
-        first_error: Optional[Exception] = None
-        for job in jobs:
-            try:
-                self._complete(job, on_complete)
-            except Exception as exc:
-                self._inflight.append(job)
-                if first_error is None:
-                    first_error = exc
-                else:
-                    logger.error(
-                        "drain_all: additional failure for gen_id=%s",
-                        job.gen_id,
-                        exc_info=exc,
-                    )
-        if first_error is not None:
-            raise first_error
-
-    def next_step(
-        self,
-        *,
-        rollout_id: int,
-        sync_interval: int,
-        max_inflight: int,
-        max_staleness: int,
-        num_rollouts: int,
-        current_version: int,
-        build_sample: BuildSample,
-        on_complete: CompleteGeneration,
-    ) -> List[BufferedRolloutGroup]:
-        """Return the freshest full training step, blocking only when needed.
-
-        A step is ``groups_per_step`` complete rollout groups. The launch ceiling
-        is the load-bearing on-policy invariant: at ``max_staleness=0`` no
-        generation is launched into a future weight-sync window. Whether each
-        iteration launches or reaps first is fixed by ``reap_before_launch`` (see
-        the class docstring).
-
-        ``sync_interval`` and ``max_inflight`` must already be ``>= 1``; callers
-        (e.g. ``AsyncARTrainer``) clamp config before invoking this method.
-        """
-
-        if sync_interval < 1:
-            raise ValueError(f"sync_interval must be >= 1, got {sync_interval}")
-        if max_inflight < 1:
-            raise ValueError(f"max_inflight must be >= 1, got {max_inflight}")
-        while True:
-            staleness_window = ((rollout_id // sync_interval) + 1 + max_staleness) * sync_interval
-            ceiling = min(num_rollouts, staleness_window)
-            if self._reap_before_launch:
-                self.reap_ready(on_complete)
-            self._top_up(
-                ceiling=ceiling,
-                max_inflight=max_inflight,
-                build_sample=build_sample,
-                current_version=current_version,
-            )
-            if not self._reap_before_launch:
-                self.reap_ready(on_complete)
-            picked = self._buffer.drain_freshest(
-                self._groups_per_step,
-                current_version=current_version,
-                max_staleness=max_staleness,
-            )
-            if picked is not None:
-                return picked
-            if self._inflight:
-                self._dispatcher.wait(self._inflight[0])
-            else:
-                raise RuntimeError("async rollout buffer underflow with no in-flight generations")
-
-
-__all__ = [
-    "AsyncRolloutScheduler",
-    "BufferedRolloutGroup",
-    "GenerationDispatcher",
-    "InflightGeneration",
-    "RayGenerationDispatcher",
-    "VersionedGroupBuffer",
-]
diff --git a/unirl/rollout/engine/__init__.py b/unirl/rollout/engine/__init__.py
index ba880dc59..9e8a3e09f 100644
--- a/unirl/rollout/engine/__init__.py
+++ b/unirl/rollout/engine/__init__.py
@@ -1,56 +1,42 @@
 """Rollout engines over the canonical ``Sample`` request type.
 
-The broad ABC includes coordinator engines; single-turn engines refine it with
-the ``Sample`` → ``Sample`` contract.
+Two halves of one design: ``synchronous.py`` records the worker-side sync contracts
+(``BaseRolloutEngine`` — the broad ABC including coordinator engines — and
+``SyncRolloutEngine``, the ``Sample`` → ``Sample`` refinement the per-backend
+subpackages implement); ``asynchronous.py`` records the driver side (the
+``AsyncRolloutEngine`` protocol, its batch/agentic engines, and their
+mechanisms).
+
+Re-exports are lazy so importing the driver-side module stays ray/torch-free.
 """
 
-from typing import List, Optional
-
-from unirl.rollout.engine.base import BaseRolloutEngine, BaseSingleTurnRolloutEngine
-from unirl.types.sample import Sample
-
-
-def chunked_engine_generate(
-    engine: BaseSingleTurnRolloutEngine,
-    sample: Sample,
-    *,
-    chunk_size: Optional[int],
-) -> Sample:
-    """Call ``engine.generate`` over mini-batch chunks of *sample* and concat outputs.
-
-    Splits the request into per-root-group sub-Samples (:meth:`Sample.split`,
-    tree-complete — each shard holds one prompt's whole subtree across all
-    parts), regroups them into chunks of ``chunk_size`` roots, calls
-    ``engine.generate`` per chunk, and reassembles via :meth:`Sample.concat`
-    (segment rows stay 1:1 with samples, so the merge is a plain per-field
-    concat).
-
-    Fast path (zero overhead): when ``chunk_size`` is ``None`` or ``>=`` the
-    root count, this is a single direct call to ``engine.generate(sample)``.
-
-    Determinism caveat: per-step SDE noise and the x_T recipe inside the engine
-    are independent of chunking (keyed by the sample's path id + step index, not
-    batch position — and split preserves ids), so chunked vs unchunked runs
-    produce bit-identical outputs.
-    """
-    groups = sample.split()
-    n_roots = len(groups)
-    if n_roots == 0:
-        raise ValueError(f"chunked_engine_generate requires a non-empty Sample; got 0 roots (sample={sample!r}).")
-    if chunk_size is None:
-        return engine.generate(sample)
-    if not isinstance(chunk_size, int) or chunk_size < 1:
-        raise ValueError(
-            f"chunk_size must be a positive int when set; got {chunk_size!r} (type={type(chunk_size).__name__})."
-        )
-    if n_roots <= chunk_size:
-        return engine.generate(sample)
-
-    outputs: List[Sample] = []
-    for start in range(0, n_roots, chunk_size):
-        chunk = groups[start : start + chunk_size]
-        outputs.append(engine.generate(Sample.concat(chunk)))
-    return Sample.concat(outputs)
-
-
-__all__ = ["BaseRolloutEngine", "BaseSingleTurnRolloutEngine", "chunked_engine_generate"]
+from __future__ import annotations
+
+import importlib
+from typing import Dict, Tuple
+
+_LAZY_ATTRS: Dict[str, Tuple[str, str]] = {
+    # worker-side sync contracts
+    "BaseRolloutEngine": ("unirl.rollout.engine.synchronous", "BaseRolloutEngine"),
+    "SyncRolloutEngine": ("unirl.rollout.engine.synchronous", "SyncRolloutEngine"),
+    "chunked_engine_generate": ("unirl.rollout.engine.synchronous", "chunked_engine_generate"),
+    # driver-side async engines
+    "AsyncRolloutEngine": ("unirl.rollout.engine.asynchronous", "AsyncRolloutEngine"),
+    "AsyncBatchRolloutEngine": ("unirl.rollout.engine.asynchronous", "AsyncBatchRolloutEngine"),
+    "AsyncAgenticRolloutEngine": ("unirl.rollout.engine.asynchronous", "AsyncAgenticRolloutEngine"),
+}
+
+__all__ = list(_LAZY_ATTRS.keys())
+
+
+def __getattr__(name: str):
+    if name in _LAZY_ATTRS:
+        module_name, attr_name = _LAZY_ATTRS[name]
+        value = getattr(importlib.import_module(module_name), attr_name)
+        globals()[name] = value
+        return value
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__():
+    return sorted(set(globals().keys()) | set(__all__))
diff --git a/unirl/rollout/engine/agentic/config.py b/unirl/rollout/engine/agentic/config.py
index dc349232b..5f758ddd0 100644
--- a/unirl/rollout/engine/agentic/config.py
+++ b/unirl/rollout/engine/agentic/config.py
@@ -18,7 +18,7 @@
 from dataclasses import dataclass
 from typing import Any
 
-from unirl.rollout.engine.base import BaseEngineConfig
+from unirl.rollout.engine.synchronous import BaseEngineConfig
 
 
 @dataclass
diff --git a/unirl/rollout/engine/agentic/engine.py b/unirl/rollout/engine/agentic/engine.py
index 7086f1d46..66a299139 100644
--- a/unirl/rollout/engine/agentic/engine.py
+++ b/unirl/rollout/engine/agentic/engine.py
@@ -9,7 +9,10 @@
 
 **Partial rollout interface (LIN-531).** The engine is the *mechanism*; the trainer owns the *policy*
 (how many to over-sample, staleness, when to sync). The coordinator exposes a
-**submit / poll / finalize / abort** interface over a **background** drain the trainer reaps and interrupts:
+**submit / poll / finalize / abort** interface over a **background** drain the trainer reaps and
+interrupts — consumed driver-side through
+:class:`~unirl.rollout.engine.asynchronous.AsyncAgenticRolloutEngine` (rank-0 unwrap + group assembly +
+versioned buffering):
 
 - ``submit(request)`` — enqueue a pool (fresh prompts and/or carried partials) and fire the drain
   **non-blocking**; the trainer reaps completions with ``poll`` and cuts the tail with ``abort``.
@@ -55,7 +58,7 @@
 from unirl.config.require import require
 from unirl.distributed.group.dispatch import Dispatch, Execute, distributed
 from unirl.rollout.engine.agentic.config import AgenticRolloutEngineConfig
-from unirl.rollout.engine.base import BaseRolloutEngine, BaseSingleTurnRolloutEngine
+from unirl.rollout.engine.synchronous import BaseRolloutEngine, SyncRolloutEngine
 from unirl.types.sample import Sample, _part_with_field
 from unirl.types.sampling import total_samples_per_prompt
 
@@ -98,7 +101,7 @@ def __init__(
         # ``inner.chat_template_kwargs.tools`` in the recipe still wins.
         self._maybe_inject_tool_schemas(config.inner, self._env)
         inner = config.inner.make_engine(strategy=strategy, **deps)
-        if not isinstance(inner, BaseSingleTurnRolloutEngine):
+        if not isinstance(inner, SyncRolloutEngine):
             shutdown = getattr(inner, "shutdown", None)
             if callable(shutdown):
                 try:
@@ -108,7 +111,7 @@ def __init__(
             raise ValueError(
                 f"AgenticRolloutEngine inner must implement the single-turn engine contract; got {type(inner).__name__}"
             )
-        self._inner: BaseSingleTurnRolloutEngine = inner
+        self._inner: SyncRolloutEngine = inner
 
         self._sp = config.episode_sampling  # per-turn sampling params; carries n via samples_per_prompt
         self._n = total_samples_per_prompt(self._sp)  # GRPO group size
@@ -240,9 +243,9 @@ def finalize_if_drained(self) -> Optional[List[Sample]]:
         cleared, and the now-stable completed buffers are drained exactly once. An
         empty ref set is already finalized and returns ``[]`` without polling workers.
 
-        Use this instead of a separate ``drained()`` check followed by ``poll()`` when
-        the next action is ``submit()``: ``submit`` resets per-drive worker buffers, so
-        the readiness check and final reap must remain one coordinator operation.
+        The readiness check and final reap must remain ONE coordinator operation
+        when the next action is ``submit()``: ``submit`` resets per-drive worker
+        buffers, so a separate check-then-poll would race it.
         """
         if not self._drain_refs:
             return []
@@ -304,21 +307,6 @@ def generate(self, sample: Sample) -> List[Sample]:
         self._drain_refs = []
         return self._fan("drain_completed")
 
-    @distributed(dispatch_mode=Dispatch.BROADCAST, execute_mode=Execute.RANK_ZERO)
-    def drained(self) -> bool:
-        """Rank-0 compatibility probe: True once the in-flight drive is finished (queue empty + every
-        trajectory terminal) — i.e. every ``run_drain`` ref is resolved. The async
-        trainer polls this to know a drive is exhausted so it can safely re-``submit``
-        (firing a drain over a still-running one would double-pull the queue).
-
-        This does not join or reap the completed buffers. Call
-        :meth:`finalize_if_drained` for a lossless transition to the next drive.
-        """
-        if not self._drain_refs:
-            return True
-        ready, _ = ray.wait(self._drain_refs, num_returns=len(self._drain_refs), timeout=0)
-        return len(ready) == len(self._drain_refs)
-
     def next_task(self, worker_rank: int) -> Optional[Sample]:
         """Hand out the next trajectory task, or ``None`` when the queue is drained.
 
diff --git a/unirl/rollout/engine/asynchronous.py b/unirl/rollout/engine/asynchronous.py
new file mode 100644
index 000000000..5d7580c6d
--- /dev/null
+++ b/unirl/rollout/engine/asynchronous.py
@@ -0,0 +1,457 @@
+"""Driver-side async rollout engines and their mechanisms (LIN-631).
+
+The async half of the engine design: ``synchronous.py`` records the worker-side sync
+contracts (``BaseRolloutEngine`` / ``SyncRolloutEngine``); this module records
+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
+loops live in the trainers):
+
+- :class:`VersionedBuffer` — payload-agnostic freshness/staleness buffer.
+- :class:`InflightPool` — non-blocking generation pool over one Handle method.
+
+Engines, sharing the :class:`AsyncRolloutEngine` protocol:
+
+- :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.
+- :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)``
+  are stamped at COMPLETION; the per-turn version spread inside a carried
+  trajectory is corrected per-token by each gen Part's own ``weight_version``.
+
+Submission is deliberately engine-specific (incompatible signatures and
+stamping semantics); the protocol is the shared consumer surface the async
+trainers program against. The colocate barrier path (``AgenticTrainer``) keeps
+calling ``rollout.generate(sample)[0]`` directly.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    Generic,
+    Iterable,
+    List,
+    Optional,
+    Protocol,
+    Tuple,
+    TypeVar,
+)
+
+if TYPE_CHECKING:
+    from unirl.types.sample import Sample
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+G = TypeVar("G")
+
+
+# ---------------------------------------------------------------------------
+# Mechanisms
+# ---------------------------------------------------------------------------
+
+
+class VersionedBuffer(Generic[T]):
+    """Payload-agnostic freshness buffer of ``(payload, weight_version, gen_id)`` items.
+
+    Unifies the batch path's per-``Sample`` buffering and the agentic path's
+    per-group (``List[Sample]``) buffering; stamping semantics belong to the
+    caller (batch stamps at launch, agentic at completion).
+    """
+
+    def __init__(self) -> None:
+        self._items: List[Tuple[T, int, int]] = []
+        self._evicted: List[T] = []
+
+    def put(self, payload: T, *, weight_version: int, gen_id: int) -> None:
+        self._items.append((payload, int(weight_version), int(gen_id)))
+
+    def size(self) -> int:
+        return len(self._items)
+
+    def drain_freshest(
+        self,
+        n: int,
+        *,
+        current_version: Optional[int] = None,
+        max_staleness: Optional[int] = None,
+    ) -> Optional[List[T]]:
+        """Pop the ``n`` freshest eligible payloads, carrying leftovers forward.
+
+        Over-stale items are evicted first (retrievable via :meth:`pop_evicted`),
+        then remaining items are sorted by descending ``gen_id`` (stable — ties
+        keep insertion order). Returns ``None`` without consuming anything when
+        fewer than ``n`` remain after eviction.
+        """
+        if max_staleness is not None and current_version is not None:
+            kept: List[Tuple[T, int, int]] = []
+            for item in self._items:
+                if current_version - item[1] <= max_staleness:
+                    kept.append(item)
+                else:
+                    self._evicted.append(item[0])
+            self._items = kept
+        if len(self._items) < n:
+            return None
+        self._items.sort(key=lambda item: item[2], reverse=True)
+        picked, self._items = self._items[:n], self._items[n:]
+        return [payload for payload, _, _ in picked]
+
+    def pop_evicted(self) -> List[T]:
+        """Return and clear payloads rejected by the latest staleness checks."""
+        evicted, self._evicted = self._evicted, []
+        return evicted
+
+
+#: Reap-time completion hook: ``(gen_id, weight_version, completed_payload)``.
+Complete = Callable[[int, int, Any], None]
+
+
+@dataclass(frozen=True)
+class _InflightJob:
+    gen_id: int
+    weight_version: int
+    pending: Any  # PendingHandleCall
+
+
+class InflightPool:
+    """Non-blocking generation pool over one ``@distributed`` Handle method.
+
+    Mechanism only: launch ceilings, 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
+    fallible work must happen before it mutates caller state, because a job
+    whose completion raises stays in flight for retry.
+    """
+
+    def __init__(self, rollout: Any, *, start_gen_id: int = 0, method: str = "generate") -> None:
+        self._rollout = rollout
+        self._method = method
+        self._next_gen_id = int(start_gen_id)
+        self._jobs: List[_InflightJob] = []
+
+    @property
+    def next_gen_id(self) -> int:
+        return self._next_gen_id
+
+    def __len__(self) -> int:
+        return len(self._jobs)
+
+    def launch(self, sample: Any, *, weight_version: int) -> int:
+        gen_id = self._next_gen_id
+        pending = self._rollout.launch_nowait(self._method, sample)
+        self._jobs.append(_InflightJob(gen_id, int(weight_version), pending))
+        self._next_gen_id += 1
+        return gen_id
+
+    def reap_ready(self, complete: Complete) -> int:
+        """Complete every ready job; leave unresolved and failed jobs in flight.
+
+        A job whose ``result()``/``complete`` raises stays in flight for retry;
+        the first error re-raises after the sweep, later errors are logged.
+        KeyboardInterrupt/SystemExit propagate immediately (not deferred behind
+        remaining ready jobs). Returns the number completed.
+        """
+        still: List[_InflightJob] = []
+        first_error: Optional[Exception] = None
+        completed = 0
+        for job in self._jobs:
+            if not job.pending.ready():
+                still.append(job)
+                continue
+            try:
+                complete(job.gen_id, job.weight_version, job.pending.result())
+                completed += 1
+            except Exception as exc:
+                still.append(job)
+                if first_error is None:
+                    first_error = exc
+                else:
+                    logger.error("reap_ready: additional failure for gen_id=%s", job.gen_id, exc_info=exc)
+        self._jobs = still
+        if first_error is not None:
+            raise first_error
+        return completed
+
+    def drain_all(self, complete: Complete) -> int:
+        """Quiesce: complete every job, blocking as needed. Same error contract as
+        :meth:`reap_ready`."""
+        jobs, self._jobs = self._jobs, []
+        first_error: Optional[Exception] = None
+        completed = 0
+        for job in jobs:
+            try:
+                complete(job.gen_id, job.weight_version, job.pending.result())
+                completed += 1
+            except Exception as exc:
+                self._jobs.append(job)
+                if first_error is None:
+                    first_error = exc
+                else:
+                    logger.error("drain_all: additional failure for gen_id=%s", job.gen_id, exc_info=exc)
+        if first_error is not None:
+            raise first_error
+        return completed
+
+    def wait_oldest(self) -> None:
+        """Block until the oldest in-flight generation resolves, without collecting."""
+        if self._jobs:
+            self._jobs[0].pending.wait()
+
+
+# ---------------------------------------------------------------------------
+# The async engine contract
+# ---------------------------------------------------------------------------
+
+
+class AsyncRolloutEngine(Protocol[G]):
+    """Driver-side async rollout engine: version-stamped buffering over a rollout Handle.
+
+    The shared consumer surface; each concrete engine adds its own submission
+    verbs. ``drain_freshest`` uses the engine's own ``weight_version`` as the
+    current version, so trainers only announce syncs via ``bump_weight_version``.
+    """
+
+    @property
+    def weight_version(self) -> int: ...
+
+    def bump_weight_version(self) -> int:
+        """Advance the policy-version counter; call right after ``weight_sync.sync()``."""
+        ...
+
+    def poll(self) -> int:
+        """Non-blocking: move finished work into the buffer; returns the count ingested."""
+        ...
+
+    def drain_freshest(self, n: int, *, max_staleness: int) -> Optional[List[G]]:
+        """Pop the ``n`` freshest groups within ``max_staleness``, or ``None`` if short."""
+        ...
+
+    def pop_evicted(self) -> List[G]:
+        """Return and clear groups rejected by the latest staleness checks."""
+        ...
+
+    def quiesce(self) -> List["Sample"]:
+        """Stop in-flight work and return the carried tail (``[]`` for the batch engine)."""
+        ...
+
+
+# ---------------------------------------------------------------------------
+# Batch engine
+# ---------------------------------------------------------------------------
+
+
+class AsyncBatchRolloutEngine:
+    """``AsyncRolloutEngine[Sample]`` 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.
+
+    ``quiesce()`` (drain everything) is MANDATORY before a weight sync, eval, or
+    checkpoint: a weight + KV update corrupts an in-flight generation.
+    """
+
+    def __init__(
+        self,
+        rollout: Any,
+        *,
+        complete: Callable[[int, "Sample"], List["Sample"]],
+        start_gen_id: int = 0,
+    ) -> None:
+        self._complete = complete
+        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
+
+    @property
+    def next_gen_id(self) -> int:
+        """The gen_id the next ``submit`` will get (1:1 with rollout ids)."""
+        return self._pool.next_gen_id
+
+    @property
+    def inflight(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)
+
+    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"]:
+        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)
+
+
+# ---------------------------------------------------------------------------
+# Agentic engine (driver-side facade over the rank-0 coordinator)
+# ---------------------------------------------------------------------------
+
+
+def root_of(traj: "Sample") -> str:
+    """Root id shared by a prompt's ``n`` sibling trajectories."""
+    return traj.parts[0].sample_ids[0]
+
+
+class PendingGroups:
+    """Bucket a flat stream of terminal trajectories into complete GRPO groups.
+
+    Poll returns variable-depth trajectory ``Sample``s; a prompt's ``n``
+    siblings share its slash-free root id. A group is complete once all ``n``
+    of a root's siblings are terminal. Variable-depth trajectories are NOT
+    concatenated — a group stays a ``List[Sample]`` (the trainer flattens their
+    gen Parts at train time).
+    """
+
+    def __init__(self, n: int) -> None:
+        self._n = int(n)
+        self._by_root: Dict[str, List["Sample"]] = {}
+
+    def add_completed(self, trajs: List["Sample"]) -> None:
+        for t in trajs:
+            self._by_root.setdefault(root_of(t), []).append(t)
+
+    def pop_complete_groups(self) -> List[List["Sample"]]:
+        ready = [root for root, sibs in self._by_root.items() if len(sibs) >= self._n]
+        out: List[List["Sample"]] = []
+        for root in ready:
+            sibs = self._by_root.pop(root)
+            out.append(sibs[: self._n])
+        return out
+
+    def discard_roots(self, roots: Iterable[str]) -> int:
+        """Drop incomplete buckets for abandoned roots; returns the number of
+        terminal sibling trajectories that were being held for them."""
+        discarded = 0
+        for root in set(roots):
+            discarded += len(self._by_root.pop(root, []))
+        return discarded
+
+    def size(self) -> int:
+        return len(self._by_root)
+
+
+class AsyncAgenticRolloutEngine:
+    """``AsyncRolloutEngine[List[Sample]]`` over the ``AgenticRolloutEngine``
+    rank-0 coordinator Handle.
+
+    Normalizes the coordinator's BROADCAST+RANK_ZERO returns (every value
+    unwraps ``[0]``). Groups are stamped at COMPLETION: ``weight_version`` is
+    the engine's counter when a root's last sibling lands, ``gen_id`` a
+    monotonic completed-group counter.
+
+    ``submit`` requires the prior drive to be finalized or quiesced — two live
+    drains would double-pull the coordinator queue.
+    """
+
+    def __init__(self, rollout: Any, *, group_size: int, start_gen_id: int = 0) -> None:
+        self._rollout = rollout
+        self._pending = PendingGroups(group_size)
+        self._buffer: VersionedBuffer[List["Sample"]] = VersionedBuffer()
+        self._gen_id = int(start_gen_id)
+        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
+
+    def submit(self, tasks: List["Sample"]) -> None:
+        """Fire a background drive over a flat task list (fresh siblings + carried partials)."""
+        self._rollout.submit(tasks)
+
+    def poll(self) -> int:
+        return self._ingest(self._rollout.poll()[0])
+
+    def finalize_if_drained(self) -> Optional[int]:
+        """``None`` while the in-flight drive is still running; otherwise join it
+        and ingest its final completions — atomic with the readiness check (a
+        separate poll would race the next ``submit``'s worker-buffer reset)."""
+        completed = self._rollout.finalize_if_drained()[0]
+        if completed is None:
+            return None
+        return self._ingest(completed)
+
+    def drain_freshest(self, n: int, *, max_staleness: int) -> Optional[List[List["Sample"]]]:
+        return self._buffer.drain_freshest(n, current_version=self._weight_version, max_staleness=max_staleness)
+
+    def pop_evicted(self) -> List[List["Sample"]]:
+        return self._buffer.pop_evicted()
+
+    def quiesce(self) -> List["Sample"]:
+        """Turn-boundary stop: abort, then one final poll for trajectories that
+        completed DURING the quiesce (before the next ``submit`` resets worker
+        buffers). Call before ``bump_weight_version`` so those groups carry the
+        version they completed under."""
+        carried = self._rollout.abort()[0]
+        self.poll()
+        return carried
+
+    def discard_roots(self, roots: Iterable[str]) -> int:
+        """Drop abandoned roots' incomplete pending buckets (tail-drop policy)."""
+        return self._pending.discard_roots(roots)
+
+    def pending_groups(self) -> int:
+        """Roots with some-but-not-all siblings terminal (the pending backlog)."""
+        return self._pending.size()
+
+    def buffered_groups(self) -> int:
+        return self._buffer.size()
+
+    def _ingest(self, completed: List["Sample"]) -> int:
+        if completed:
+            self._pending.add_completed(completed)
+            for group in self._pending.pop_complete_groups():
+                self._buffer.put(group, weight_version=self._weight_version, gen_id=self._gen_id)
+                self._gen_id += 1
+        return len(completed)
+
+
+__all__ = [
+    "AsyncAgenticRolloutEngine",
+    "AsyncBatchRolloutEngine",
+    "AsyncRolloutEngine",
+    "Complete",
+    "InflightPool",
+    "PendingGroups",
+    "VersionedBuffer",
+    "root_of",
+]
diff --git a/unirl/rollout/engine/composed/config.py b/unirl/rollout/engine/composed/config.py
index b4991f452..e50c67dce 100644
--- a/unirl/rollout/engine/composed/config.py
+++ b/unirl/rollout/engine/composed/config.py
@@ -17,7 +17,7 @@
 from dataclasses import dataclass
 from typing import Any, Optional
 
-from unirl.rollout.engine.base import BaseEngineConfig
+from unirl.rollout.engine.synchronous import BaseEngineConfig
 
 
 @dataclass
diff --git a/unirl/rollout/engine/composed/engine.py b/unirl/rollout/engine/composed/engine.py
index 60a2ac2f8..d7906a357 100644
--- a/unirl/rollout/engine/composed/engine.py
+++ b/unirl/rollout/engine/composed/engine.py
@@ -31,8 +31,8 @@
 from unirl.config.require import require
 from unirl.distributed.group.dispatch import Dispatch, distributed
 from unirl.models.pe.instruction import postprocess_pe_texts
-from unirl.rollout.engine.base import BaseSingleTurnRolloutEngine
 from unirl.rollout.engine.composed.config import ComposedRolloutEngineConfig
+from unirl.rollout.engine.synchronous import SyncRolloutEngine
 from unirl.types.primitives import Texts
 from unirl.types.sample import Part, Sample
 from unirl.types.sampling import ARSamplingParams, DiffusionSamplingParams
@@ -50,7 +50,7 @@ def _cleanup_constructed_child(name: str, child: Any) -> None:
         logger.warning("Child %r cleanup after construction failure raised: %s", name, exc)
 
 
-class ComposedRolloutEngine(BaseSingleTurnRolloutEngine):
+class ComposedRolloutEngine(SyncRolloutEngine):
     """Two-child rollout engine for prompt-enhancement (PE) serial flow."""
 
     _component_name = "composed"
@@ -83,8 +83,8 @@ def __init__(
         ar = config.ar.make_engine(strategy=None, **deps)
         try:
             require(
-                isinstance(ar, BaseSingleTurnRolloutEngine),
-                f"ComposedRolloutEngine ar child must be a BaseSingleTurnRolloutEngine; got {type(ar).__name__}",
+                isinstance(ar, SyncRolloutEngine),
+                f"ComposedRolloutEngine ar child must be a SyncRolloutEngine; got {type(ar).__name__}",
             )
         except BaseException:
             _cleanup_constructed_child("ar", ar)
@@ -97,8 +97,8 @@ def __init__(
             raise
         try:
             require(
-                isinstance(diffusion, BaseSingleTurnRolloutEngine),
-                "ComposedRolloutEngine diffusion child must be a BaseSingleTurnRolloutEngine; "
+                isinstance(diffusion, SyncRolloutEngine),
+                "ComposedRolloutEngine diffusion child must be a SyncRolloutEngine; "
                 f"got {type(diffusion).__name__}",
             )
         except BaseException:
@@ -109,7 +109,7 @@ def __init__(
         self._ar = ar
         self._diffusion = diffusion
 
-        self._child_by_name: Dict[str, BaseSingleTurnRolloutEngine] = {
+        self._child_by_name: Dict[str, SyncRolloutEngine] = {
             "ar": self._ar,
             "diffusion": self._diffusion,
         }
@@ -396,7 +396,7 @@ def _demux_by_prefix(
                 result[child_name] = subset
         return result
 
-    def _children_for_track_prefix(self, track_prefix: str) -> List[BaseSingleTurnRolloutEngine]:
+    def _children_for_track_prefix(self, track_prefix: str) -> List[SyncRolloutEngine]:
         """Resolve the tensor-payload track routing hint to child engines."""
         if not track_prefix:
             return list(self._child_by_name.values())
diff --git a/unirl/rollout/engine/fastvideo/config.py b/unirl/rollout/engine/fastvideo/config.py
index ba9dc82e3..2c8a04553 100644
--- a/unirl/rollout/engine/fastvideo/config.py
+++ b/unirl/rollout/engine/fastvideo/config.py
@@ -17,8 +17,8 @@
 from omegaconf import SI
 
 from unirl.config.require import require
-from unirl.rollout.engine.base import BaseEngineConfig
 from unirl.rollout.engine.ports import ReservedPorts
+from unirl.rollout.engine.synchronous import BaseEngineConfig
 
 
 @dataclass(frozen=True)
diff --git a/unirl/rollout/engine/fastvideo/engine.py b/unirl/rollout/engine/fastvideo/engine.py
index 5d526d72a..dfe1de185 100644
--- a/unirl/rollout/engine/fastvideo/engine.py
+++ b/unirl/rollout/engine/fastvideo/engine.py
@@ -35,8 +35,8 @@
 
 from unirl.config.require import require
 from unirl.distributed.group.dispatch import Dispatch, distributed
-from unirl.rollout.engine.base import BaseSingleTurnRolloutEngine
 from unirl.rollout.engine.fastvideo.config import FastVideoEngineConfig, FastVideoPorts
+from unirl.rollout.engine.synchronous import SyncRolloutEngine
 from unirl.sde.noise import _derive_group_seed
 from unirl.sde.runtime import FlowMatchSchedulePolicy, ensure_sample_sigmas
 from unirl.types.conditions import TextEmbedCondition
@@ -65,7 +65,7 @@ def _resolve_sde_window(raw_indices: Any, num_steps: int) -> tuple[Optional[List
     return selected, selected
 
 
-class FastVideoRolloutEngine(BaseSingleTurnRolloutEngine):
+class FastVideoRolloutEngine(SyncRolloutEngine):
     """Rollout engine backed by FastVideo ``VideoGenerator`` (RL fork, PR #1222)."""
 
     _component_name = "fastvideo"
diff --git a/unirl/rollout/engine/sglang/backends/native.py b/unirl/rollout/engine/sglang/backends/native.py
index c30bf3dfe..aa182469d 100644
--- a/unirl/rollout/engine/sglang/backends/native.py
+++ b/unirl/rollout/engine/sglang/backends/native.py
@@ -363,7 +363,7 @@ async def _agen_one(self, payload: Dict[str, Any]) -> List[Any]:
         kwargs = payload_to_generate_kwargs(payload)
         async with self._sem:
             try:
-                response = await self._engine.async_generate(**kwargs)
+                response = await self._engine.asynchronousgenerate(**kwargs)
             except Exception as exc:
                 raise RuntimeError(f"sglang NativeBackend.generate failed: {exc}") from exc
         parsed = parse_generate_response(response)
diff --git a/unirl/rollout/engine/sglang/config.py b/unirl/rollout/engine/sglang/config.py
index ca3f62a8e..6284c11b6 100644
--- a/unirl/rollout/engine/sglang/config.py
+++ b/unirl/rollout/engine/sglang/config.py
@@ -19,8 +19,8 @@
 from typing import Any, Dict, Optional
 
 from unirl.config.require import require
-from unirl.rollout.engine.base import BaseEngineConfig
 from unirl.rollout.engine.ports import ReservedPorts
+from unirl.rollout.engine.synchronous import BaseEngineConfig
 
 _SGLANG_GRPC_PORT_OFFSET = 30000
 _SGLANG_MAX_DERIVED_GRPC_BASE_PORT = 65535 - _SGLANG_GRPC_PORT_OFFSET
diff --git a/unirl/rollout/engine/sglang/engine.py b/unirl/rollout/engine/sglang/engine.py
index cc9dc35cc..b2d1bcd37 100644
--- a/unirl/rollout/engine/sglang/engine.py
+++ b/unirl/rollout/engine/sglang/engine.py
@@ -6,7 +6,7 @@
 owns the SRT runtime — server subprocess + HTTP, or the in-process Engine,
 picked by ``config.backend``). Weight sync is a :class:`WeightSync` component
 constructed over the seam; the offload lifecycle (the two staged flags) lives
-directly on the engine. The frozen ``base.py`` surface is implemented as thin
+directly on the engine. The frozen ``synchronous.py`` surface is implemented as thin
 forwards here — they must be real class attributes anyway (``Worker.call``
 dispatches by name; ``@distributed`` binds the most-derived attribute) — which
 also absorbs the surface quirks (``track_prefix``) so the component keeps clean
@@ -14,7 +14,7 @@
 
 One-shot construction: after ``__init__`` returns, the SRT server is spawned and
 healthy and the engine is usable. ``generate`` / ``sleep`` / ``wake_up``
-re-apply ``@distributed`` (the decorator is not inherited — see ``base.py``).
+re-apply ``@distributed`` (the decorator is not inherited — see ``synchronous.py``).
 No environment mutation happens here — the spawn-scoped env the SRT
 subprocesses need is quarantined in the backends' ``boot``.
 """
@@ -28,18 +28,18 @@
 
 from unirl.config.require import require
 from unirl.distributed.group.dispatch import Dispatch, distributed
-from unirl.rollout.engine.base import BaseSingleTurnRolloutEngine
 from unirl.rollout.engine.sglang.adapters import get_adapter
 from unirl.rollout.engine.sglang.backends import HTTPBackend, NativeBackend
 from unirl.rollout.engine.sglang.config import SGLangEngineConfig, SGLangPorts
 from unirl.rollout.engine.sglang.utils import resolve_sampling
 from unirl.rollout.engine.sglang.weight_sync import WeightSync
+from unirl.rollout.engine.synchronous import SyncRolloutEngine
 from unirl.types.sample import Sample
 
 logger = logging.getLogger(__name__)
 
 
-class SGLangRolloutEngine(BaseSingleTurnRolloutEngine):
+class SGLangRolloutEngine(SyncRolloutEngine):
     """LLM/VLM rollout engine backed by a SGLang SRT server (v2 layout)."""
 
     _component_name = "sglang"
@@ -192,7 +192,7 @@ def resume(self) -> None:
 
     # ------------------------------------------------------------------ #
     # Lifecycle — the offload flags live here; decorators re-applied
-    # (base.py footgun)
+    # (synchronous.py footgun)
     # ------------------------------------------------------------------ #
 
     @distributed(dispatch_mode=Dispatch.BROADCAST)
@@ -273,7 +273,7 @@ def __del__(self):
             pass
 
     # ------------------------------------------------------------------ #
-    # Weight sync — frozen base.py surface; thin forwards to the component.
+    # Weight sync — frozen synchronous.py surface; thin forwards to the component.
     # Un-decorated: reached per worker via the raw ``Worker.call`` RPC, not
     # through ``@distributed``. ``track_prefix`` is absorbed here.
     # ------------------------------------------------------------------ #
diff --git a/unirl/rollout/engine/sglang/weight_sync.py b/unirl/rollout/engine/sglang/weight_sync.py
index 0baee9661..0265cd093 100644
--- a/unirl/rollout/engine/sglang/weight_sync.py
+++ b/unirl/rollout/engine/sglang/weight_sync.py
@@ -3,7 +3,7 @@
 ``WeightSync`` is a plain object the engine constructs over the seam: it takes
 the backend explicitly and owns all sync/LoRA state (``_lora_version`` /
 ``_lora_loaded`` / ``_active_adapter``). Method names mirror the frozen
-``base.py`` surface minus ``track_prefix`` (the engine's forwards absorb that,
+``synchronous.py`` surface minus ``track_prefix`` (the engine's forwards absorb that,
 along with the per-worker ``Worker.call`` dispatch concern), so a grep for a
 trainer-side entry point lands here.
 
diff --git a/unirl/rollout/engine/sglang_diffusion/config.py b/unirl/rollout/engine/sglang_diffusion/config.py
index 8608022ab..66b028ae7 100644
--- a/unirl/rollout/engine/sglang_diffusion/config.py
+++ b/unirl/rollout/engine/sglang_diffusion/config.py
@@ -21,8 +21,8 @@
 from omegaconf import SI
 
 from unirl.config.require import require
-from unirl.rollout.engine.base import BaseEngineConfig
 from unirl.rollout.engine.ports import ReservedPorts
+from unirl.rollout.engine.synchronous import BaseEngineConfig
 
 
 @dataclass(frozen=True)
diff --git a/unirl/rollout/engine/sglang_diffusion/engine.py b/unirl/rollout/engine/sglang_diffusion/engine.py
index 34b7efdf5..06f540a3f 100644
--- a/unirl/rollout/engine/sglang_diffusion/engine.py
+++ b/unirl/rollout/engine/sglang_diffusion/engine.py
@@ -4,7 +4,7 @@
 from the registry by ``config.model_family``, owns the ``Sample`` → ``Sample``
 conversion) and no concrete backend (the seam owns the runtime). Weight sync is a
 :class:`WeightSync` component constructed over the seam; the offload lifecycle (a
-single flag) lives directly on the engine. The frozen ``base.py`` surface is
+single flag) lives directly on the engine. The frozen ``synchronous.py`` surface is
 implemented as thin forwards here — they must be real class attributes anyway
 (``Worker.call`` dispatches by name; ``@distributed`` binds the most-derived
 attribute) — which also absorbs the surface quirks (``track_prefix``) so the
@@ -12,7 +12,7 @@
 
 One-shot construction: after ``__init__`` returns, the generator is spawned and the
 engine is usable. ``generate`` / ``sleep`` / ``wake_up`` re-apply ``@distributed``
-(the decorator is not inherited — see ``base.py``).
+(the decorator is not inherited — see ``synchronous.py``).
 """
 
 from __future__ import annotations
@@ -25,7 +25,6 @@
 
 from unirl.config.require import require
 from unirl.distributed.group.dispatch import Dispatch, distributed
-from unirl.rollout.engine.base import BaseSingleTurnRolloutEngine
 from unirl.rollout.engine.sglang_diffusion.adapters import get_adapter
 from unirl.rollout.engine.sglang_diffusion.backends import SGLangBackend
 from unirl.rollout.engine.sglang_diffusion.config import (
@@ -33,6 +32,7 @@
     SGLangDiffusionPorts,
 )
 from unirl.rollout.engine.sglang_diffusion.weight_sync import WeightSync
+from unirl.rollout.engine.synchronous import SyncRolloutEngine
 from unirl.sde.noise import generate_latents
 from unirl.sde.runtime import ensure_sample_sigmas
 from unirl.types.noise_recipe import NoiseRecipe
@@ -48,7 +48,7 @@
 _CPU_BACKUP_TAGS = ("vae", "text_encoder")
 
 
-class SGLangDiffusionRolloutEngine(BaseSingleTurnRolloutEngine):
+class SGLangDiffusionRolloutEngine(SyncRolloutEngine):
     """Rollout engine backed by ``sglang.multimodal_gen.DiffGenerator`` (v2 layout)."""
 
     _component_name = "sglang_diffusion"
@@ -227,7 +227,7 @@ def _resolve_initial_noise(self, sample: Sample) -> Optional[torch.Tensor]:
         )
 
     # ------------------------------------------------------------------ #
-    # Lifecycle — the offload flag lives here; decorators re-applied (base.py footgun)
+    # Lifecycle — the offload flag lives here; decorators re-applied (synchronous.py footgun)
     # ------------------------------------------------------------------ #
 
     @distributed(dispatch_mode=Dispatch.BROADCAST)
@@ -274,7 +274,7 @@ def shutdown(self) -> None:
             self._shutdown_complete = True
 
     # ------------------------------------------------------------------ #
-    # Weight sync — frozen base.py surface; thin forwards to the component.
+    # Weight sync — frozen synchronous.py surface; thin forwards to the component.
     # Un-decorated: reached per worker via the raw ``Worker.call`` RPC, not
     # through ``@distributed``. ``track_prefix`` is absorbed here.
     # ------------------------------------------------------------------ #
diff --git a/unirl/rollout/engine/sglang_diffusion/weight_sync.py b/unirl/rollout/engine/sglang_diffusion/weight_sync.py
index d736315f3..31c02ee1b 100644
--- a/unirl/rollout/engine/sglang_diffusion/weight_sync.py
+++ b/unirl/rollout/engine/sglang_diffusion/weight_sync.py
@@ -3,7 +3,7 @@
 ``WeightSync`` is a plain object the engine constructs over the seam: it takes the
 backend and the LoRA spec explicitly and owns all sync/LoRA state
 (``_lora_loaded`` / ``_active_adapter``). Method names mirror
-the frozen ``base.py`` surface minus ``track_prefix`` (the engine's forwards absorb
+the frozen ``synchronous.py`` surface minus ``track_prefix`` (the engine's forwards absorb
 that, along with the per-worker ``Worker.call`` dispatch concern), so a grep for a
 trainer-side entry point lands here.
 
diff --git a/unirl/rollout/engine/base.py b/unirl/rollout/engine/synchronous.py
similarity index 51%
rename from unirl/rollout/engine/base.py
rename to unirl/rollout/engine/synchronous.py
index ed42c20b8..01657a673 100644
--- a/unirl/rollout/engine/base.py
+++ b/unirl/rollout/engine/synchronous.py
@@ -1,11 +1,4 @@
-"""Rollout engine base class for the ``Sample`` → ``Sample`` path.
-
-Concrete engines take all runtime deps as ``__init__`` kwargs and complete
-construction in one shot — no separate ``initialize(device)`` step. After
-``__init__`` returns the engine is fully usable: model loaded, worker
-subprocesses spawned, dist groups brought up. This matches the actor flow where
-``_setup_distributed_env`` runs before the engine is built.
-"""
+"""Rollout engine base classes. Engines complete construction in ``__init__``; no separate initialize step."""
 
 from __future__ import annotations
 
@@ -18,60 +11,35 @@
 from unirl.distributed.group.remote import Remote
 from unirl.types.sample import Sample
 
-#: The rollout batch contract (LIN-522). Single-turn engines return one ``Sample``;
-#: the agentic engine returns a list of variable-depth trajectory ``Sample``s (one
-#: per rollout). The broad generation method uses this union; the single-turn
-#: subclass refines its return to ``Sample``.
+# Single-turn engines return one Sample; the agentic engine returns a trajectory list.
 RolloutOutput = Union[Sample, List[Sample]]
 
 
 class BaseEngineConfig(ABC):
-    """Marker base for all rollout engine config dataclasses.
-
-    Used as the type annotation / base class for the engine config dataclasses.
-    Each concrete engine config maps itself to its runtime engine class via
-    :meth:`make_engine`.
-    """
+    """Marker base for rollout engine config dataclasses."""
 
     def make_engine(self, **deps: Any) -> "BaseRolloutEngine":
-        """Construct the runtime engine declared by this config.
-
-        ``deps`` carry the runtime injections (``device``, ``strategy``,
-        ``rank``, ``model_config``); the engine ctor contract is uniformly
-        ``Engine(config=self, **deps)``. Subclasses override to import (lazily,
-        so config modules stay importable without the engine's heavy optional
-        deps) and return their engine class.
-        """
+        """Construct the runtime engine declared by this config; ctor contract is ``Engine(config=self, **deps)``."""
         raise NotImplementedError(f"{type(self).__name__} must implement make_engine()")
 
 
 class BaseRolloutEngine(Remote, ABC):
-    """Rollout engine ABC. One-shot construction; new types only."""
+    """Rollout engine ABC."""
 
-    # ------------------------------------------------------------------
     # Lifecycle
-    # ------------------------------------------------------------------
 
     @abstractmethod
     def shutdown(self) -> None:
         """Release worker subprocesses and any other engine-owned resources."""
 
+    # Overrides of sleep/wake_up must re-apply @distributed; Handle binds the subclass attribute only.
     @distributed(dispatch_mode=Dispatch.BROADCAST)
     def sleep(self) -> None:
-        """Best-effort runtime offload. Default no-op.
-
-        Decorated so the driver-side ``Handle.sleep()`` dispatches to every
-        worker. Subclasses that override should re-apply ``@distributed``
-        on their override (Handle's method-binding sees the subclass's
-        attribute and won't pick up a base-class decorator alone).
-        """
+        """Best-effort runtime offload. Default no-op."""
 
     @distributed(dispatch_mode=Dispatch.BROADCAST)
     def wake_up(self) -> None:
-        """Restore runtime resources after ``sleep``. Default no-op.
-
-        Same dispatch contract as :meth:`sleep`; see its docstring.
-        """
+        """Restore runtime resources after ``sleep``. Default no-op."""
 
     def onload_weights(self, *, track_prefix: str = "") -> None:
         """Restore the resources needed to receive a weight update."""
@@ -88,7 +56,7 @@ def health_check(self) -> bool:
         return True
 
     def get_memory_info(self) -> Dict[str, float]:
-        """Per-engine GPU memory snapshot. Default reads CUDA totals."""
+        """Per-engine GPU memory snapshot."""
         if not torch.cuda.is_available():
             return {}
         return {
@@ -96,31 +64,16 @@ def get_memory_info(self) -> Dict[str, float]:
             "cached_gb": torch.cuda.memory_reserved() / 1e9,
         }
 
-    # ------------------------------------------------------------------
     # Generation
-    # ------------------------------------------------------------------
 
     @abstractmethod
     def generate(self, sample: Sample) -> RolloutOutput:
-        """Synchronously run rollout generation for one request batch.
+        """Synchronously run rollout generation; each concrete contract owns its dispatch mode."""
 
-        Dispatch belongs to each concrete contract: single-turn engines use
-        ``DP_SCATTER`` and return one ``Sample``; the agentic coordinator uses
-        ``BROADCAST + RANK_ZERO`` and returns a trajectory list.
-        """
-
-    # ------------------------------------------------------------------
-    # Control plane — sync methods reached via the raw ``Worker.call`` RPC (the
-    # un-decorated weight-sync pattern), so they interleave with an in-flight
-    # ``generate`` on a threaded Worker (``worker_max_concurrency>1``).
-    # ------------------------------------------------------------------
+    # Control plane — reached via raw Worker.call, so calls interleave with an in-flight generate.
 
     def abort(self, ids: Optional[List[str]] = None) -> List[Sample]:
-        """Best-effort cancel of in-flight generation; return any partials.
-
-        Default no-op (``[]``). Engines whose backend supports it (SGLang) cancel
-        running requests; sync/batch backends can only drop not-yet-started work.
-        """
+        """Best-effort cancel of in-flight generation; return any partials. Default no-op."""
         del ids
         return []
 
@@ -130,9 +83,7 @@ def pause(self) -> None:
     def resume(self) -> None:
         """Resume generation after :meth:`pause`. Default no-op."""
 
-    # ------------------------------------------------------------------
-    # Weight sync — bucketed CUDA-IPC (verl-omni pattern)
-    # ------------------------------------------------------------------
+    # Weight sync — bucketed CUDA-IPC
 
     def update_weights_from_ipc(
         self,
@@ -145,9 +96,7 @@ def update_weights_from_ipc(
         """Receive a state dict over a per-rank ZMQ + CUDA-IPC channel."""
         raise NotImplementedError
 
-    # ------------------------------------------------------------------
     # Weight sync — NCCL broadcast
-    # ------------------------------------------------------------------
 
     def init_weights_update_group(
         self,
@@ -186,9 +135,7 @@ def destroy_weights_update_group(
         """Tear down a previously-initialized NCCL update group."""
         raise NotImplementedError
 
-    # ------------------------------------------------------------------
     # Weight sync — LoRA tensor bag
-    # ------------------------------------------------------------------
 
     def set_lora_from_tensors(
         self,
@@ -200,9 +147,7 @@ def set_lora_from_tensors(
         """Load a LoRA adapter directly from in-memory tensors."""
         raise NotImplementedError
 
-    # ------------------------------------------------------------------
     # Weight sync — SGLang-shape one-bag tensor payload
-    # ------------------------------------------------------------------
 
     def update_weights_from_tensor(
         self,
@@ -218,20 +163,10 @@ def update_weights_from_tensor(
         raise NotImplementedError
 
 
-class BaseSingleTurnRolloutEngine(BaseRolloutEngine, ABC):
-    """Nominal contract for engines that fill and return one ``Sample``.
+class SyncRolloutEngine(BaseRolloutEngine, ABC):
+    """Engines that fill and return one ``Sample``; ``generate`` may be called concurrently (agentic drain)."""
 
-    The class intentionally does not prescribe batching or provide a batching
-    wrapper. Each engine owns those semantics and applies its own
-    ``@distributed`` decorator. The contract is synchronous. An engine meant to
-    serve as an agentic inner must make ``generate`` safe for CONCURRENT
-    callers (the agentic drain calls it from one thread per trajectory and
-    relies on the backend batching the in-flight requests together); an engine
-    that cannot serve concurrently serializes internally instead.
-    """
-
-    #: Policy weight version the current weights correspond to (bumped on each
-    #: weight sync; stamped onto generated Parts by :meth:`_stamp_weight_version`).
+    # Policy weight version of the current weights; bumped on each weight sync.
     _weight_version: int = 0
 
     @abstractmethod
@@ -247,4 +182,47 @@ def _stamp_weight_version(self, sample: Sample) -> Sample:
         return sample.with_parts([*sample.parts[:-1], gen])
 
 
-__all__ = ["BaseRolloutEngine", "BaseSingleTurnRolloutEngine", "RolloutOutput"]
+def chunked_engine_generate(
+    engine: SyncRolloutEngine,
+    sample: Sample,
+    *,
+    chunk_size: Optional[int],
+) -> Sample:
+    """Call ``engine.generate`` over mini-batch chunks of *sample* and concat outputs.
+
+    Splits the request into per-root-group sub-Samples (:meth:`Sample.split`,
+    tree-complete — each shard holds one prompt's whole subtree across all
+    parts), regroups them into chunks of ``chunk_size`` roots, calls
+    ``engine.generate`` per chunk, and reassembles via :meth:`Sample.concat`
+    (segment rows stay 1:1 with samples, so the merge is a plain per-field
+    concat).
+
+    Fast path (zero overhead): when ``chunk_size`` is ``None`` or ``>=`` the
+    root count, this is a single direct call to ``engine.generate(sample)``.
+
+    Determinism caveat: per-step SDE noise and the x_T recipe inside the engine
+    are independent of chunking (keyed by the sample's path id + step index, not
+    batch position — and split preserves ids), so chunked vs unchunked runs
+    produce bit-identical outputs.
+    """
+    groups = sample.split()
+    n_roots = len(groups)
+    if n_roots == 0:
+        raise ValueError(f"chunked_engine_generate requires a non-empty Sample; got 0 roots (sample={sample!r}).")
+    if chunk_size is None:
+        return engine.generate(sample)
+    if not isinstance(chunk_size, int) or chunk_size < 1:
+        raise ValueError(
+            f"chunk_size must be a positive int when set; got {chunk_size!r} (type={type(chunk_size).__name__})."
+        )
+    if n_roots <= chunk_size:
+        return engine.generate(sample)
+
+    outputs: List[Sample] = []
+    for start in range(0, n_roots, chunk_size):
+        chunk = groups[start : start + chunk_size]
+        outputs.append(engine.generate(Sample.concat(chunk)))
+    return Sample.concat(outputs)
+
+
+__all__ = ["BaseRolloutEngine", "SyncRolloutEngine", "RolloutOutput", "chunked_engine_generate"]
diff --git a/unirl/rollout/engine/trainside/__init__.py b/unirl/rollout/engine/trainside/__init__.py
index 3ddc1d220..18816b762 100644
--- a/unirl/rollout/engine/trainside/__init__.py
+++ b/unirl/rollout/engine/trainside/__init__.py
@@ -1,7 +1,7 @@
 """In-process rollout engine adapter for direct-sampling mode.
 
 Exposes a materialized ``models`` ``Pipeline`` as a
-:class:`unirl.rollout.engine.base.BaseRolloutEngine`. Used when the
+:class:`unirl.rollout.engine.synchronous.BaseRolloutEngine`. Used when the
 training Policy itself is the sampler (direct sampling, on-policy RL) —
 the rollout runs in the same Ray actor / Python process / GPU as
 training, so no worker subprocess and no weight sync are needed.
diff --git a/unirl/rollout/engine/trainside/config.py b/unirl/rollout/engine/trainside/config.py
index fa8de060f..5fd04f4c7 100644
--- a/unirl/rollout/engine/trainside/config.py
+++ b/unirl/rollout/engine/trainside/config.py
@@ -13,7 +13,7 @@
 
 from dataclasses import dataclass
 
-from unirl.rollout.engine.base import BaseEngineConfig
+from unirl.rollout.engine.synchronous import BaseEngineConfig
 
 
 @dataclass
diff --git a/unirl/rollout/engine/trainside/engine.py b/unirl/rollout/engine/trainside/engine.py
index d879a3c3e..3e5096ca6 100644
--- a/unirl/rollout/engine/trainside/engine.py
+++ b/unirl/rollout/engine/trainside/engine.py
@@ -18,14 +18,14 @@
 from unirl.models.types.ar import ARStage
 from unirl.models.types.diffusion import DiffusionStage
 from unirl.models.types.pipeline import Pipeline
-from unirl.rollout.engine.base import BaseSingleTurnRolloutEngine
+from unirl.rollout.engine.synchronous import SyncRolloutEngine
 from unirl.sde.runtime import FlowMatchSchedulePolicy, ensure_sample_sigmas
 from unirl.types.sample import Part, Sample
 
 Stage = Union[DiffusionStage, ARStage]
 
 
-class TrainsideRolloutEngine(BaseSingleTurnRolloutEngine):
+class TrainsideRolloutEngine(SyncRolloutEngine):
     """In-process rollout engine: the train actor's Pipeline IS the sampler.
 
     Args:
diff --git a/unirl/rollout/engine/vllm_omni/__init__.py b/unirl/rollout/engine/vllm_omni/__init__.py
index 6347c13c9..7f1c7b1e3 100644
--- a/unirl/rollout/engine/vllm_omni/__init__.py
+++ b/unirl/rollout/engine/vllm_omni/__init__.py
@@ -8,7 +8,7 @@
 (``worker/`` / ``pipelines/`` / ``patches/``). Recipes select it by pointing
 their rollout ``_target_`` lines here.
 
-Imports are lazy: engine modules pull ``rollout.engine.base`` whose import
+Imports are lazy: engine modules pull ``rollout.engine.synchronous`` whose import
 chain is still initializing when reached from ``base → types → distributed``.
 """
 
diff --git a/unirl/rollout/engine/vllm_omni/config.py b/unirl/rollout/engine/vllm_omni/config.py
index 8f681164b..38ce869ec 100644
--- a/unirl/rollout/engine/vllm_omni/config.py
+++ b/unirl/rollout/engine/vllm_omni/config.py
@@ -25,8 +25,8 @@
 from omegaconf import MISSING
 
 from unirl.config.require import require
-from unirl.rollout.engine.base import BaseEngineConfig
 from unirl.rollout.engine.ports import ReservedPorts
+from unirl.rollout.engine.synchronous import BaseEngineConfig
 
 
 @dataclass(frozen=True)
diff --git a/unirl/rollout/engine/vllm_omni/engine.py b/unirl/rollout/engine/vllm_omni/engine.py
index af17df7a6..eb8030218 100644
--- a/unirl/rollout/engine/vllm_omni/engine.py
+++ b/unirl/rollout/engine/vllm_omni/engine.py
@@ -6,7 +6,7 @@
 and no concrete backend (the seam owns the runtime — boot, ports, env quirks,
 the per-stage ``collective_rpc`` fan-out). Weight sync is a :class:`WeightSync`
 component constructed over the seam; the offload lifecycle (a single flag)
-lives directly on the engine. The frozen ``base.py`` surface is implemented as
+lives directly on the engine. The frozen ``synchronous.py`` surface is implemented as
 thin forwards here — they must be real class attributes anyway (``Worker.call``
 dispatches by name; ``@distributed`` binds the most-derived attribute) — which
 also absorbs the surface quirks (``track_prefix``) so the component keeps
@@ -14,7 +14,7 @@
 
 One-shot construction: after ``__init__`` returns, the ``Omni`` orchestrator
 is spawned and the engine is usable. ``generate`` / ``sleep`` / ``wake_up``
-re-apply ``@distributed`` (the decorator is not inherited — see ``base.py``).
+re-apply ``@distributed`` (the decorator is not inherited — see ``synchronous.py``).
 ``set_lora_from_tensors_copy`` additionally keeps v1's ``@distributed(BROADCAST)``
 — the documented exception to the "weight-sync entry points undecorated" rule:
 it is how the HI3 two-engine LoRA sync reaches engines anchored on disjoint
@@ -31,7 +31,7 @@
 
 from unirl.config.require import require
 from unirl.distributed.group.dispatch import Dispatch, distributed
-from unirl.rollout.engine.base import BaseSingleTurnRolloutEngine
+from unirl.rollout.engine.synchronous import SyncRolloutEngine
 from unirl.rollout.engine.vllm_omni.adapters import get_adapter
 from unirl.rollout.engine.vllm_omni.backends import VLLMOmniBackend
 from unirl.rollout.engine.vllm_omni.config import VLLMOmniEngineConfig, VLLMOmniPorts
@@ -42,7 +42,7 @@
 logger = logging.getLogger(__name__)
 
 
-class VLLMOmniRolloutEngine(BaseSingleTurnRolloutEngine):
+class VLLMOmniRolloutEngine(SyncRolloutEngine):
     """Rollout engine backed by vllm-omni's ``Omni`` orchestrator (v2 layout)."""
 
     _component_name = "vllm_omni"
@@ -256,7 +256,7 @@ def tp_per_stage(self) -> Dict[int, int]:
         return self._backend.tp_per_stage()
 
     # ------------------------------------------------------------------ #
-    # Weight sync — frozen base.py surface; thin forwards to the component.
+    # Weight sync — frozen synchronous.py surface; thin forwards to the component.
     # Un-decorated (except the documented copy-variant): reached per worker
     # via the raw ``Worker.call`` RPC, not through ``@distributed``.
     # ``track_prefix`` is absorbed here.
diff --git a/unirl/rollout/engine/vllm_omni/patches/runtime.py b/unirl/rollout/engine/vllm_omni/patches/runtime.py
index 553dc6925..eb2941045 100644
--- a/unirl/rollout/engine/vllm_omni/patches/runtime.py
+++ b/unirl/rollout/engine/vllm_omni/patches/runtime.py
@@ -473,7 +473,7 @@ def patch_lora_request_passthrough() -> None:
     Replaces pod-local file patch on ``vllm_omni/entrypoints/omni.py``.
     """
     try:
-        from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
+        from vllm_omni.engine.asynchronousomni_engine import AsyncOmniEngine
         from vllm_omni.entrypoints.omni import Omni
     except (ImportError, AttributeError):
         return  # vllm-omni not available in this process; skip
@@ -578,7 +578,7 @@ def patch_per_request_ar_seed() -> None:
     try:
         import msgspec as _msgspec
         from vllm import SamplingParams as VLLMSamplingParams
-        from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
+        from vllm_omni.engine.asynchronousomni_engine import AsyncOmniEngine
     except (ImportError, AttributeError):
         return
 
@@ -632,7 +632,7 @@ def patch_master_port_unstrip() -> None:
     in ``docs/vllm-omni-v2-engine.md``).
     """
     try:
-        from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
+        from vllm_omni.engine.asynchronousomni_engine import AsyncOmniEngine
 
         _orig = AsyncOmniEngine._strip_single_engine_args
         if getattr(_orig, "_diffrl_master_port_unstrip", False):
diff --git a/unirl/rollout/engine/vllm_omni/weight_sync.py b/unirl/rollout/engine/vllm_omni/weight_sync.py
index 97cad1181..fe24a7b6f 100644
--- a/unirl/rollout/engine/vllm_omni/weight_sync.py
+++ b/unirl/rollout/engine/vllm_omni/weight_sync.py
@@ -3,7 +3,7 @@
 ``WeightSync`` is a plain object the engine constructs over the seam: it takes
 the backend and the LoRA transport choice explicitly and owns ALL sync/LoRA
 state (``_lora_loaded`` / ``_weights_released`` / ``_last_lora_*``). Method
-names mirror the frozen ``base.py`` surface minus ``track_prefix`` (the
+names mirror the frozen ``synchronous.py`` surface minus ``track_prefix`` (the
 engine's forwards absorb that), so a grep for a trainer-side entry point lands
 here. The transports declared are exactly what vllm-omni supports: bucketed
 CUDA-IPC, NCCL (init/transfer/destroy), the SGLang-shape tensor bag, and the
diff --git a/unirl/rollout/loop/README.md b/unirl/rollout/loop/README.md
index b4ccf9aa1..960948223 100644
--- a/unirl/rollout/loop/README.md
+++ b/unirl/rollout/loop/README.md
@@ -8,7 +8,7 @@ agentic rollout. The distributed runtime is
 ## Contracts
 
 - `RolloutEnginePort.generate(sample) -> Sample` fills one generation frontier. The production
-  agentic engine requires its inner engine to implement `BaseSingleTurnRolloutEngine`.
+  agentic engine requires its inner engine to implement `SyncRolloutEngine`.
 - `Environment.reset(request) -> Sample` performs per-trajectory setup and may augment or replace
   the request.
 - `Environment.step(sample) -> (observation, done, info)` consumes the latest generated action.
diff --git a/unirl/rollout/loop/engine_port.py b/unirl/rollout/loop/engine_port.py
index f0c628de1..6caf42385 100644
--- a/unirl/rollout/loop/engine_port.py
+++ b/unirl/rollout/loop/engine_port.py
@@ -1,7 +1,7 @@
 """RolloutEnginePort — the generation seam the agent loop calls (LIN-492).
 
 See ``unirl/rollout/loop/README.md``. A structural ``Protocol`` for a single-turn
-engine; ``BaseSingleTurnRolloutEngine`` is its nominal runtime counterpart.
+engine; ``SyncRolloutEngine`` is its nominal runtime counterpart.
 """
 
 from __future__ import annotations
diff --git a/unirl/trainer/README.md b/unirl/trainer/README.md
index e7c107ef6..66f782da7 100644
--- a/unirl/trainer/README.md
+++ b/unirl/trainer/README.md
@@ -68,6 +68,13 @@ The current trainer surface is:
 | `AgenticPartialTrainer` / `AgenticEnvPartialTrainer` | freshest complete trajectory groups → concatenated turn `Part` | Colocated over-sample/commit/abort loop. `carry` is for Sample-resumable stateless tools; `drop` purges tails from stateful environments that restart episodes. |
 | `AsyncAgenticTrainer` / `AsyncAgenticEnvTrainer` | buffered complete trajectory groups → concatenated turn `Part` | Disaggregated train/rollout slabs, resident agentic drive, weight-version staleness control, and the same explicit `carry`/`drop` tail policy. |
 
+The async variants program against the driver-side `AsyncRolloutEngine` protocol in
+`unirl/rollout/engine/asynchronous.py`: `AsyncBatchRolloutEngine` (AR/diffusion — non-blocking
+batched generations, launch-time version stamps) 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,
+and tail carry/drop.
+
 **Extending it:** a new domain is a new `<Domain>Trainer(BaseTrainer)` that builds its
 remotes inside a `placement(...)` scope and implements `train_step` + `train`; the
 matching `../train_<domain>.py` entrypoint composes the recipe and calls it.
diff --git a/unirl/trainer/agentic_async.py b/unirl/trainer/agentic_async.py
index 103a4b3f6..4e78ae1a1 100644
--- a/unirl/trainer/agentic_async.py
+++ b/unirl/trainer/agentic_async.py
@@ -9,13 +9,14 @@
 
 Mechanism vs policy (LIN-531): the **engine** exposes a ``submit`` / ``poll`` /
 ``finalize_if_drained`` / ``abort`` interface over a background drain (see
-:class:`~unirl.rollout.engine.agentic.engine.AgenticRolloutEngine`); this **trainer**
-owns the *policy* —
+:class:`~unirl.rollout.engine.agentic.engine.AgenticRolloutEngine`), consumed
+through the driver-side :class:`~unirl.rollout.engine.asynchronous.AsyncAgenticRolloutEngine`
+(group assembly + versioned buffering); this **trainer** owns the *policy* —
 
 * **Producer** — keep the rollout slab saturated: ``submit`` a pool of fresh prompt
-  siblings + resumed carried partials, ``poll`` completed trajectories, bucket them by
-  root id into complete GRPO groups (:class:`_GroupAssembler`), and push each into a
-  staleness-bounded :class:`_GroupBuffer`.
+  siblings + resumed carried partials and ``poll`` completed trajectories; the facade
+  buckets them by root id into complete GRPO groups and stamps each into its
+  staleness-bounded versioned buffer.
 * **Consumer** — drain the freshest ``batch_size`` complete groups (within
   ``buffer_max_staleness``), reward + GRPO advantage + one optimizer step (reusing
   :class:`AgenticTrainer`'s helpers), then **quiesce + sync**: ``abort`` the in-flight
@@ -43,13 +44,14 @@
 import logging
 import sys
 import time
-from typing import Dict, Iterable, List, Literal, Optional, Set, Tuple
+from typing import Dict, List, Literal, Optional, Tuple
 
 import torch
 from hydra.utils import instantiate
 from omegaconf import DictConfig
 
 from unirl.distributed.group.placement import placement, remote
+from unirl.rollout.engine.asynchronous import AsyncAgenticRolloutEngine, root_of
 from unirl.train.stack import TrainStepResult
 from unirl.trainer.agentic import AgenticTrainer
 from unirl.trainer.base import BaseTrainer, build_sampling_dict
@@ -61,118 +63,7 @@
 
 
 # --------------------------------------------------------------------------- #
-# Producer-side bookkeeping
-# --------------------------------------------------------------------------- #
-
-
-class _GroupAssembler:
-    """Bucket a flat stream of terminal trajectories into complete GRPO groups.
-
-    The agentic engine ``poll`` returns variable-depth trajectory ``Sample``s; a
-    prompt's ``n`` siblings share its slash-free **root id**
-    (``sample.parts[0].sample_ids[0]``). A group is *complete* once all ``n`` of a
-    root's siblings are terminal. Variable-depth trajectories are NOT concatenated —
-    a group is kept as a ``List[Sample]`` (the trainer flattens their gen Parts at
-    train time, like :meth:`AgenticTrainer.train_step`).
-    """
-
-    def __init__(self, n: int) -> None:
-        self._n = int(n)
-        self._by_root: Dict[str, List[Sample]] = {}
-
-    @staticmethod
-    def root_of(traj: Sample) -> str:
-        return traj.parts[0].sample_ids[0]
-
-    def add_completed(self, trajs: List[Sample]) -> None:
-        """Accumulate terminal (poll-completed) trajectories, bucketed by root id."""
-        for t in trajs:
-            self._by_root.setdefault(self.root_of(t), []).append(t)
-
-    def pop_complete_groups(self) -> List[List[Sample]]:
-        """Emit + drop every root that has all ``n`` siblings terminal (each group a
-        ``List[Sample]`` of exactly ``n`` trajectories)."""
-        ready = [root for root, sibs in self._by_root.items() if len(sibs) >= self._n]
-        out: List[List[Sample]] = []
-        for root in ready:
-            sibs = self._by_root.pop(root)
-            out.append(sibs[: self._n])
-        return out
-
-    def pending_roots(self) -> Set[str]:
-        """Roots with some-but-not-all siblings terminal (their done siblings are held
-        here; their unfinished siblings are carried by the trainer for resume)."""
-        return set(self._by_root)
-
-    def discard_roots(self, roots: Iterable[str]) -> int:
-        """Drop incomplete buckets for abandoned roots.
-
-        Returns the number of terminal sibling trajectories that were being held
-        while the other siblings were still in flight.
-        """
-        discarded = 0
-        for root in set(roots):
-            discarded += len(self._by_root.pop(root, []))
-        return discarded
-
-    def size(self) -> int:
-        return len(self._by_root)
-
-
-class _GroupBuffer:
-    """Staleness-bounded buffer of complete GRPO groups (the ``AsyncARTrainer._RolloutBuffer``
-    shape, but each item is a ``List[Sample]`` group of variable-depth trajectories).
-
-    A group is stamped with the ``weight_version`` it *completed* under and a monotonic
-    ``gen_id`` for freshness ordering. The per-token ratio corrects the within-trajectory
-    version spread (a carried trajectory's turns); this buffer only bounds how stale a
-    *completed* group may be before the consumer trains it.
-    """
-
-    def __init__(self) -> None:
-        self._items: List[Tuple[List[Sample], int, int]] = []  # (group, weight_version, gen_id)
-        self._evicted: List[List[Sample]] = []
-
-    def put(self, group: List[Sample], *, weight_version: int, gen_id: int) -> None:
-        self._items.append((list(group), int(weight_version), int(gen_id)))
-
-    def size(self) -> int:
-        return len(self._items)
-
-    def drain_freshest(
-        self,
-        n: int,
-        *,
-        current_version: Optional[int] = None,
-        max_staleness: Optional[int] = None,
-    ) -> Optional[List[List[Sample]]]:
-        """Pop the ``n`` freshest complete groups, evicting over-stale ones first.
-
-        Returns ``None`` if fewer than ``n`` groups remain after eviction (the consumer
-        then waits for the producer to fill more).
-        """
-        if max_staleness is not None and current_version is not None:
-            kept: List[Tuple[List[Sample], int, int]] = []
-            for item in self._items:
-                if current_version - item[1] <= max_staleness:
-                    kept.append(item)
-                else:
-                    self._evicted.append(item[0])
-            self._items = kept
-        if len(self._items) < n:
-            return None
-        self._items.sort(key=lambda it: it[2], reverse=True)  # freshest gen_id first
-        picked, self._items = self._items[:n], self._items[n:]
-        return [grp for grp, _, _ in picked]
-
-    def pop_evicted_groups(self) -> List[List[Sample]]:
-        """Return and clear groups rejected by the latest staleness checks."""
-        evicted, self._evicted = self._evicted, []
-        return evicted
-
-
-# --------------------------------------------------------------------------- #
-# The trainer
+# The trainer (producer-side bookkeeping lives in AsyncAgenticRolloutEngine)
 # --------------------------------------------------------------------------- #
 
 
@@ -230,14 +121,13 @@ def __init__(
         self._tail_policy = str(tail_policy)
         if self._tail_policy not in ("carry", "drop"):
             raise ValueError(f"tail_policy must be 'carry' or 'drop'; got {self._tail_policy!r}")
-        self._weight_version = 0
         # Monotonic per-DRIVE nonce. ``rollout_id`` alone does not make root ids
         # unique: :meth:`_next_batch` refills re-submit under the SAME rollout_id, and a
         # data source may restart its ids on every ``get_samples`` (DefaultDataSource
         # numbers by batch position, not prompt identity). Two drives would then
-        # namespace different source rows identically, and :class:`_GroupAssembler`
-        # — which buckets purely by root id — would merge siblings of unrelated
-        # prompts into one GRPO group and overwrite their ``_gt_by_root`` answers.
+        # namespace different source rows identically, and the facade's group
+        # assembler — which buckets purely by root id — would merge siblings of
+        # unrelated prompts into one GRPO group and overwrite their ``_gt_by_root`` answers.
         self._drive_seq = 0
         self._carried_tail_trajectories = 0
         self._dropped_tail_trajectories = 0
@@ -336,21 +226,8 @@ def _build_tasks(self, carried: List[Sample], rollout_id: int) -> List[Sample]:
 
     def _submit_drive(self, carried: List[Sample], rollout_id: int) -> None:
         """Submit a fresh over-sampled drive (non-blocking). Call only when the prior
-        drive is finalized/``abort``ed (else two drains would double-pull)."""
-        self.rollout.submit(self._build_tasks(carried, rollout_id))
-
-    def _ingest_completed(self, completed: List[Sample]) -> int:
-        """Ingest terminal trajectories and promote newly complete groups."""
-        if completed:
-            self._assembler.add_completed(completed)
-            for group in self._assembler.pop_complete_groups():
-                self._buffer.put(group, weight_version=self._weight_version, gen_id=self._gen_id)
-                self._gen_id += 1
-        return len(completed)
-
-    def _pump(self) -> int:
-        """Poll completed trajectories into the assembler and buffer."""
-        return self._ingest_completed(self.rollout.poll()[0])
+        drive is finalized/quiesced (else two drains would double-pull)."""
+        self._engine.submit(self._build_tasks(carried, rollout_id))
 
     def _apply_tail_policy(self, carried: List[Sample], rollout_id: int) -> List[Sample]:
         """Keep resumable tails or purge all state belonging to dropped roots."""
@@ -359,8 +236,8 @@ def _apply_tail_policy(self, carried: List[Sample], rollout_id: int) -> List[Sam
             logger.info("rollout %d async: carry tail=%d trajectories", rollout_id, len(carried))
             return carried
 
-        roots = {self._assembler.root_of(sample) for sample in carried}
-        discarded_completed = self._assembler.discard_roots(roots)
+        roots = {root_of(sample) for sample in carried}
+        discarded_completed = self._engine.discard_roots(roots)
         for root in roots:
             self._gt_by_root.pop(root, None)
         self._dropped_tail_trajectories += len(carried)
@@ -384,42 +261,35 @@ def _log_tail_metrics(self, rollout_step: int) -> None:
                 "async/dropped_tail_trajectories": self._dropped_tail_trajectories,
                 "async/dropped_tail_roots": self._dropped_tail_roots,
                 "async/discarded_completed_trajectories": self._discarded_completed_trajectories,
-                "async/assembler_pending_roots": self._assembler.size(),
+                "async/assembler_pending_roots": self._engine.pending_groups(),
             },
         )
 
     def _drain_buffer(self, n: int, *, max_staleness: int) -> Optional[List[List[Sample]]]:
         """Drain fresh groups and forget ground truth for stale evictions."""
-        picked = self._buffer.drain_freshest(
-            n,
-            current_version=self._weight_version,
-            max_staleness=max_staleness,
-        )
-        for group in self._buffer.pop_evicted_groups():
+        picked = self._engine.drain_freshest(n, max_staleness=max_staleness)
+        for group in self._engine.pop_evicted():
             if group:
-                self._gt_by_root.pop(self._assembler.root_of(group[0]), None)
+                self._gt_by_root.pop(root_of(group[0]), None)
         return picked
 
     def _next_batch(self, rollout_id: int) -> List[List[Sample]]:
         """Pump the producer until the buffer holds ``batch_size`` complete groups within
         the staleness bound, then drain the freshest ones. If the in-flight drive drains
         without filling the buffer (staleness eviction / failures / small over-sample),
-        refill with a fresh drive (resuming any carried partials)."""
+        refill with a fresh drive."""
         stale = self._buffer_max_staleness if self._buffer_max_staleness is not None else 0
         refills = 0
         while True:
-            self._pump()
+            self._engine.poll()
             picked = self._drain_buffer(self.batch_size, max_staleness=stale)
             if picked is not None:
                 return picked
-            completed = self.rollout.finalize_if_drained()[0]
-            if completed is None:
+            # finalize_if_drained joins the drain and ingests its last completions
+            # atomically, before a new submit is allowed to reset worker buffers.
+            if self._engine.finalize_if_drained() is None:
                 time.sleep(self._POLL_INTERVAL_S)  # in-flight drive still generating; back off
                 continue
-
-            # Atomic engine finalization joins the drain and returns its last
-            # completions before a new submit is allowed to reset worker buffers.
-            self._ingest_completed(completed)
             picked = self._drain_buffer(self.batch_size, max_staleness=stale)
             if picked is not None:
                 return picked
@@ -428,11 +298,10 @@ def _next_batch(self, rollout_id: int) -> List[List[Sample]]:
             if refills > self._MAX_REFILLS:
                 raise RuntimeError(
                     f"async-agentic rollout {rollout_id}: buffer underflow after {refills} refills "
-                    f"(buffer={self._buffer.size()} < batch={self.batch_size}); raise "
+                    f"(buffer={self._engine.buffered_groups()} < batch={self.batch_size}); raise "
                     f"oversample_batch_size or buffer_max_staleness."
                 )
-            self._submit_drive(self._pending_carried, rollout_id)
-            self._pending_carried = []
+            self._submit_drive([], rollout_id)  # fresh refill (carried tails resubmit at sync time)
 
     # ------------------------------------------------------------------
     # Consumer — reward + GRPO advantage + one optimizer step over a group batch
@@ -485,10 +354,10 @@ def _train_on_groups(
             extra_metrics={
                 "agent/mean_turns": (sum(depths) / len(depths)) if depths else 0.0,
                 "agent/max_turns": max(depths) if depths else 0,
-                "async/buffer_groups": self._buffer.size(),
-                "async/weight_version": self._weight_version,
+                "async/buffer_groups": self._engine.buffered_groups(),
+                "async/weight_version": self._engine.weight_version,
                 "async/version_span": (max(versions) - min(versions)) if versions else 0,
-                "async/assembler_pending_roots": self._assembler.size(),
+                "async/assembler_pending_roots": self._engine.pending_groups(),
                 "async/carried_tail_trajectories": self._carried_tail_trajectories,
                 "async/dropped_tail_trajectories": self._dropped_tail_trajectories,
                 "async/dropped_tail_roots": self._dropped_tail_roots,
@@ -532,10 +401,7 @@ def train(
             },
         )
 
-        self._buffer = _GroupBuffer()
-        self._assembler = _GroupAssembler(self._n)
-        self._pending_carried: List[Sample] = []
-        self._gen_id = start_rollout
+        self._engine = AsyncAgenticRolloutEngine(self.rollout, group_size=self._n, start_gen_id=start_rollout)
 
         if start_rollout < num_rollouts and start_rollout and self.weight_sync is not None:
             self.weight_sync.sync()  # push restored weights into the fresh engine
@@ -558,8 +424,7 @@ def train(
                 if need_save or need_sync:
                     # ONE turn-boundary quiesce for both: checkpoint the in-flight tail so
                     # the engine is decode-idle (safe to sync / save), then resume it.
-                    checkpointed = self.rollout.abort()[0]
-                    self._pump()  # grab trajectories that completed DURING the quiesce (before submit resets)
+                    checkpointed = self._engine.quiesce()
                     carried = self._apply_tail_policy(checkpointed, rollout_id)
                     if checkpointed:
                         self._log_tail_metrics(step)
@@ -573,14 +438,13 @@ def train(
                         )
                     if need_sync:
                         self.weight_sync.sync()
-                        self._weight_version += 1
+                        self._engine.bump_weight_version()
                     if step < num_rollouts:
                         self._submit_drive(carried=carried, rollout_id=step)  # resume safe tails + fresh
         finally:
             active_error = sys.exc_info()[0] is not None
             try:
-                checkpointed = self.rollout.abort()[0]  # stop the resident drive; leak no drives
-                self._pump()
+                checkpointed = self._engine.quiesce()  # stop the resident drive; leak no drives
                 self._apply_tail_policy(checkpointed, num_rollouts)
                 if checkpointed:
                     self._log_tail_metrics(num_rollouts)
diff --git a/unirl/trainer/agentic_partial.py b/unirl/trainer/agentic_partial.py
index 8a456281b..b0e1d7a52 100644
--- a/unirl/trainer/agentic_partial.py
+++ b/unirl/trainer/agentic_partial.py
@@ -16,8 +16,9 @@
 Motivation (LIN-531 ALFWorld comparison): the fully-async trainer *lost* on ALFWorld because
 disaggregation halves the generation GPUs; colocate+partial keeps all GPUs for generation while
 still cutting the straggler tail — the slime "over-sample + abort + recycle" / verl `bypass_mode`
-pattern. The engine interface (`submit`/`poll`/`finalize_if_drained`/`abort`) and `_GroupAssembler`/
-`_GroupBuffer` are reused verbatim; only the colocate wake/sync/sleep choreography is new.
+pattern. The driver-side :class:`~unirl.rollout.engine.asynchronous.AsyncAgenticRolloutEngine`
+(submit/poll/finalize_if_drained/quiesce + group assembly and versioned buffering) is shared with
+``AsyncAgenticTrainer``; only the colocate wake/sync/sleep choreography is new.
 
 Correctness: `TensorWeightSync.sync` writes the live SRT weight pool, so it must run **awake +
 decode-idle** — sync sits at the top (post-`wake_up`, pre-`submit`, barrier parity) and `abort`
@@ -34,8 +35,8 @@
 from collections import Counter
 from typing import Dict, List, Literal, Optional
 
+from unirl.rollout.engine.asynchronous import AsyncAgenticRolloutEngine, root_of
 from unirl.trainer.agentic import AgenticTrainer
-from unirl.trainer.agentic_async import _GroupAssembler, _GroupBuffer
 from unirl.trainer.agentic_env import _EnvRewardSource
 from unirl.types.sample import Part, Sample
 
@@ -75,7 +76,6 @@ def __init__(
         self._tail_policy = str(tail_policy)
         if self._tail_policy not in ("carry", "drop"):
             raise ValueError(f"tail_policy must be 'carry' or 'drop'; got {self._tail_policy!r}")
-        self._weight_version = 0
         # Refills happen under the same optimizer ``rollout_id``. Keep a
         # per-drive nonce so data sources whose ids restart on every draw cannot
         # mix unrelated siblings in the root-keyed group assembler.
@@ -105,19 +105,6 @@ def _build_tasks(self, carried: List[Sample], rollout_id: int) -> List[Sample]:
         tasks.extend(carried)
         return tasks
 
-    def _ingest_completed(self, completed: List[Sample]) -> int:
-        """Ingest terminal trajectories and promote newly complete groups."""
-        if completed:
-            self._assembler.add_completed(completed)
-            for group in self._assembler.pop_complete_groups():
-                self._buffer.put(group, weight_version=self._weight_version, gen_id=self._gen_id)
-                self._gen_id += 1
-        return len(completed)
-
-    def _pump(self) -> int:
-        """Poll completed trajectories → assembler → promote complete groups into the buffer."""
-        return self._ingest_completed(self.rollout.poll()[0])
-
     def _reconstruct_request(self, trajs: List[Sample]) -> Sample:
         """A request whose root Part carries every trajectory's root id + ground-truth answer,
         so the inherited answer-grader (`_rewards_and_groups`) works unchanged. The env-reward
@@ -134,10 +121,10 @@ def _apply_tail_policy(self, carried: List[Sample], rollout_id: int) -> None:
             self._carried = carried
             return
 
-        roots = {self._assembler.root_of(sample) for sample in carried}
+        roots = {root_of(sample) for sample in carried}
         self._last_dropped_trajectories = len(carried)
         self._last_dropped_roots = len(roots)
-        self._last_discarded_completed_trajectories = self._assembler.discard_roots(roots)
+        self._last_discarded_completed_trajectories = self._engine.discard_roots(roots)
         for root in roots:
             self._gt_by_root.pop(root, None)
         self._carried = []
@@ -151,14 +138,10 @@ def _apply_tail_policy(self, carried: List[Sample], rollout_id: int) -> None:
 
     def _drain_buffer(self, n: int, *, max_staleness: int) -> Optional[List[List[Sample]]]:
         """Drain fresh groups and forget ground truth for stale evictions."""
-        picked = self._buffer.drain_freshest(
-            n,
-            current_version=self._weight_version,
-            max_staleness=max_staleness,
-        )
-        for group in self._buffer.pop_evicted_groups():
+        picked = self._engine.drain_freshest(n, max_staleness=max_staleness)
+        for group in self._engine.pop_evicted():
             if group:
-                self._gt_by_root.pop(self._assembler.root_of(group[0]), None)
+                self._gt_by_root.pop(root_of(group[0]), None)
         return picked
 
     def _collect_until(self, batch_size: int, rollout_id: int, stale: int) -> List[List[Sample]]:
@@ -167,18 +150,15 @@ def _collect_until(self, batch_size: int, rollout_id: int, stale: int) -> List[L
         buffer (small over-sample / failures / eviction), refill with a fresh drive."""
         refills = 0
         while True:
-            self._pump()
+            self._engine.poll()
             picked = self._drain_buffer(batch_size, max_staleness=stale)
             if picked is not None:
                 return picked
-            completed = self.rollout.finalize_if_drained()[0]
-            if completed is None:
+            # finalize_if_drained joins + ingests the drive's final completions
+            # atomically, before submit() can reset the worker buffers for a refill.
+            if self._engine.finalize_if_drained() is None:
                 time.sleep(self._POLL_INTERVAL_S)  # in-flight drive still generating; back off
                 continue
-
-            # Join and drain the worker buffers atomically before submit() can
-            # reset them for a refill.
-            self._ingest_completed(completed)
             picked = self._drain_buffer(batch_size, max_staleness=stale)
             if picked is not None:
                 return picked
@@ -187,10 +167,10 @@ def _collect_until(self, batch_size: int, rollout_id: int, stale: int) -> List[L
             if refills > self._MAX_REFILLS:
                 raise RuntimeError(
                     f"colocate-partial rollout {rollout_id}: buffer underflow after {refills} refills "
-                    f"(buffer={self._buffer.size()} < batch={batch_size}); raise oversample_batch_size "
+                    f"(buffer={self._engine.buffered_groups()} < batch={batch_size}); raise oversample_batch_size "
                     f"or buffer_max_staleness."
                 )
-            self.rollout.submit(self._build_tasks([], rollout_id))  # fresh refill (carried already in flight)
+            self._engine.submit(self._build_tasks([], rollout_id))  # fresh refill (carried already in flight)
 
     # ------------------------------------------------------------------
     # One colocate partial drive: wake → sync → submit → collect-N → abort → sleep
@@ -202,13 +182,13 @@ def _drive_partial(self, rollout_id: int, sync_weights: bool, stale: int) -> Lis
         # SRT weight pool, which a full sleep() would release.
         if sync_weights and self.weight_sync is not None:
             self.weight_sync.sync()
-            self._weight_version += 1
+            self._engine.bump_weight_version()
         tasks = self._build_tasks(self._carried, rollout_id)  # fresh + carried
         self._carried = []  # consumed into this drive
-        self.rollout.submit(tasks)
+        self._engine.submit(tasks)
         groups = self._collect_until(self.batch_size, rollout_id, stale)
-        carried = self.rollout.abort()[0]  # checkpoint the in-flight tail (turn boundary) → decode-idle, still awake
-        self._pump()  # grab trajectories that completed DURING the quiesce (before next submit resets buffers)
+        # Turn-boundary checkpoint + final poll for quiesce-time completions → decode-idle, still awake.
+        carried = self._engine.quiesce()
         self.rollout.sleep()  # engine quiesced → safe to offload → frees GPU for the train step
         # Diagnostic (LIN-531): the tail we cut — its turn depths show whether the commit-N actually
         # skipped stragglers (wide tail depth) or just wasted uniform-depth over-sample.
@@ -259,10 +239,8 @@ def train(
             },
         )
 
-        self._buffer = _GroupBuffer()
-        self._assembler = _GroupAssembler(self._n)
+        self._engine = AsyncAgenticRolloutEngine(self.rollout, group_size=self._n, start_gen_id=start_rollout)
         self._carried = []
-        self._gen_id = start_rollout
 
         try:
             if self.eval_interval > 0:
@@ -277,7 +255,7 @@ def train(
                 groups = self._drive_partial(rollout_id, sync_weights, stale)
                 trajs: List[Sample] = [t for group in groups for t in group]
                 rewards, group_ids = self._rewards_and_groups(self._reconstruct_request(trajs), trajs, rollout_id)
-                for root in {self._assembler.root_of(traj) for traj in trajs}:
+                for root in {root_of(traj) for traj in trajs}:
                     self._gt_by_root.pop(root, None)
                 result, mean_reward = self._advantage_train_and_log(
                     trajs,
@@ -292,9 +270,9 @@ def train(
                         "partial/dropped_trajectories": self._last_dropped_trajectories,
                         "partial/dropped_roots": self._last_dropped_roots,
                         "partial/discarded_completed_trajectories": self._last_discarded_completed_trajectories,
-                        "partial/assembler_pending_roots": self._assembler.size(),
-                        "partial/buffer_groups": self._buffer.size(),
-                        "partial/weight_version": self._weight_version,
+                        "partial/assembler_pending_roots": self._engine.pending_groups(),
+                        "partial/buffer_groups": self._engine.buffered_groups(),
+                        "partial/weight_version": self._engine.weight_version,
                     },
                 )
                 self.wandb_logger.log_progress(rollout_id, num_rollouts, result, mean_reward, logger=logger)
@@ -307,8 +285,7 @@ def train(
         finally:
             active_error = sys.exc_info()[0] is not None
             try:
-                carried = self.rollout.abort()[0]  # stop any drive left running
-                self._pump()
+                carried = self._engine.quiesce()  # stop any drive left running
                 self._apply_tail_policy(carried, num_rollouts)
             except BaseException:  # noqa: BLE001 — preserve an active training failure
                 if active_error:
diff --git a/unirl/trainer/async_ar.py b/unirl/trainer/async_ar.py
index cfedb09c6..f3f828f44 100644
--- a/unirl/trainer/async_ar.py
+++ b/unirl/trainer/async_ar.py
@@ -19,11 +19,12 @@
   regime). ``>0`` = **off-policy continuous buffer**: generations may run ahead
   across syncs, bounded by eviction; the rollout-anchored DRPO ratio absorbs it.
 
-Generation is launched as **non-blocking Ray futures** by
-``RayGenerationDispatcher`` and reaped by ``AsyncRolloutScheduler`` on the
-single driver thread — no producer thread, no locks. 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
+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
+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.
 
 Subclasses ``ARTrainer`` to reuse ``_build_request_sample``/``evaluate`` and ``BaseTrainer``
@@ -43,12 +44,7 @@
 
 from unirl.distributed.group.placement import placement, remote
 from unirl.distributed.tensor import hydrate
-from unirl.rollout.async_runtime import (
-    AsyncRolloutScheduler,
-    BufferedRolloutGroup,
-    InflightGeneration,
-    RayGenerationDispatcher,
-)
+from unirl.rollout.engine.asynchronous import AsyncBatchRolloutEngine
 from unirl.train.stack import TrainStepResult
 from unirl.trainer.ar import ARTrainer
 from unirl.trainer.base import BaseTrainer, build_sampling_dict
@@ -115,7 +111,6 @@ def __init__(
         self._train_fraction = float(train_fraction)
         self._max_inflight = max(1, int(max_inflight))
         self._buffer_max_staleness = buffer_max_staleness
-        self._weight_version = 0  # driver-tracked policy version (# of weight syncs issued)
         # DP size of the TRAIN slab — the divisor for balance_shards (the parent
         # uses self.num_devices because colocate training spans the whole pool;
         # here training only spans the train slab).
@@ -205,20 +200,16 @@ def _build_async_sample(self, gen_id: int) -> Sample:
         """Consume one data batch and build the request Sample for ``gen_id``."""
         return self._build_request_sample(self.data_source.get_samples(self.batch_size), gen_id)
 
-    def _score_completed(
-        self,
-        job: InflightGeneration,
-        completed: Sample,
-    ) -> List[Sample]:
+    def _score_completed(self, gen_id: int, completed: Sample) -> List[Sample]:
         """Score a completed Sample and split it into tree-complete groups.
 
-        Scoring must precede ``_drop_decoded`` (the reward reads the decoded
-        primitive). Keyed by ``gen_id`` so media panels behave like the old path.
-        The filled ``Sample`` is self-contained (it carries its input Parts), so
-        no request handle is kept on the in-flight record.
+        Runs at reap time inside the engine. Scoring must precede
+        ``_drop_decoded`` (the reward reads the decoded primitive). Keyed by
+        ``gen_id`` so media panels behave like the old path. The filled
+        ``Sample`` is self-contained (it carries its input Parts).
         """
         scored = self.reward.score_and_attach(completed)
-        self._drop_decoded(scored, rollout_id=job.gen_id)
+        self._drop_decoded(scored, rollout_id=gen_id)
         return scored.split()
 
     def _drain_all(self) -> None:
@@ -228,7 +219,7 @@ def _drain_all(self) -> None:
         when weights + KV cache update mid-flight), before eval/checkpoint (shared
         engine), and in ``finally`` (no leaked ObjectRefs).
         """
-        self._async_scheduler.drain_all(self._score_completed)
+        self._async_engine.quiesce()
 
     # ------------------------------------------------------------------
     # Train tail (mirrors ar.py:152-182, minus wake/sleep) — reward parity
@@ -305,11 +296,12 @@ def train(
             },
         )
 
-        self._async_scheduler = AsyncRolloutScheduler(
-            RayGenerationDispatcher(self.rollout),
-            groups_per_step=self.batch_size,
+        # gen_id is seeded by start_rollout so launches stay 1:1 with rollout_id.
+        self._async_engine = AsyncBatchRolloutEngine(
+            self.rollout,
+            complete=self._score_completed,
+            start_gen_id=start_rollout,
         )
-        self._async_scheduler.reset(start_rollout)
 
         if resumed and self.weight_sync is not None:
             self.weight_sync.sync()  # push restored weights into the fresh engine
@@ -322,7 +314,7 @@ def train(
                 picked = self._next_step(rollout_id, interval, M, stale, num_rollouts)
                 # Reassemble the drained per-prompt group Samples into one batched
                 # Sample [input(P), gen(P*N)] — the inverse of Sample.split.
-                sample = Sample.concat([item.sample for item in picked])
+                sample = Sample.concat(picked)
                 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
@@ -341,7 +333,7 @@ def train(
                 if step % interval == 0 and self.weight_sync is not None:
                     self._drain_all()  # MANDATORY: weight/KV update corrupts in-flight generations
                     self.weight_sync.sync()
-                    self._weight_version += 1
+                    self._async_engine.bump_weight_version()
         finally:
             # Match BaseTrainer._finish_wandb: cleanup failures must not mask
             # the exception that caused teardown.
@@ -362,23 +354,27 @@ def _next_step(
         M: int,
         stale: int,
         num_rollouts: int,
-    ) -> List[BufferedRolloutGroup]:
+    ) -> List[Sample]:
         """Top up launches, reap completed generations, and return the freshest
-        ``groups_per_step`` (``batch_size``) groups for ``rollout_id`` (blocking
-        on the oldest in-flight generation if the buffer is short).
+        ``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).
         """
-        return self._async_scheduler.next_step(
-            rollout_id=rollout_id,
-            sync_interval=interval,
-            max_inflight=M,
-            max_staleness=stale,
-            num_rollouts=num_rollouts,
-            current_version=self._weight_version,
-            build_sample=self._build_async_sample,
-            on_complete=self._score_completed,
-        )
+        engine = self._async_engine
+        while True:
+            ceiling = min(num_rollouts, ((rollout_id // interval) + 1 + stale) * interval)
+            while engine.next_gen_id < ceiling and engine.inflight < M:
+                engine.submit(self._build_async_sample(engine.next_gen_id))
+            engine.poll()
+            picked = engine.drain_freshest(self.batch_size, max_staleness=stale)
+            engine.pop_evicted()  # over-stale groups are discarded on the batch path
+            if picked is not None:
+                return picked
+            if engine.inflight:
+                engine.wait_oldest()
+            else:
+                raise RuntimeError("async rollout buffer underflow with no in-flight generations")
diff --git a/unirl/trainer/async_diffusion.py b/unirl/trainer/async_diffusion.py
index 06e442598..83ac178fe 100644
--- a/unirl/trainer/async_diffusion.py
+++ b/unirl/trainer/async_diffusion.py
@@ -8,10 +8,11 @@
 plumbing (``_build_request_sample`` / ``_drop_decoded`` / ``evaluate`` /
 checkpoint / FlowGRPO ``stack.train_track``).
 
-The async loop itself is the shared
-:class:`~unirl.rollout.async_runtime.AsyncRolloutScheduler` that ``AsyncARTrainer``
-drives — one single-threaded driver loop over non-blocking Ray dispatch, no
-producer thread and no locks. This trainer supplies only the diffusion hooks:
+The async loop runs over the shared
+:class:`~unirl.rollout.engine.asynchronous.AsyncBatchRolloutEngine` (the same engine
+``AsyncARTrainer`` drives) — one single-threaded driver loop over non-blocking
+Ray dispatch, no producer thread and no locks. This trainer supplies only the
+diffusion hooks:
 
 * ``_build_async_sample`` — one data batch → one request ``Sample``.
 * ``_score_completed`` — reward at reap time, then split into tree-complete
@@ -26,13 +27,13 @@
     cross. ``0`` (default) never crosses a sync; ``>0`` enables a bounded
     policy-lag buffer.
 
-The scheduler runs in ``reap_before_launch`` mode, 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 — measured ~150s/rollout on
-BAGEL instead of ~8s. Reaping first hands the send idle workers, and the launch
-that follows still happens before the step returns, so the next generation
-overlaps this step's training.
+``_next_step`` 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 —
+measured ~150s/rollout on BAGEL instead of ~8s. Reaping first hands the send
+idle workers, and the launch that follows still happens before the step returns,
+so the next generation overlaps this step's training.
 
 Draining all in-flight generations before each weight sync is MANDATORY (a
 weight + KV update corrupts an in-flight generation); that is the
@@ -49,12 +50,7 @@
 import torch
 
 from unirl.distributed.tensor import hydrate
-from unirl.rollout.async_runtime import (
-    AsyncRolloutScheduler,
-    BufferedRolloutGroup,
-    InflightGeneration,
-    RayGenerationDispatcher,
-)
+from unirl.rollout.engine.asynchronous import AsyncBatchRolloutEngine
 from unirl.train.stack import TrainStepResult
 from unirl.trainer.diffusion import DiffusionTrainer
 from unirl.types.sample import Sample
@@ -93,7 +89,6 @@ def __init__(
         # ---- async state ----
         self._max_inflight = max_inflight
         self._buffer_max_staleness = buffer_max_staleness
-        self._weight_version = 0  # driver-tracked policy version (# of weight syncs issued)
 
     # ------------------------------------------------------------------
     # Generic async-runtime hooks
@@ -103,22 +98,17 @@ def _build_async_sample(self, gen_id: int) -> Sample:
         """Consume one data batch and build the request Sample for ``gen_id``."""
         return self._build_request_sample(self.data_source.get_samples(self.batch_size), gen_id)
 
-    def _score_completed(
-        self,
-        job: InflightGeneration,
-        completed: Sample,
-    ) -> List[Sample]:
+    def _score_completed(self, gen_id: int, completed: Sample) -> List[Sample]:
         """Score a completed Sample and split it into tree-complete groups.
 
-        Scoring is synchronous at reap time — before the next launch and before
+        Runs at reap time inside the engine — before the next launch and before
         training consumes the batch — and must precede ``_drop_decoded`` (the
         reward reads the decoded primitive). Keyed by ``gen_id`` so media panels
-        behave like the synchronous path. The filled ``Sample`` is self-contained
-        (it carries its input Parts), so no request handle is kept on the
-        in-flight record.
+        behave like the synchronous path. The filled ``Sample`` is
+        self-contained (it carries its input Parts).
         """
         scored = self.reward.score_and_attach(completed)
-        self._drop_decoded(scored, rollout_id=job.gen_id)
+        self._drop_decoded(scored, rollout_id=gen_id)
         return scored.split()
 
     def _drain_all(self) -> None:
@@ -128,7 +118,7 @@ def _drain_all(self) -> None:
         generate), before eval/checkpoint (shared engine), and in ``finally`` (no
         leaked ObjectRefs).
         """
-        self._async_scheduler.drain_all(self._score_completed)
+        self._async_engine.quiesce()
 
     # ------------------------------------------------------------------
     # Train tail (mirrors DiffusionTrainer.train_step's post-generate half:
@@ -199,15 +189,12 @@ def train(
             },
         )
 
-        # reap_before_launch: 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 (see the module docstring).
-        self._async_scheduler = AsyncRolloutScheduler(
-            RayGenerationDispatcher(self.rollout),
-            groups_per_step=self.batch_size,
-            reap_before_launch=True,
+        # gen_id is seeded by start_rollout so launches stay 1:1 with rollout_id.
+        self._async_engine = AsyncBatchRolloutEngine(
+            self.rollout,
+            complete=self._score_completed,
+            start_gen_id=start_rollout,
         )
-        self._async_scheduler.reset(start_rollout)
 
         if resumed and self.weight_sync is not None:
             self.weight_sync.sync()  # push restored weights into the fresh engine
@@ -222,7 +209,7 @@ def train(
                 picked = self._next_step(rollout_id, interval, M, stale, num_rollouts)
                 # Reassemble the drained per-prompt group Samples into one batched
                 # Sample [input(P), gen(P*N)] — the inverse of Sample.split.
-                sample = Sample.concat([item.sample for item in picked])
+                sample = Sample.concat(picked)
                 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
@@ -241,7 +228,7 @@ def train(
                 if step % interval == 0 and self.weight_sync is not None:
                     self._drain_all()  # MANDATORY: weight/KV update corrupts in-flight generations
                     self.weight_sync.sync()
-                    self._weight_version += 1
+                    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
@@ -261,23 +248,31 @@ def _next_step(
         M: int,
         stale: int,
         num_rollouts: int,
-    ) -> List[BufferedRolloutGroup]:
+    ) -> List[Sample]:
         """Reap completed generations, top up launches, and return the freshest
-        ``batch_size`` groups for ``rollout_id`` (blocking on the oldest in-flight
-        generation if the buffer is short).
+        ``batch_size`` scored group Samples for ``rollout_id`` (blocking on the
+        oldest in-flight generation if the buffer is short).
+
+        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.
         """
-        return self._async_scheduler.next_step(
-            rollout_id=rollout_id,
-            sync_interval=interval,
-            max_inflight=M,
-            max_staleness=stale,
-            num_rollouts=num_rollouts,
-            current_version=self._weight_version,
-            build_sample=self._build_async_sample,
-            on_complete=self._score_completed,
-        )
+        engine = self._async_engine
+        while True:
+            ceiling = min(num_rollouts, ((rollout_id // interval) + 1 + stale) * interval)
+            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()  # over-stale groups are discarded on the batch path
+            if picked is not None:
+                return picked
+            if engine.inflight:
+                engine.wait_oldest()
+            else:
+                raise RuntimeError("async rollout buffer underflow with no in-flight generations")