perf(async): eval-priority rollout scheduling on the shared pool

[jeffreysijuntan]

Problem

In fully-async training, periodic validation currently stops the world:
_validate_async_with_pause calls coordinator.pause_generation() then
await coordinator.wait_for_drain() — it waits for all in-flight training
rollouts to finish before any eval rollout starts. With long-horizon SWE tasks,
that drain can leave the rollout pool idle for many minutes every test_freq
steps.

Change

Run eval on the same shared rollout pool and let eval rollouts preempt
training rollouts for concurrency slots — no separate pool, no drain barrier.

• PrioritySemaphore (rllm/utils/priority_semaphore.py): an asyncio
  counting semaphore that hands a freed permit to the highest-priority
  waiter (FIFO within a priority). Exposes available / waiting for the
  existing inflight / pending diagnostics, plus a slot(priority) async
  context manager. Handles the cancel-after-grant case (returns the permit to
  the pool before re-granting) the same way asyncio.Semaphore does.
• AgentFlowEngine uses it in place of asyncio.Semaphore.
  process_task_with_retry acquires at EVAL_PRIORITY when is_validation
  else TRAIN_PRIORITY (the flag already threads down to this call).
• unified_trainer: the fully-async validation path calls _validate_async
  directly; the pause_generation + wait_for_drain + resume_generation
  wrapper is removed.

Behavior

• Eval rollouts win contention: as in-flight training rollouts release slots,
  freed slots go to eval first, so eval saturates the pool as fast as
  training naturally drains (no waiting for a full drain first).
• Once every eval task holds a slot, remaining slots go to training — training
  fills whatever eval leaves free, exactly as requested.
• eval tasks ≥ pool → eval takes the whole pool, training only resumes in the
  eval tail. eval tasks < pool → training keeps the leftover capacity busy.

Consistency

Eval still runs on frozen weights: await self._validate_async(...) blocks the
training loop, and weight sync (_perform_weight_sync) only runs from that same
loop, so no sync happens for the eval duration. No explicit weight pinning
needed. The generation loop keeps producing training groups during eval; the
coordinator's staleness/concurrency caps bound the pile-up (blocked training
tasks still count toward max_concurrent_rollouts).

Scope

• Gateway / agent-flow fully-async path only (where AgentFlowEngine._semaphore
  governs rollout concurrency). Sync training paths already call
  _validate_async directly and are unaffected.
• Coordinator pause_generation / wait_for_drain / resume_generation remain
  in use by weight sync and the generation loop's end-of-run drain.

Tests

tests/engine/test_priority_semaphore.py — priority ordering, FIFO within a
priority, context-manager release on exception, and cancel-after-grant permit
hand-off. Existing tests/engine/test_agentflow_engine.py still passes (7/7).

🤖 Generated with Claude Code