refactor(rollout): move the driver-side async engines to rollout/manager/

unirl/rollout/README.md

@@ -63,13 +63,13 @@ 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
+- **Driver-side async engines** (`manager/`, the driver-side half next
   to `engine/synchronous.py`'s worker-side sync contracts). Both engines expose the
   same consumer verbs the async trainers program against: `poll` / `drain_freshest` /
-  `pop_evicted` / `quiesce` + engine-owned `weight_version`. `AsyncBatchRolloutEngine`
+  `pop_evicted` / `quiesce` + engine-owned `weight_version`. `BatchManager`
   (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
+  `AgenticManager` (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).
 
@@ -93,10 +93,10 @@ implements its weight-receive method and a matching `sync:` handler in
 - **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
+  `BatchManager.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()`; its `sync_weights()` rejects a live
+  `AgenticManager.quiesce()`; its `sync_weights()` rejects a live
   drive, then pairs the weight push with the version bump and logs the sync.
   Reap-vs-launch ordering is trainer
   statement order (diffusion polls before topping up; see its `_next_step`).

unirl/rollout/engine/__init__.py

@@ -1,12 +1,11 @@
 """Rollout engines over the canonical ``Sample`` request type.
 
-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
-batch/agentic async engines and their mechanisms).
+``synchronous.py`` records the worker-side contracts (``BaseRolloutEngine`` — the
+broad ABC including coordinator engines — and ``SyncRolloutEngine``, the ``Sample``
+→ ``Sample`` refinement the per-backend subpackages implement). The driver side
+lives in ``../manager/``, which owns admission, acceptance and disposal over time
+and holds no model.
 
-Deliberately empty otherwise: importing the driver-side ``asynchronous`` module
-must stay ray/torch-free, so this init imports nothing and consumers import the
-halves directly.
+Deliberately empty otherwise: this init imports nothing, so consumers import the
+module they need directly and the manager package stays ray/torch-free.
 """

unirl/rollout/engine/agentic/engine.py

@@ -12,7 +12,7 @@
 (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 — consumed driver-side through
-:class:`~unirl.rollout.engine.asynchronous.AsyncAgenticRolloutEngine` (rank-0 unwrap + group assembly +
+:class:`~unirl.rollout.manager.AgenticManager` (rank-0 unwrap + group assembly +
 versioned buffering):
 
 - ``submit(request)`` — enqueue a pool (fresh prompts and/or carried partials) and fire the drain